一段时间后停止 goroutine

Question

像这样在一段时间后停止 goroutine 并发安全吗？（注意：由于另一个 goroutine 发生变化而导致数据竞争）：okpackage mainimport (    "fmt"    "time")func main() {    var ok byte    time.AfterFunc(1000*time.Millisecond, func() {        ok = 1    })    var i uint64    for ok == 0 {        i++ // CPU intensive task    }    fmt.Println(i) // 2_776_813_033}终端：go run -race .==================WARNING: DATA RACEWrite at 0x00c000132010 by goroutine 8:  main.main.func1()      ./main.go:11 +0x46Previous read at 0x00c000132010 by main goroutine:  main.main()      ./main.go:15 +0xf4Goroutine 8 (running) created at:  time.goFunc()      go/src/time/sleep.go:180 +0x51==================80849692Found 1 data race(s)代码（无数据竞争）：package mainimport (    "fmt"    "sync/atomic"    "time")func main() {    var ok int32    time.AfterFunc(1000*time.Millisecond, func() {        atomic.StoreInt32(&ok, 1)    })    var i uint64    for atomic.LoadInt32(&ok) == 0 {        i++ // CPU intensive task    }    fmt.Println(i) // 2_835_935_488}终端：go run -race .31934042

Answer 1

即使被另一个 goroutineok设置为busy-wait for 循环，也不能保证它会终止。false在设置和读取 的过程中没有显式同步ok，因此不能保证主 Goroutine 能够看到对其所做的更改。换句话说，没有办法在两个 goroutine 之间建立发生之前的关系。

https://golang.org/ref/mem

代码的第二个版本是安全的，尽管 Go 内存模型中没有针对 进行说明ok，但它并不安全，因为如此紧密的循环可能不允许其他 goroutine 执行。原子读/写具有发生之前关系所需的内存屏障。您应该使用同步原语之一（互斥体、通道）来保证这一点。

Answer 2

第二个代码适用于 Go 1.14+：

去1.14

Goroutines 现在是异步可抢占的。因此，没有函数调用的循环不再可能导致调度程序死锁或显着延迟垃圾收集。除了 windows/arm、darwin/arm、js/wasm 和 plan9/* 之外，所有平台都支持此功能。

一段时间后停止 goroutine

BenchmarkAfterFunc-8            1000000000 0.4468 ns/op  0 B/op  0 allocs/op

BenchmarkDoneChannel-8          121966824   9.855 ns/op  0 B/op  0 allocs/op

BenchmarkTimeSince-8            89790115    12.95 ns/op  0 B/op  0 allocs/op

BenchmarkContextErr-8           58508900    19.78 ns/op  0 B/op  0 allocs/op

BenchmarkAfterFuncMutex-8       58323207    20.00 ns/op  0 B/op  0 allocs/op

BenchmarkContext-8              48947625    27.43 ns/op  0 B/op  0 allocs/op

测试：

package main

import (

    "context"

    "sync"

    "sync/atomic"

    "testing"

    "time"

)

const d = 200 * time.Millisecond //  To stop a task after a period of time

func BenchmarkTimeSince(b *testing.B) {

    t0 := time.Now()

    var count = 0

    for i := 0; i < b.N; i++ {

        if time.Since(t0) < d {

            count++

        }

    }

    _ = count

}

func BenchmarkContext(b *testing.B) {

    var ctx, cancel = context.WithTimeout(context.Background(), d)

    defer cancel()

    var count = 0

    for i := 0; i < b.N; i++ {

        select {

        case <-ctx.Done():

            // break

        default:

            count++

        }

    }

    _ = count

}

func BenchmarkContextErr(b *testing.B) {

    var ctx, cancel = context.WithTimeout(context.Background(), d)

    defer cancel()

    var count = 0

    for i := 0; i < b.N; i++ {

        if ctx.Err() == nil {

            count++

        }

    }

    _ = count

}

func BenchmarkAfterFunc(b *testing.B) {

    var done uint32

    time.AfterFunc(d, func() { atomic.StoreUint32(&done, 1) })

    var count = 0

    for i := 0; i < b.N; i++ {

        if atomic.LoadUint32(&done) == 0 {

            count++

        }

    }

    _ = count

}

func BenchmarkDoneChannel(b *testing.B) {

    var done = make(chan struct{})

    time.AfterFunc(d, func() { close(done) })

    var count = 0

    for i := 0; i < b.N; i++ {

        select {

        case <-done:

            // break

        default:

            count++

        }

    }

    _ = count

}

type foo struct {

    sync.Mutex

    state bool

}

func (p *foo) end() {

    p.Lock()

    p.state = true

    p.Unlock()

}

func (p *foo) isDone() bool {

    var b bool

    p.Lock()

    b = p.state

    p.Unlock()

    return b

}

func BenchmarkAfterFuncMutex(b *testing.B) {

    var it = foo{}

    time.AfterFunc(d, func() { it.end() })

    var count = 0

    for i := 0; i < b.N; i++ {

        if it.isDone() {

            count++

        }

    }

    _ = count

}

https://medium.com/a-journey-with-go/go-asynchronous-preemption-b5194227371c

抢占是调度程序的重要组成部分，它可以在 goroutine 之间分配运行时间。事实上，如果没有抢占，一个长时间运行的 goroutine 会占用 CPU，从而阻止其他 goroutine 被调度。1.14版本引入了异步抢占的新技术，为调度程序提供了更多的权力和控制权。