在 Go 中使用 sync.Atomic 包防止竞争条件

Question

我在 Go 中实现了以下计数器，我想同时使用它。我正在使用 atomic 包来存储状态，但不确定是否会遇到任何竞争条件。例如，我是否还需要添加一个额外的互斥锁以防止增量低于零，或者原子操作是否提供足够的安全性？谢谢！type Counter struct {    counter  uint64    finished uint32    sync.Mutex}// Inc increments the counter by onefunc (c *Counter) Inc() error {    if c.isFinished() {        return fmt.Errorf("counter is finished")    }    atomic.AddUint64(&c.counter, 1)    return nil}// Dec decrements the counter by one, but prevents the counter from going to zerofunc (c *Counter) Dec() {    // prevent overflow    if !c.isZero() {        atomic.AddUint64(&c.counter, ^uint64(0))    }}// Cancel sets the finished flag, and sets counter to zerofunc (c *Counter) Cancel() {    if !c.isFinished() {        atomic.StoreUint32(&c.finished, 1)        atomic.StoreUint64(&c.counter, 0)    }}// isFinished returns true if finishedfunc (c *Counter) isFinished() bool {    return atomic.LoadUint32(&c.finished) == uint32(1)}// isZero returns true if counter is zerofunc (c *Counter) isZero() bool {    return atomic.LoadUint64(&c.counter) == uint64(0)}更新：通过使用-race标志运行下面的代码，我能够检测到我确实需要包含互斥锁。var counter *Counter = &Counter{}func main() {    wg := sync.WaitGroup{}    numberOfLoops := 10    wg.Add(numberOfLoops)    for i := 0; i < numberOfLoops; i++ {        go incrementor(&wg)    }    wg.Wait()    fmt.Println("Final Counter:", counter.counter)}func incrementor(wg *sync.WaitGroup) {    rand.Seed(time.Now().UnixNano())    for i := 0; i < 20; i++ {        counter.Inc()        time.Sleep(time.Duration(rand.Intn(3)) * time.Millisecond)    }    wg.Done()    counter.Cancel()}

Answer 1

您不需要额外的互斥锁，您的主要功能是在counter.counter不使用原子负载的情况下读取，而您的incrementor调用wg.Done()before counter.Cancel()，因此您会获得竞争条件。

通过在竞争条件解决wg.Done()后移动：counter.Cancel()

func main() {

    wg := sync.WaitGroup{}

    numberOfLoops := 10

    wg.Add(numberOfLoops)

    for i := 0; i < numberOfLoops; i++ {

        go incrementor(&wg)

    }

    wg.Wait()

    fmt.Println("Final Counter:", counter.counter)

}

func incrementor(wg *sync.WaitGroup) {

    rand.Seed(time.Now().UnixNano())

    for i := 0; i < 20; i++ {

        counter.Inc()

        time.Sleep(time.Duration(rand.Intn(3)) * time.Millisecond)

    }

    counter.Cancel()

    wg.Done()

}

Answer 2

为了避免“可被比赛检测器检测到”的比赛：

正如@caveman 回答的那样，您遇到的问题与wg.Done() / wg.Wait()指令排序中的问题有关，并且您不使用 anatomic.Load()来访问counter.counter.

对于这种竞争条件，您的方法是“安全的”。

为了避免比赛，如“不要为计数器达到不连贯的状态”：

你确实有一个问题，（*）因为你的方法运行了几个连续的指令来检查和更新对象（例如 : if condition { update_fields }），并且你没有同步机制来检查condition应用时是否仍然正确update_fields。

通过将您的incrementor功能更改为：

func incrementor(wg *sync.WaitGroup) {

    for i := 0; i < 20000; i++ {

        counter.Inc()

    }

    counter.Cancel()

    wg.Done()

}

并多次运行您的程序，您应该能够看到“Final Counter：”并不总是以 0（操场）结尾。

这是如何发生这种情况的说明：

假设 goroutine 1 执行counter.Cancel()

而 goroutine 2 执行counter.Inc()

可能会发生以下执行顺序：

   goroutine 1                                 goroutine 2

                                            1. if c.isFinished() {

                                                   return fmt.Errorf("counter is finished")

                                               }

2. if !c.isFinished() {                         

3.     atomic.StoreUint32(&c.finished, 1)       

4.     atomic.StoreUint64(&c.counter, 0)        

   }                                            

                                            5. atomic.AddUint64(&c.counter, 1)

                                            6. return nil

中的c.isFinished()指令.Inc()可能在执行之前发生， .Cancel()

并且atomic.AddUint64(&c.counter, 1)可能在将计数器重置为零之后发生。 .Cancel()

为了避免这种竞争，您需要选择一种同步inspect + update指令的方式。

一种常见的方法是使用互斥锁：

type Counter struct {

    counter  uint64

    finished uint32

    m        sync.Mutex

}

// Inc increments the counter by one

func (c *Counter) Inc() error {

    c.m.Lock()

    defer c.m.Unlock()

    if c.finished != 0 {

        return fmt.Errorf("counter is finished")

    }

    c.counter++

    return nil

}

// Dec decrements the counter by one, but prevents the counter from going to zero

func (c *Counter) Dec() {

    c.m.Lock()

    defer c.m.Unlock()

    // prevent overflow

    if c.counter > 0 {

        c.counter--

    }

}

// Cancel sets the finished flag, and sets counter to zero

func (c *Counter) Cancel() {

    c.m.Lock()

    defer c.m.Unlock()

    if c.finished == 0 {

        c.finished = 1

        c.counter = 0

    }

}