Как в Go асинхронно выполнять поступающие задачи и выводить конечный результат?

Question

[Алексей]

Имеется N однотипных задач. Необходимо выполнить их асинхронно и записать результаты.
Для выполнения задачи раньше я запускал N горутин с разными параметрами для каждой задачи, а результаты складывал в MAP
Дожидался выполнения всех функций с помощью sync.WaitGroup и потом уже генерировал ответ.

Теперь появилась необходимость дополнять список из N задач прямо из уже работающей функции.
Например запустилось 10 горутин, и внутри одной горутины появилась необходимость выполнить еще 2 задачи. (В какой горутине, сколько новых задач нужно будет добавить и нужно ли будет вообще что-то добавлять - заранее знать не можем) То есть общее количество становится 12, и эти вот две дополнительных задачи как-то должны сразу начать выполнение не дожидаясь окончания первоначально запущенных 10 задач.
И уже надо дожидаться выполнения всех 12-ти задач.
Также каждая из двух новых задач в процессе выполнения может также захотеть запустить M задач.

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

Возможно ли такое?
Есть ли возможность реализовать подобное с помощью той же sync.WaitGroup или необходимо уже подключать каналы?
Какие каналы использовать - с буфером или без? Если с буффером, то с каким, если неизвестно число задач.
Пробовал сделать канал с буфером большого размера. Отправил в него нужные данные.
В другой подпрограмме делаю range по каналу - запускаю задачи асинхронно, и внутри некоторых задач шлю в канал еще сообщение о новом задании, но как мне отследить в таком случае, что они уже все выполнены и не требуется выполнять новых заданий?

И может есть уже готовое решение, как реализовать подобное.
Жду подсказок.

Answer 1

[RidgeA]

Я бы сделал так.
1. Есть канал куда попадают все задачи.
2. Есть горутина, которая разгребает канал и на каждоее сообщение из канала запускает горутину для обработки сообщения.
3. При необходимости эта горутина ставит сообщения в тот же канал.
4. При запуске горутины-обработчика добавлять в wait group и не забыть 'defer wg.Done()'

Вот как-то так: https://play.golang.org/p/z0fS5O8ynSI
Не уверен, что это хороший код, но общую идею показть должен.
Размер буфера - то максимально возможное количество задач, которые могут прийти в качестве подзадач (с учетом вложенности, если у подзадачи могут быть свои подзадачи + 1 (на текущую задачу)

Comments

[Алексей]

Благодарю!
Но мне немного не так надо было. Я должен проанализировать ответ job(), прежде чем запустить задачу. Затем я проверю кое-что в базе данных. И только потом принимаю решение запустить задачу. А у вас подзадачи заранее готовы.

Я попробовал в воркере убрать цикл для отправления новых подзадач в канал
И в job передаю канал, по которому тот бы смог отправлять сообщения.

for t := range ch {
		go job(t, wg, ch)
	}

И перенес условие запуска новой задачи в конец job() уже когда он выполнит основную работу

// ... основная работа..
if t.taskId == 7 { // какое-то условие после выполнения job() сообщающее, что надо запустить еще задачу
		wg.Add(1)
		ch <- task{taskId:999}
		fmt.Printf("%v. start new task 999\n", t.taskId)
	}

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

Answer 2

[Sly_tom_cat .]

Вам похоже сюда: https://gobyexample.com/worker-pools

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

Новые job-ы лучше добавлять в теле самих job-ов, которые порождают новые.