mapreduce框架运行发生race

1. 运行环境

go版本：go1.21.4 linux/amd64
系统版本：Ubuntu 20.04.6 LTS
系统架构：x86_64

2. 问题描述？

我遇到的问题是我运行我自己实现的mapreduce框架（mit6.824的lab 1）来进行单词计数时发生race，我加了很多的锁，但是依然发生读写操作的竞争条件.
以下是报错内容：

oot@ubuntu:/home/sdr/go_project/6.824/src/main# ./mrcoordinator pg-being_ernest.txt pg-huckleberry_finn.txt pg-dorian_gray.txt pg-metamorphosis.txt pg-frankenstein.txt pg-sherlock_holmes.txt pg-grimm.txt pg-tom_sawyer.txt
==================
WARNING: DATA RACE
Write at 0x00c000027178 by goroutine 131:
  6.824/mr.(*Coordinator).DistrbuteTask()
      /home/sdr/go_project/6.824/src/mr/coordinator.go:234 +0x40e
  runtime.call32()
      /opt/go/src/runtime/asm_amd64.s:748 +0x42
  reflect.Value.Call()
      /opt/go/src/reflect/value.go:380 +0xb5
  net/rpc.(*service).call()
      /opt/go/src/net/rpc/server.go:382 +0x26c
  net/rpc.(*Server).ServeCodec.func2()
      /opt/go/src/net/rpc/server.go:479 +0x15e

Previous read at 0x00c000027178 by goroutine 7:
  6.824/mr.(*Coordinator).loopRemoveTimeOutTask()
      /home/sdr/go_project/6.824/src/mr/coordinator.go:60 +0x44
  6.824/mr.MakeCoordinator.func1()
      /home/sdr/go_project/6.824/src/mr/coordinator.go:104 +0x33

Goroutine 131 (running) created at:
  net/rpc.(*Server).ServeCodec()
      /opt/go/src/net/rpc/server.go:479 +0x64c
  net/rpc.(*Server).ServeConn()
      /opt/go/src/net/rpc/server.go:454 +0x624
  net/rpc.(*Server).ServeHTTP()
      /opt/go/src/net/rpc/server.go:709 +0x564
  net/http.(*ServeMux).ServeHTTP()
      /opt/go/src/net/http/server.go:2514 +0xbc
  net/http.serverHandler.ServeHTTP()
      /opt/go/src/net/http/server.go:2938 +0x2a1
  net/http.(*conn).serve()
      /opt/go/src/net/http/server.go:2009 +0xc24
  net/http.(*Server).Serve.func3()
      /opt/go/src/net/http/server.go:3086 +0x4f

Goroutine 7 (running) created at:
  6.824/mr.MakeCoordinator()
      /home/sdr/go_project/6.824/src/mr/coordinator.go:104 +0x1a6
  main.main()
      /home/sdr/go_project/6.824/src/main/mrcoordinator.go:26 +0xba
==================
FINISH.
Found 1 data race(s)

然后以下是上述报错中提及的相关方法或者结构体的代码实现

type Coordinator struct {
    Id                int //分发给task和work的id，通过该id认证任务是否接收
    Phase             int //记录任务处于map阶段还是reduce阶段
    MapTaskChannel    chan *Task
    ReduceTaskChannel chan *Task
    MapTaskNum        int
    ReduceTaskNum     int
    Lock              sync.Mutex
    IsDone            bool //用于判断是否结束程序
}


type Task struct {
    Taskid   int
    TaskType int
    Nreduce  int
    Files    []string
    Begin    time.Time
    Status   int
}


// start a thread that listens for RPCs from worker.go
func (c *Coordinator) server() {
    rpc.Register(c)
    rpc.HandleHTTP()
    //l, e := net.Listen("tcp", "localhost:1234")
    sockname := coordinatorSock()
    os.Remove(sockname)
    l, e := net.Listen("unix", sockname)
    if e != nil {
        log.Fatal("listen error:", e)
    }
    go http.Serve(l, nil)

}

// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
func (c *Coordinator) Done() bool {
    c.Lock.Lock()
    ret := c.IsDone
    c.Lock.Unlock()
    return ret
}

