multithreading - Go 中的并发读取/关闭，以跨平台的方式

Question

我最近意识到我不知道如何在 Go 中同时正确地Read 和Close。在我的特殊情况下，我需要使用串行端口来执行此操作，但问题更为普遍。

如果我们在没有任何额外努力同步事物的情况下这样做，就会导致竞争条件。简单示例:

package main

import (
    "fmt"
    "os"
    "time"
)

func main() {
    f, err := os.Open("/dev/ttyUSB0")
    if err != nil {
        panic(err)
    }

    // Start a goroutine which keeps reading from a serial port
    go reader(f)

    time.Sleep(1000 * time.Millisecond)
    fmt.Println("closing")
    f.Close()
    time.Sleep(1000 * time.Millisecond)
}

func reader(f *os.File) {
    b := make([]byte, 100)
    for {
        f.Read(b)
    }
}

如果我们将上面的内容保存为 main.go，然后运行 ​​go run --race main.go，输出将如下所示:

closing
==================
WARNING: DATA RACE
Write at 0x00c4200143c0 by main goroutine:
  os.(*file).close()
      /usr/local/go/src/os/file_unix.go:143 +0x124
  os.(*File).Close()
      /usr/local/go/src/os/file_unix.go:132 +0x55
  main.main()
      /home/dimon/mydata/projects/go/src/dmitryfrank.com/testfiles/main.go:20 +0x13f

Previous read at 0x00c4200143c0 by goroutine 6:
  os.(*File).read()
      /usr/local/go/src/os/file_unix.go:228 +0x50
  os.(*File).Read()
      /usr/local/go/src/os/file.go:101 +0x6f
  main.reader()
      /home/dimon/mydata/projects/go/src/dmitryfrank.com/testfiles/main.go:27 +0x8b

Goroutine 6 (running) created at:
  main.main()
      /home/dimon/mydata/projects/go/src/dmitryfrank.com/testfiles/main.go:16 +0x81
==================
Found 1 data race(s)
exit status 66

好的，但是如何正确处理呢？当然，我们不能在调用 f.Read() 之前就锁定一些互斥锁，因为互斥锁基本上会一直处于锁定状态。为了使其正常工作，我们需要在读取和锁定之间进行某种协作，就像条件变量所做的那样:互斥体在让 goroutine 等待之前被解锁，并且在 goroutine 唤醒时被锁定。

我会手动实现类似的东西，但是我需要一些方法来在阅读时选择东西。像这样:(伪代码)

select {
case b := <-f.NextByte():
  // process the byte somehow
default:
}

我检查了包的文档 os和 sync ，到目前为止，我还没有看到任何方法。

Answer 1

我相信你需要 2 个信号:

1. main -> reader，告诉它停止阅读
2. reader -> main，告诉reader已经终止

当然，您可以选择您喜欢的信号原语( channel 、 WaitGroup 、上下文等)。

下面的示例，我使用 WaitGroup 和上下文。原因是您可以旋转多个阅读器，只需要关闭上下文即可告诉所有阅读器 go-routine 停止。

我创建了多个 go routine 就像一个示例，您甚至可以使用它来协调多个 go 例程。

package main

import (
    "context"
    "fmt"
    "os"
    "sync"
    "time"
)

func main() {

    ctx, cancelFn := context.WithCancel(context.Background())

    f, err := os.Open("/dev/ttyUSB0")
    if err != nil {
        panic(err)
    }

    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {
        wg.Add(1)

        // Start a goroutine which keeps reading from a serial port
        go func(i int) {
            defer wg.Done()
            reader(ctx, f)
            fmt.Printf("reader %d closed\n", i)
        }(i)
    }

    time.Sleep(1000 * time.Millisecond)
    fmt.Println("closing")
    cancelFn() // signal all reader to stop
    wg.Wait()  // wait until all reader finished
    f.Close()
    fmt.Println("file closed")
    time.Sleep(1000 * time.Millisecond)
}

func reader(ctx context.Context, f *os.File) {
    b := make([]byte, 100)
    for {
        select {
        case <-ctx.Done():
            return
        default:
            f.Read(b)
        }
    }
}