Обработка данных в конкурентных программах

В Go у нас есть функциональность горутин из коробки. Мы можем запускать код параллельно. Однако в нашем параллельно выполняющемся коде мы можем работать с общими переменными, и не совсем понятно, как именно Go обрабатывает такие ситуации.

В Go у нас есть функциональность горутин из коробки. Мы можем запускать код параллельно. Однако в нашем параллельно выполняющемся коде мы можем работать с общими переменными, и не совсем понятно, как именно Go обрабатывает такие ситуации.

Начнём с задачи “счётчик” — попробуем увеличить переменную-счётчик 200 раз в нескольких горутинах.

c := 0
wg := sync.WaitGroup{}
n := 200
wg.Add(n)
for i := 0; i < n; i++ {
	go func() {
		c++
		wg.Done()
	}()
}
wg.Wait()

fmt.Println(c)

// 194

Результирующее значение счётчика каждый раз отличается и в большинстве случаев не равно 200. Таким образом, этот код не является потокобезопасным и не работает как задумано, даже если у нас нет ошибок компилятора или времени выполнения.

Следующий случай — попробуем вставить 200 значений в срез параллельно и проверим, есть ли там ровно 200 значений.

c := []int{}
wg := sync.WaitGroup{}
n := 200
wg.Add(n)
for i := 0; i < n; i++ {
	go func() {
		c = append(c, 1)
		wg.Done()
	}()
}
wg.Wait()

fmt.Println(len(c))

// 129

Количество значений в срезе ещё дальше от 200, чем было в задаче со счётчиком. Этот код также не является потокобезопасным.

Попробуем вставить 200 значений в map параллельно:

c := map[int]int{}
wg := sync.WaitGroup{}
n := 200
wg.Add(n)
for i := 0; i < n; i++ {
	go func(i int) {
		c[i] = i
		wg.Done()
	}(i)
}
wg.Wait()

fmt.Println(len(c))

// panic: concurrent map writes

Мы не можем проверить результат из-за паники.

Во всех 3 задачах у нас есть неработающий код, но только с map есть сообщение об ошибке о конкурентных записях в map, реализованное разработчиками Go.

Обнаружение гонок данных

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

Можно запустить любой из приведённых выше тестовых случаев с флагом race — go test -race ./test.go. В результате Go отображает горутины с гонкой данных:

go test -race ./test.go

==================
WARNING: DATA RACE
Read at 0x00c0000a6070 by goroutine 9:
command-line-arguments.Test.func1()
/go/src/github.com/antelman107/go_blog/test.go:16 +0x38

Previous write at 0x00c0000a6070 by goroutine 8:
command-line-arguments.Test.func1()
/go/src/github.com/antelman107/go_blog/test.go:16 +0x4e

Goroutine 9 (running) created at:
command-line-arguments.Test()
/go/src/github.com/antelman107/go_blog/test.go:15 +0xe8
testing.tRunner()
/usr/local/Cellar/go/1.14/libexec/src/testing/testing.go:992 +0x1eb
--- FAIL: Test (0.01s)
testing.go:906: race detected during execution of test
FAIL
FAIL    command-line-arguments  0.025s
FAIL

Обнаружение гонок данных — это не функциональность go test. Можно даже собрать программу в режиме обнаружения гонок:

$ go test -race mypkg    // для тестирования пакета
$ go run -race .  // для запуска исходного файла
$ go build -race .   // для сборки команды
$ go install -race mypkg // для установки пакета

Хорошо, что можно напрямую обнаруживать гонки данных в программе.

Даже популярная проблема “замыкания цикла” может быть обнаружена:

wg := sync.WaitGroup{}
n := 10
wg.Add(n)
for i := 0; i < n; i++ {
	go func() {
		fmt.Println(i)
		wg.Done()
	}()
}
wg.Wait()

Проблема здесь в том, что код не выведет точные числа 0, 1, 2 … 9, а случайные числа от 0 до 9.

Решение

Опишем решение для задачи со счётчиком. Это решение можно использовать для задач со срезом и map.

Итак, у нас есть значение счётчика, которое меньше ожидаемого. Несмотря на краткость вызова инкремента (c++), программа фактически выполняет следующий список действий:

1. чтение текущего значения счётчика из памяти,
2. его увеличение,
3. сохранение результата в память.

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

Чем больше у нас таких ситуаций чтения одного и того же начального значения, тем больше результат счётчика отличается от 200.

Решением здесь может быть атомарное изменение переменной. Если какая-то горутина читает начальное значение счётчика, следующим действием должно быть единственное обновление счётчика от этой горутины. Ни одна из других горутин не должна обращаться к счётчику или изменять его в середине этой операции.

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

Решение sync.Mutex/sync.RWMutex

Мы можем использовать методы Lock и Unlock для гарантии того, что только одна горутина работает со счётчиком в каждый момент времени.

Мы также можем использовать sync.RWMutex для обеспечения параллельных чтений.

Но в нашей задаче Mutex полностью достаточен:

c := 0
n := 200
m := sync.Mutex{}
wg := sync.WaitGroup{}
wg.Add(n)
for i := 0; i < n; i++ {
	go func(i int) {
		m.Lock()
		c++
		m.Unlock()
		wg.Done()
	}(i)
}
wg.Wait()

fmt.Println(c)

// 200 == OK

Решение через каналы

Операции с каналами являются атомарными из коробки.

Мы можем отправлять любые данные в канал с одним читателем для обеспечения последовательной обработки.

Но для этого нам нужен дополнительный код:

c := 0
n := 200

ch := make(chan struct{}, n)
chanWg := sync.WaitGroup{}
chanWg.Add(1)
go func() {
	for range ch {
		c++
	}
	chanWg.Done()
}()

wg := sync.WaitGroup{}
wg.Add(n)
for i := 0; i < n; i++ {
	go func(i int) {
		ch <- struct{}{}
		wg.Done()
	}(i)
}
wg.Wait()
close(ch)
chanWg.Wait()

fmt.Println(c)

// 200 = OK

Мы также использовали здесь пустую структуру, потому что это наименьший по размеру тип данных переменной в Go.

Решение через пакет atomic

Стандартный пакет Go под названием atomic предоставляет набор атомарных операций.

Благодаря функциям runtime_procPin / runtime_procUnpin (в исходниках Go).

Функция Pin гарантирует, что планировщик Go не запустит никакую другую горутину до тех пор, пока не будет вызван Unpin.

У нас есть несколько функций счётчика в пакете atomic, которые помогают реализовать наш атомарный счётчик:

c := int32(0)
n := 200

wg := sync.WaitGroup{}
wg.Add(n)
for i := 0; i < n; i++ {
	go func(i int) {
		atomic.AddInt32(&c, 1)
		wg.Done()
	}(i)
}
wg.Wait()

fmt.Println(c)

// 200 = OK

Проблему атомарного изменения данных можно встретить во многих ситуациях разработки. Например, та же проблема возникает с запросами SELECT + UPDATE в SQL базах данных при работе нескольких процессов.