Как присоединиться к Kotlin SupervisorJob
Я пытаюсь обработать дерево объектов данных. Предполагается, что каждый лист дерева обрабатывается функцией с использованием сопрограммы. Весь процесс должен выполняться с использованием катушки фиксированного размера.
Итак, я придумал это:
val node = an instance of WorkspaceEntry (tree structure)
val localDispatcher = newFixedThreadPoolContext(16)
fun main() {
val job = SupervisorJob()
val scope = CoroutineScope(localDispatcher + job)
handleEntry(node, scope)
runBlocking {
job.join()
}
}
Метод handleEntry рекурсивно запускает дочернее задание в супервизоре для каждого листа дерева.
Все дочерние задания супервизора завершаются успешно, но соединение никогда не возвращается. Я неправильно это понимаю?
Обновлено: функция HandleEntry
private fun handleEntry(workspaceEntry: WorkspaceEntry, scope: CoroutineScope) {
if (workspaceEntry is FileEntry) {
scope.launch {
FileTypeRegistry.processFile(workspaceEntry.fileBlob)
}
} else {
workspaceEntry.children.forEach { child -> handleEntry(child, scope) }
}
}
Под «соединением никогда не возвращается» вы имеете в виду, что поток заблокирован, а приложение зависло? Функция main не завершается?
да. Он будет бесконечно ждать завершения задания. Я проверил: дочерние задания завершаются и уничтожаются, пока у супервизора не останется детей. Но сама работа никогда не переходит в завершенное состояние.
Похоже, что если вы cancel, то SupervisorJob отменяет все свои дочерние элементы, а вы может ждете его и его дочернего завершения, затем с помощью join. Но, конечно, вы получаете, возможно, нежелательный CancellationException в каждом дочернем задании, что приводит к остановке выполнения, если оно вызывает некоторую функцию suspend.
Ответы 2
Кажется, что Job, который используется для создания CoroutineContext (в вашем случае SupervisorJob), не предназначен для ожидания завершения дочерних сопрограмм, поэтому вы не можете использовать job.join(). Я предполагаю, что основная цель этого Job - отменить дочерние сопрограммы. Замена блока runBlocking на следующий будет работать:
runBlocking {
job.children.forEach {
it.join()
}
}
Хотя это сработает, я почти уверен, что это не предполагаемое решение. В документации по объединению явно указано, что дочерние задания тоже должны быть завершены. Кроме того, поскольку самому супервизору не назначено задание, он должен переключиться в состояние завершения сразу после запуска задания. Запуск задания происходит как при запуске дочерних заданий, так и при вызове join (). В какой-то момент во время разработки существовала функция с именем joinChildren (). Но теперь его нет.
Но да, раз уж я, наверное, что-то совсем не понял, придется так и поступить
Вы смешали две роли:
- главная работа, найденная в области сопрограммы, которая никогда не завершается сама по себе и используется для управления жизненным циклом всего остального
- задание, соответствующее единице работы, возможно, разложенное на несколько дочерних заданий
Вам нужно и то, и другое, например:
val masterJob = SupervisorJob()
val scope = CoroutineScope(localDispatcher + masterJob)
val unitOfWork = scope.launch { handleEntry(node, scope) }
runBlocking { unitOfWork.join() }
Приведенный выше код на самом деле не мотивирует существование главного задания, потому что вы начинаете с него только одно дочернее задание, но он может иметь смысл в более широкой картине, где у вас есть некоторый контекст, из которого вы запускаете много заданий, и хотите быть умеет писать
masterJob.cancel()
чтобы отменить все до того, как это будет сделано.
на самом деле я использую задание супервизора, потому что я запускаю из него несколько дочерних заданий (handleEntry рекурсивно запускает больше заданий), и я хочу, чтобы они могли сбой независимо, не прерывая весь процесс.
Но решение не работает. Вместо блокировки на неопределенный срок главное задание ожидает завершения handleEntry () и мгновенно переключается в состояние завершения: возвращается join (). Однако дополнительные задания, запущенные в handleEntry (с использованием параметра scope), все еще выполняются.
Вы говорите о моем коде или о вашем? masterJob не ждет завершения чего-либо, поэтому он никогда не меняет состояние. unitOfWork распространяет сбой на другие сопрограммы, это правда. Таким образом, ваша ситуация на самом деле нестандартная, и поэтому подход с masterJob.children.forEach { it.join() } является правильным.
Хорошо, но тогда какой смысл добавлять masterJob в прицел?
masterJob - это SupervisorJob, поэтому он не распространяет отказ ваших дочерних сопрограмм. Он также является сборщиком всех сопрограмм, запущенных в рамках своей области, так что вы можете join их все.
Итак, я должен иметь ссылку на masterJob, чтобы отменить и, присоединить все его дочерние элементы? Я не могу сделать это со ссылкой на scope, хотя я могу отменить cancel, объем и все дочерние элементы задания?
Вы этого не сделаете. Работа есть на scope.coroutineContext[Job]!!.
добавьте пожалуйста код для функции
handleEntry.