go - 同时，如何管理值/状态并避免竞争条件

Question

在处理 Goroutines 时，如何根据进程启动后发生的事件/条件正确设置/修改值没有创造竞争条件。

例如，下面的“工作(有问题)”和输出是:

ping, foo=true
ping, foo=false
ping, foo=true
ping, foo=true
ping, foo=true

https://play.golang.org/p/Y3FafF-nBc

package main

import "fmt"

type test struct {
    ch  chan string
    foo bool
}

func (t *test) run() {
    for {
        select {
        case v := <-t.ch:
            fmt.Printf("%+v, foo=%+v\n", v, t.foo)
            t.foo = false
        default:
        }
    }
}

func (t *test) Ping() {
    t.ch <- "ping"
}

func New() *test {
    t := &test{
        ch: make(chan string),
    }
    go t.run()
    return t
}

func main() {
    t := New()
    for i := 0; i <= 10; i++ {
        if t.foo {
            t.Ping()
        }
        if i%3 == 0 {
            t.foo = true
        }
    }
}

但是如果使用 -race 编译或运行选项，我得到这个输出:

$ go run -race main.go
ping, foo=true
==================
WARNING: DATA RACE
Write at 0x00c4200761b8 by goroutine 6:
  main.(*test).run()
      /main.go:16 +0x1fb

Previous read at 0x00c4200761b8 by main goroutine:
  main.main()
      /main.go:37 +0x5e

Goroutine 6 (running) created at:
  main.New()
      /main.go:30 +0xd0
  main.main()
      /main.go:35 +0x33
==================
ping, foo=false
ping, foo=true
ping, foo=true
ping, foo=true
Found 1 data race(s)
exit status 66

因此，我想知道我可以使用什么并发模式来更改 gorutine 外部和 gorutine 内部的 foo 的值，而无需创建比赛条件。

Answer 1

你有一些选择:

• 使用 atomic.Value:示例 (1)
• 使用 sync.RWMutex:示例 (3)
• 使用sync/atomic:示例 (6)
• 仅使用 channel 和协程:示例 (7)

另见:Use a sync.Mutex or a channel?

---

1- 您可以使用 atomic.Value:

A Value provides an atomic load and store of a consistently typed value. Values can be created as part of other data structures. The zero value for a Value returns nil from Load. Once Store has been called, a Value must not be copied.

A Value must not be copied after first use.

喜欢这个工作示例:

// to test the panic use go build -race
package main

import (
    "fmt"
    "sync/atomic"
)

type test struct {
    ch chan string
    atomic.Value
}

func (t *test) run() {
    for {
        select {
        case v := <-t.ch:
            fmt.Printf("%+v, foo=%+v\n", v, t.Load())
            t.Store(false)
        default:
        }
    }
}

func (self *test) Ping() {
    self.ch <- "ping"
}

func New() *test {
    t := &test{
        ch: make(chan string),
    }
    t.Store(false)
    go t.run()
    return t
}

func main() {
    t := New()
    for i := 0; i <= 10; i++ {
        if x, _ := t.Load().(bool); x {
            t.Ping()
        }
        //  time.Sleep(time.Second)
        if i%3 == 0 {
            t.Store(true)
        }
    }
}

使用 go build -race 输出:

ping, foo=true
ping, foo=false
ping, foo=false
ping, foo=false
ping, foo=false

---

2- 对 func (t *test) run() 的一点改进:

func (t *test) run() {
    for v := range t.ch {
        fmt.Printf("%+v, foo=%+v\n", v, t.Load())
        t.Store(false)
    }
}

---

3- 您可以使用 sync.RWMutex 和 sync.WaitGroup，就像这个工作示例:

// to test the panic use go build -race
package main

import (
    "fmt"
    "sync"
)

type test struct {
    ch  chan string
    foo bool
    sync.RWMutex
    sync.WaitGroup
}

func (t *test) run() {
    for v := range t.ch {
        t.Lock()
        r := t.foo
        t.foo = false
        t.Unlock()
        fmt.Printf("%+v, foo=%+v\n", v, r)

    }
    t.Done()
}

