并行读取/关闭在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事情，而阅读。像这样的：（伪）

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

我检查包os和sync的文档，到目前为止我没有看到任何方式做到这一点。

Answer 1

我相信你需要2个信号：

1. 主 - >读者，告诉它停止阅读
2. 阅读器 - >主，告诉读者已经终止

的

当然你可以选择你喜欢的去信令原语（信道，等待组，上下文等）。

下面的例子，我使用了waitgroup和context。原因是 ，你可以旋转多个阅读器，只需要关闭上下文来告诉所有阅读器去程序停止。

我创建了多个去程序，就像 一个例子，您甚至可以协调多个去程序。

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) 
     } 
    } 
}