Golang 数据竞争由同步 map 读取引起

Question

我有一个服务器来处理事件，这个服务器有一个互斥锁和一个事件表(映射结构)。当服务器收到一个新的事件时，它会获取lock来防止数据竞争，将这个事件存储在events表中，并启动一个goroutine来监控这个事件已经完成。如果我使用 -race 标志运行程序，它将输出 data race。

package main

import (
    "sync"
    "time"
)

type event struct {
    done chan bool
}

type server struct {
    mu     sync.Mutex
    events map[int]*event
}

func main() {
    s := server{}
    s.events = make(map[int]*event)

    for i := 0; i < 10; i++ {
        go func(i int) {
            s.mu.Lock()
            s.events[i] = &event{}
            s.events[i].done = make(chan bool)
            s.mu.Unlock()
            go func() {
                time.Sleep(1 * time.Millisecond)
                <-s.events[i].done
                // server do something.
            }()
        }(i)
    }

    time.Sleep(1 * time.Second)

    for i := 0; i < 10; i++ {
        // event happen.
        s.events[i].done <- true
    }
}

输出

==================
WARNING: DATA RACE
Read at 0x00c00010dd10 by goroutine 14:
  runtime.mapaccess1_fast64()
      c:/go/src/runtime/map_fast64.go:12 +0x0    
  main.main.func1.1()
      C:/SimpleAsyncBFT/race/main.go:29 +0x7c    

Previous write at 0x00c00010dd10 by goroutine 15:
  runtime.mapassign_fast64()
      c:/go/src/runtime/map_fast64.go:92 +0x0    
  main.main.func1()
      C:/SimpleAsyncBFT/race/main.go:24 +0xbe    

Goroutine 14 (running) created at:
  main.main.func1()
      C:/SimpleAsyncBFT/race/main.go:27 +0x1c6

Goroutine 15 (finished) created at:
  main.main()
      C:/SimpleAsyncBFT/race/main.go:22 +0xed

我知道在monitor goroutine中加入lock可以解决这个问题，但是会造成死锁! done channel 只是用来通知服务器事件已经完成。如果 channel 不适合这种情况，如何实现？

Answer 1

根据评论，您的代码尝试同时读取和写入 map ，并且根据 go 1.6 release notes :

if one goroutine is writing to a map, no other goroutine should be reading or writing the map concurrently

查看您的代码，似乎没有必要这样做。您可以提前创建 channel ；创建它们后，您只能从 map 中读取，所以没有问题:

package main

import (
    "sync"
    "time"
)

type event struct {
    done chan bool
}

type server struct {
    mu     sync.Mutex
    events map[int]*event
}

func main() {
    s := server{}
    s.events = make(map[int]*event)

    for i := 0; i < 10; i++ {
        s.events[i] = &event{}
        s.events[i].done = make(chan bool)
    }

    for i := 0; i < 10; i++ {
        go func(i int) {
            time.Sleep(1 * time.Millisecond)
            <-s.events[i].done
            // server do something.
        }(i)
    }

    time.Sleep(1 * time.Second)

    for i := 0; i < 10; i++ {
        // event happen.
        s.events[i].done <- true
    }
}

或者不在 go 例程中访问 map :

package main

import (
    "sync"
    "time"
)

type event struct {
    done chan bool
}

type server struct {
    mu     sync.Mutex
    events map[int]*event
}

func main() {
    s := server{}
    s.events = make(map[int]*event)

    for i := 0; i < 10; i++ {
        s.events[i] = &event{}
        c := make(chan bool)
        s.events[i].done = c

        go func(i int, c chan bool) {
            time.Sleep(1 * time.Millisecond)
            <-c
            // server do something.
        }(i, c)
    }

    time.Sleep(1 * time.Second)

    for i := 0; i < 10; i++ {
        // event happen.
        s.events[i].done <- true
    }
}

在您询问如何处理不知道事件数量的情况的评论中。解决方案将取决于具体情况，但这是我用来处理类似情况的一种方法(这看起来很复杂，但我认为使用 map 并在 Mutex ).

package main

import (
    "sync"
    "time"
)

type event struct {
    done chan bool
}

type server struct {
    events map[int]*event
}

func main() {
    s := server{}
    s.events = make(map[int]*event)

    // Routine to trigger channels
    triggerChan := make(chan chan bool) // Send new triggers to this...
    eventChan := make(chan struct{})    // Close this when the event happens and go routines should continue
    go func() {
        var triggers []chan bool
        eventReceived := false
        for {
            select {
            case t, ok := <-triggerChan:
                if !ok { // You want some way for the goRoutine to shut down - in this case we wait on the closure of triggerChan
                    return
                }
                if eventReceived {
                    t <- true // The event has already happened so go routine can proceed immediately
                } else {
                    triggers = append(triggers, t)
                }
            case <-eventChan:
                for _, c := range triggers {
                    c <- true
                }
                eventReceived = true
                eventChan = nil // Don't want select to be triggered again...
            }
        }
    }()

    // Start up the event handlers...
    var wg sync.WaitGroup
    wg.Add(10)
    for i := 0; i < 10; i++ {
        s.events[i] = &event{}
        c := make(chan bool)
        triggerChan <- c

        go func(i int, c chan bool) {
            time.Sleep(1 * time.Millisecond)
            <-c
            // server do something.
            wg.Done()
        }(i, c)
    }

    time.Sleep(1 * time.Second)

    // Event happened - release the go routines
    close(eventChan)
    wg.Wait()
    close(triggerChan)
}