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

Недавно я понял, что не знаю, как правильно 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(), потому что мьютекс в конечном итоге будет заблокирован в основном все время. Чтобы заставить его работать правильно, нам потребуется какое-то взаимодействие между чтением и блокировкой, как это делают условные переменные: мьютекс разблокируется до того, как горутина будет ждать, и снова заблокируется, когда горутина проснется.

Я бы реализовал что-то подобное вручную, но тогда мне нужен какой-то способ select во время чтения. Вот так: (псевдокод)

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

Я изучил документы пакетов os и sync, и пока я не вижу способа сделать это.


go multithreading race-condition concurrency goroutine
person Dmitry Frank    schedule 18.01.2017    source источник
comment
Вам действительно нужно закрыть файл? Самый безопасный метод — просто оставить горутину чтения до завершения процесса.   -  person JimB    schedule 18.01.2017
comment
Я не понимаю, почему вы хотите использовать дескриптор файла в другом потоке выполнения или, если на то пошло, любой ресурс. Это только усложнит ваш код.   -  person Ankur    schedule 18.01.2017
comment
@JimB, мне нужно закрыть файл, чтобы реализовать переподключение: например. когда я отключу устройство, узел которого был /dev/ttyUSB0, и не закрою файл, то файл /dev/ttyUSB0 все равно будет открыт, а когда я снова подключу устройство, он станет /dev/ttyUSB1. Мне нужно, чтобы снова было /dev/ttyUSB0.   -  person Dmitry Frank    schedule 18.01.2017
comment
@DmitryFrank: если удаление устройства не разблокирует чтение, как узнать, что устройство было удалено?   -  person JimB    schedule 18.01.2017
comment
@JimB, хорошо, это плохой пример: в этом конкретном случае Read действительно вернет io.EOF, но вопрос более общий. Например. рассмотрите возможность наличия горутины, которая считывает данные из последовательного порта и отправляет полученные данные на какой-либо канал, а у пользователя есть кнопка отключения, которая, очевидно, должна закрывать файл, и делать это без гонок.   -  person Dmitry Frank    schedule 18.01.2017
comment
@DmitryFrank: дело не только в гонке в Go - POSIX не позволяет отменить чтение, закрыв файл из другого потока. Единственным безопасным решением здесь является опрос файла через select/epoll/etc в файловом дескрипторе. Это по-прежнему оставляет крошечную гонку в пределах тайм-аута опроса, но это не является чем-то, что присуще Go в частности.   -  person JimB    schedule 18.01.2017

Ответы (1)


arrow_upward
-1
arrow_downward

Я считаю, что вам нужно 2 сигнала:

  1. main -> reader, чтобы остановить чтение
  2. читатель -> основной, чтобы сообщить, что читатель был завершен

конечно, вы можете выбрать примитив передачи сигналов (канал, группа ожидания, контекст и т. д.), который вы предпочитаете.

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

Я создал несколько процедур go только в качестве примера, с которым вы даже можете координировать несколько процедур 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)
        }
    }
}
person ahmy    schedule 18.01.2017
comment
Чтение без входящих данных будет заблокировано на неопределенный срок. Это не может прервать вызов чтения. - person JimB; 19.01.2017
comment
возможно, вы правы, я посмотрю, изменю ли я его на ch ‹-f.Read(b) вместо значения по умолчанию: - person ahmy; 19.01.2017
comment
Это по-прежнему ничего не меняет, потому что Read оценивается первым. Вы просто не можете прервать чтение непосредственно в файле на Go (пока). - person JimB; 19.01.2017