Лучший способ избежать race condition, в данном случае?

Question

[Void Vasher]

Здравствуйте.

Есть сервис, который обрабатывает некоторый поток данных пулом воркеров.

Вот упрощенный, в целях демонстрации, код:

spoiler

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

var (
	maxWorkers = 5

	pool    = make(chan struct{}, maxWorkers)
	input   = make(chan int, 0)
	results = make([]int, 0)
)

func main() {
	go func() {
		defer close(input)

		// какой-то рандомный input
		rand.Seed(time.Now().Unix())
		limit := rand.Intn(40) + 1

		for i := 0; i < limit; i++ {
			input <- i
		}
	}()

	var wg sync.WaitGroup

	// главный процесс пула воркеров
	for i := range input {
		pool <- struct{}{}
		wg.Add(1)

		go func(i int) {
			defer func() {
				wg.Done()
				<-pool
			}()

			// здесь race condition
			results = append(results, pow(i+1, 2))
		}(i)
	}

	wg.Wait()

	fmt.Println(results)
}

func pow(n int, power int) int {
	fmt.Println("worker started")
	time.Sleep(time.Second * 2)

	return n * power
}

Как в этом случае избежать race condition? Mutex здесь не поможет, так как он будет блокировать всех воркеров, и они будут выполняться по очереди.
Из вариантов: использовать дополнительный канал.
Но может есть более удобный способ?

Comments

[Василий Банников]

Канал выглядит вполне адекватным вариантом. Но я на го много не пишу.

Answer 1

[Владислав]

Примеры с каналом
Когда знаем кол-во данных:
https://play.golang.org/p/7pNGCebvqlo
Когда не знаем:
https://play.golang.org/p/7WtVnXwjtuo

Подробности в комментариях:
https://qna.habr.com/answer?answer_id=1901263#comm...

Answer 2

[raiboon]

Два варианта.
1. Мьютекс. С чего вы взяли, что воркеры будут выполняться по очереди? Сначала идет какая-то долгая работа конкурентно, а потом мьютекс на короткий промежуток времени закрывает слайс, пока идет в него добавление результата.
2. Канал. Смысл в том, что результаты собираются в одной горутине. Тогда не будет конкурентной записи в слайс. Это более go way. Это можно делать в main или запустить еще одну.

Comments

[uvelichitel]

2. Канал. Это более go way.

Это очень даже go way. Только обычно встает вопрос - где и когда этот канал результатов закрыть и тут архитектура теряет элегантность.))

//типовая обработка канала результатов
for res := range resaultchan{
    //proccess res
    results = append(results, res)
}
//но чтобы детерменировано выйти из этого цикла
//нужно где то сделать close(resaultchan)
//в коде автора вопроса где и когда лучше это сделать?

[Владислав]

uvelichitel, канал можно закрыть после wg.Wait(), только нужно будет обернуть в горутину

go func() {
    wg.Wait()
    close(ch)
}()

for range...

Цикл заблокирует дальнейшее выполнение программы, пока не закроется канал.

[uvelichitel]

Владислав, если в коде автора wg.Wait() обернуть в горутину, то main() не будет никого ждать, а быстренько отработает и выйдет.)) Нужно еще замысловатей. У меня лично такие архитектуры элегантно не получаются.)) Правда в сервисах/демонах вопрос детерминированного выхода не встает - просто работает постоянно.

[Владислав]

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

То есть, меняем

wg.Wait()

fmt.Println(results)

на

go func() {
    wg.Wait()
    close(resultCh)
}()

for res := range resultCh {
    results = append(results, res)
}

fmt.Println(results)

[Void Vasher]

Владислав, при таком раскладе получаются дедлоки, если количество инпутов больше, чем воркеров https://play.golang.org/p/M8IOixsIOzV

Или я неправильно вас понял?

[Владислав]

Void Vasher, как-то так: https://play.golang.org/p/7pNGCebvqlo

Заполняю wg.Add() заранее, так как иначе wg.Wait() нечего будет ждать.
Запускаю чтение из input в горутине.

Вариант если мы не знаем размер input: https://play.golang.org/p/7WtVnXwjtuo
В нём я добавляю горутину заполняющую input в wg.

[Ромзес Панагиотис]

Владислав, uvelichitel спасибо, полезно знать.