// 循环判断task是否超时
func (c *Coordinator) loopRemoveTimeOutTask() {
    for !c.Done() {
        c.Lock.Lock()
        switch c.Phase {
        case MapPhase:
            for i := 0; i < c.MapTaskNum; i++ {
                task := <-c.MapTaskChannel
                c.MapTaskChannel <- task
                task.checkTimeOutTask()
            }
        case ReducePhase:
            for i := 0; i < c.ReduceTaskNum; i++ {
                task := <-c.ReduceTaskChannel
                c.ReduceTaskChannel <- task
                task.checkTimeOutTask()
            }
        }
        c.Lock.Unlock()
        time.Sleep(3 * time.Second)
    }

}

func (task *Task) checkTimeOutTask() {
    if task.Status == Running && time.Since(task.Begin) > 10*time.Second {
        task.Status = Ready
    }
}

// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
func MakeCoordinator(files []string, nReduce int) *Coordinator {
    c := Coordinator{
        Id:                0, //Id初始化为零，持续递增生成不重复的任务和工作id
        Phase:             MapPhase,
        MapTaskNum:        len(files),
        ReduceTaskNum:     nReduce,
        MapTaskChannel:    make(chan *Task, len(files)),
        ReduceTaskChannel: make(chan *Task, nReduce),
        IsDone:            false,
        Lock:              sync.Mutex{},
    }
    c.generateMapTask(files)

    go c.loopRemoveTimeOutTask()
    c.server()
    return &c
}

// 生成map task，并放入map task chan queue
func (c *Coordinator) generateMapTask(files []string) {
    taskarray := make([]Task, len(files))
    c.Lock.Lock()
    for i, file := range files {
        taskarray[i] = Task{
            Taskid:   -1,
            TaskType: MapTask,
            Files:    []string{file},
            Status:   Ready,
        }
        c.MapTaskChannel <- &taskarray[i]
    }
    c.Lock.Unlock()
}

// 生成reduce task，并放入reduce task chan queue
func (c *Coordinator) generateReducTask() {
    taskarray := make([]Task, c.ReduceTaskNum)
    c.Lock.Lock()
    for i := 0; i < c.ReduceTaskNum; i++ {
        taskarray[i] = Task{
            Taskid:   -1,
            TaskType: ReduceTask,
            Files:    []string{fmt.Sprintf("mr-tmp-*-%v", i)}, //使用通配符获取所有为该reduce id的map-out-file
            Status:   Ready,
        }
        c.ReduceTaskChannel <- &taskarray[i]
    }
    c.Lock.Unlock()
}

// 检测map task是否全部完成
func (c *Coordinator) maptaskallfinish() bool {
    c.Lock.Lock()
    for i := 0; i < c.MapTaskNum; i++ {
        task := <-c.MapTaskChannel
        c.MapTaskChannel <- task
        if task.Status == Ready || task.Status == Running {
            c.Lock.Unlock()
            return false
        }
    }
    c.Lock.Unlock()
    return true
}

// 检测reduce task是否全部完成
func (c *Coordinator) reducetaskallfinish() bool {
    c.Lock.Lock()
    for i := 0; i < c.ReduceTaskNum; i++ {
        task := <-c.ReduceTaskChannel
        c.ReduceTaskChannel <- task
        if task.Status == Ready || task.Status == Running {
            c.Lock.Unlock()
            return false
        }
    }
    c.Lock.Unlock()
    return true
}

func (c *Coordinator) getphase() int {
    c.Lock.Lock()
    phase := c.Phase
    c.Lock.Unlock()
    return phase
}

