Параллельное чтение/закрытие в Go, в кроссплатформенном виде

Question

Я недавно понял, что не знаю, как правильно Read и Close в Go одновременно. В моем конкретном случае мне нужно сделать это с помощью последовательного порта, но проблема более общая.

Если мы сделаем это без каких-либо дополнительных усилий для синхронизации вещей, это приведет к состоянию гонки. Простой пример:

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 и т. д.), который вы предпочитаете.

Пример ниже, я использую waitgroup и контекст. Причиной является , что вы можете вращать несколько считывателей, и нужно только закрыть контекст, чтобы рассказать все, что нужно сделать для чтения.

Я создал процедуру множественного доступа так же, как пример, который вы можете даже координировать с ним несколько рутинных процедур.

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