func (self *test) Ping() {
    self.ch <- "ping"
}

func New() *test {
    t := &test{ch: make(chan string)}
    t.Add(1)
    go t.run()
    return t
}

func main() {
    t := New()
    for i := 0; i <= 10; i++ {
        t.RLock()
        r := t.foo
        t.RUnlock()
        if r {
            t.Ping()
        }
        //  time.Sleep(time.Second)
        if i%3 == 0 {
            t.Lock()
            t.foo = true
            t.Unlock()
        }
    }
    close(t.ch)
    t.Wait()
}

使用 go build -race 输出:

ping, foo=true
ping, foo=true
ping, foo=false
ping, foo=true
ping, foo=false
ping, foo=true

---

4- 因此，让我们遵循这种方法 https://talks.golang.org/2013/bestpractices.slide#29 :
原始代码:

package main

import (
    "fmt"
    "time"
)

type Server struct{ quit chan bool }

func NewServer() *Server {
    s := &Server{make(chan bool)}
    go s.run()
    return s
}

func (s *Server) run() {
    for {
        select {
        case <-s.quit:
            fmt.Println("finishing task")
            time.Sleep(time.Second)
            fmt.Println("task done")
            s.quit <- true
            return
        case <-time.After(time.Second):
            fmt.Println("running task")
        }
    }
}
func (s *Server) Stop() {
    fmt.Println("server stopping")
    s.quit <- true
    <-s.quit
    fmt.Println("server stopped")
}

func main() {
    s := NewServer()
    time.Sleep(2 * time.Second)
    s.Stop()
}

---

5- 让我们简化它:

package main

import (
    "fmt"
    "time"
)

var quit = make(chan bool)

func main() {
    go run()
    time.Sleep(2 * time.Second)
    fmt.Println("server stopping")

    quit <- true // signal to quit

    <-quit // wait for quit signal

    fmt.Println("server stopped")
}

func run() {
    for {
        select {
        case <-quit:
            fmt.Println("finishing task")
            time.Sleep(time.Second)
            fmt.Println("task done")
            quit <- true
            return
        case <-time.After(time.Second):
            fmt.Println("running task")
        }
    }
}

输出:

running task
running task
server stopping
finishing task
task done
server stopped

---

6- 示例的简化版本:

// to test the panic use go build -race
package main

import "fmt"
import "sync/atomic"

var ch = make(chan string)
var state int32

func main() {
    go run()
    for i := 0; i <= 10; i++ {
        if atomic.LoadInt32(&state) == 1 {
            ch <- "ping"
        }
        if i%3 == 0 {
            atomic.StoreInt32(&state, 1)
        }
    }
}

func run() {
    for v := range ch {
        fmt.Printf("%+v, state=%+v\n", v, atomic.LoadInt32(&state))
        atomic.StoreInt32(&state, 0)
    }
}

输出:

ping, state=1
ping, state=0
ping, state=1
ping, state=0
ping, state=1
ping, state=0

---

7- 使用 channel 但不使用 Lock() ( The Go Playground ) 的工作示例:

// to test the panic use go build -race
package main

import "fmt"

func main() {
    go run()
    for i := 0; i <= 10; i++ {
        signal <- struct{}{}
        if <-read {
            ping <- "ping"
        }
        if i%3 == 0 {
            write <- true
        }
    }
}

func run() {
    foo := false
    for {
        select {
        case <-signal:
            fmt.Println("signal", foo)
            read <- foo
        case foo = <-write:
            fmt.Println("write", foo)
        case v := <-ping:
            fmt.Println(v, foo)
            foo = false
        }
    }
}

var (
    ping   = make(chan string)
    signal = make(chan struct{})
    read   = make(chan bool)
    write  = make(chan bool)
)

输出:

signal false
write true
signal true
ping true
signal false
signal false
write true
signal true
ping true
signal false
signal false
write true
signal true
ping true
signal false
signal false
write true
signal true
ping true