英文:
Golang data race cause by consurrent map read
问题
我有一个用于处理事件的服务器,该服务器有一个互斥锁
和一个事件
表(映射结构)。当服务器接收到一个新事件时,它会获取锁
以防止数据竞争,将此事件存储在事件表中,并启动一个goroutine来监视此事件是否完成。如果我使用-race
标志运行程序,它会输出数据竞争
。
我知道在监视goroutine中添加锁
可以解决这个问题,但会导致死锁!done
通道只是用于通知服务器事件已完成。如果通道不适用于此条件,该如何实现呢?
英文:
I have a server to handle events, this server has a mutex lock
and a events
table(map structure). When the server receives a new event, it will acquire lock
to prevent data race, store this event in the events table, and start a goroutine to monitor this event has done. If I run the program with -race
flag, it will output 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
}
}
Output
==================
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
I know adding lock
in the monitor goroutine will solve this problem, but will cause deadlock! The done
channel is just used to notify the server that the event has been done. If channel was not suitable for this condition, how to achieve this?
答案1
得分: 4
根据你的代码注释,你的代码尝试同时读取和写入一个映射,根据go 1.6发布说明的规定:
如果一个goroutine正在写入一个映射,其他goroutine不应该同时读取或写入该映射。
从你的代码来看,似乎没有必要这样做。你可以提前创建通道;在创建后,你只是从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
// 服务器做一些操作。
}(i)
}
time.Sleep(1 * time.Second)
for i := 0; i < 10; i++ {
// 事件发生。
s.events[i].done <- true
}
}
或者在goroutine中不访问映射:
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
// 服务器做一些操作。
}(i, c)
}
time.Sleep(1 * time.Second)
for i := 0; i < 10; i++ {
// 事件发生。
s.events[i].done <- true
}
}
在评论中,你问到如何处理不知道事件数量的情况。解决方案将取决于具体情况,但下面是我处理类似情况的一种方式(这看起来很复杂,但我认为比使用映射并在每次访问时都加锁更容易理解)。
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)
// 触发通道的例程
triggerChan := make(chan chan bool) // 将新的触发器发送到这里...
eventChan := make(chan struct{}) // 当事件发生时关闭此通道,goroutine应该继续执行
go func() {
var triggers []chan bool
eventReceived := false
for {
select {
case t, ok := <-triggerChan:
if !ok { // 你希望有一种方法让goroutine关闭 - 在这种情况下,我们等待triggerChan关闭
return
}
if eventReceived {
t <- true // 事件已经发生,因此goroutine可以立即继续执行
} else {
triggers = append(triggers, t)
}
case <-eventChan:
for _, c := range triggers {
c <- true
}
eventReceived = true
eventChan = nil // 不希望再次触发select...
}
}
}()
// 启动事件处理程序...
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
// 服务器做一些操作。
wg.Done()
}(i, c)
}
time.Sleep(1 * time.Second)
// 事件发生 - 释放goroutine
close(eventChan)
wg.Wait()
close(triggerChan)
}
英文:
As per the comments your code attempts to read and write to a map simultaneously and, as per the go 1.6 release notes:
>if one goroutine is writing to a map, no other goroutine should be reading or writing the map concurrently
Looking at your code there appears to be no need for this. You can create the channels in advance; after they are created you are only reading from the map
so there is no issue:
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
}
}
Alternatively don't access the map in the go routine:
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
}
}
In the comments you asked about dealing with a situation where you don't know the number of events. The solution is going to depend on the situation but here is one way I've used to deal with similar situations (this appears complicated but I think its easier to follow then using a map and surrounding every access in a 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)
}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论