// worker 通过rpc调用该函数获取任务
func (c *Coordinator) DistrbuteTask(requesttaskargs *RequestTaskArgs, requesttaskreply *RequestTaskReply) error {
    if c.getphase() == MapPhase {
        c.Lock.Lock()
        for i := 0; i < c.MapTaskNum; i++ {
            task := <-c.MapTaskChannel
            c.MapTaskChannel <- task
            if task.Status == Ready {
                t := time.Now()
                requesttaskreply.Task = Task{
                    Taskid:   c.Id,
                    TaskType: MapTask,
                    Nreduce:  c.ReduceTaskNum,
                    Files:    task.Files,
                    Status:   Running,
                }
                requesttaskreply.Done = false

                task.Taskid = c.Id
                task.Begin = t
                task.Status = Running
                c.Id++
                c.Lock.Unlock()
                return nil
            }
        }
        c.Lock.Unlock()
        if c.maptaskallfinish() {
            c.Lock.Lock()
            c.Phase = ReducePhase
            c.Lock.Unlock()
            c.generateReducTask()
        } else {
            time.Sleep(3 * time.Second)
        }
    } else {
        c.Lock.Lock()
        for i := 0; i < c.ReduceTaskNum; i++ {
            task := <-c.ReduceTaskChannel
            c.ReduceTaskChannel <- task
            if task.Status == Ready {
                t := time.Now()
                requesttaskreply.Task = Task{
                    Taskid:   c.Id,
                    TaskType: ReduceTask,
                    Nreduce:  c.ReduceTaskNum,
                    Files:    task.Files,
                    Status:   Running,
                }
                requesttaskreply.Done = false

                task.Taskid = c.Id
                task.Begin = t
                task.Status = Running
                c.Id++ //+++++++++++++++++++++++++++++++++
                c.Lock.Unlock()
                return nil
            }
        }
        c.Lock.Unlock()
        if c.reducetaskallfinish() {
            c.Lock.Lock()
            c.Phase = Finish
            c.IsDone = true
            c.Lock.Unlock()
            requesttaskreply.Done = true
            return nil
        } else {
            time.Sleep(3 * time.Second)
        }

    }
    return nil
}

// worker 完成一个map task通过rpc调用该函数返回任务结果
func (c *Coordinator) ReceivedMapTask(maptaskdoneargs *MapTaskDoneArgs, maptaskdonereply *MapTaskDoneReply) error {
    c.Lock.Lock()
    for i := 0; i < c.MapTaskNum; i++ {
        task := <-c.MapTaskChannel
        c.MapTaskChannel <- task
        task.checkTimeOutTask()
        if task.Status == Running && task.Taskid == maptaskdoneargs.Task.Taskid {
            err := receivedfiles(maptaskdoneargs.Files)
            if err != nil {
                log.Fatal("copy file error(map): ", err)
                c.Lock.Unlock()
                return nil
            }
            task.Status = Finish
            c.Lock.Unlock()
            return nil
        }
    }
    c.Lock.Unlock()
    return nil
}

// worker 完成一个reduce task通过rpc调用该函数返回任务结果
func (c *Coordinator) ReceivedReduceTask(reducetaskdoneargs *ReduceTaskDoneArgs, reducetaskdonereply *ReduceTaskDoneReply) error {
    c.Lock.Lock()
    for i := 0; i < c.ReduceTaskNum; i++ {
        task := <-c.ReduceTaskChannel
        c.ReduceTaskChannel <- task
        task.checkTimeOutTask()
        if task.Status == Running && task.Taskid == reducetaskdoneargs.Task.Taskid {
            err := receivedfiles(reducetaskdoneargs.Files)
            if err != nil {
                log.Fatal("copy file error(reduce): ", err)
                c.Lock.Unlock()
                return nil
            }
            task.Status = Finish
            c.Lock.Unlock()
            return nil
        }
    }
    c.Lock.Unlock()
    return nil
}

// 用于将/var/tmp中的临时文件copy到当前目录，表示接收任务结果
func receivedfiles(files []string) error {
    for _, filename := range files {
        srcf, e := os.Open(Tempfilepath + filename)
        if e != nil {
            return e
        }
        defer srcf.Close()
        destf, er := os.Create(filename)
        if er != nil {
            return er
        }
        defer destf.Close()
        _, err := io.Copy(destf, srcf)
        if err != nil {
            return err
        }
    }
    return nil
}

3. 您期望得到的结果？

我希望能解决发生data race问题，就是运行该程序时不发生race的报错。
//: <> (能截图就截图。)

4. 您实际得到的结果？

报错信息就在上面问题描述那里。
//: <> (有报错信息的话把堆栈信息提供出来)