Обойти неизвестной глубины дерево параллельно

[valchkou]

Имеется структура ввиде несбалансированного дерева типа файловой системы
Количество фолдеров и файлов заранее неизвестно.
Нужно обойти все элементы сверху вниз и обработать используя многопоточность.
Имеется некий синхронизированный SharedState к которому все потоки имеют доступ
(SharedState к тому же persistant но суть это не меняет)

Вопрос - как понять что обход дерева завершен ?

Интересует не имплементация а идея.
вариантов имплементаций море - весь арсенал жава многопоточности + akka.

в голову пока пришел такой вариант: иметь отдельный поток который валидирует этот SharedState.
предположить что каждый элемент не может обрабатываться дольше чем maxThreshold.

пример рекурсивного судокода чисто для пояснения идеи.

обработка

Code: Select all

@Async //this wraps each call with thread/worker
processAsync (element, state) {
     process(element)
     List children = element.getChildren()
     children.each{
         state.incrementSubmitted()
         processAsync(it, state)
     } 
     state.incrementCompleted()
}

критерий завершения

Code: Select all

    if (state.intervalSinceLatUpdate>maxThreshold && state.submitted == state.completed) {
         state.done()
     }

[valchkou]

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

[Medium-rare]

Если используем многопоточность, без ожидания не обойтись.

Atomic variables? Ноды должные нести атомарные переменные.
Заглянул в переменную, что нод уже потрогали, и отвалил, ничего не блокируя.

[M. Ridcully]

А о какой многопоточности речь? Если вопрос про Java - то как понимаю, речь про CPU, а не про GPU? То есть реальная польза от максимум 8-16-32 потоков, или вроде того?

Я бы не городил огород, а решал бы "в лоб":

Code: Select all

walkTree(node):
    processNodeInfo()
    for each child:
        if (numRunningThreads < limit):
            async walkTree(child)
        else:
            walkTree(child)

Доступ к shared state - обычный, рабоче-крестьянский locking.

Вдогонку - не совсем понял проблемы про то, чтобы определить, когда проход закончен. Очевидно, когда numRunningThreads == 0.

[valchkou]

Вдогонку - не совсем понял проблемы про то, чтобы определить, когда проход закончен. Очевидно, когда numRunningThreads == 0.

хотел как проще... хорошо сами напросились
numRunningThreads == 0 нельзя

количество работающих потоков фиксировано и ранятся они всегда (Flyweight pattern).
когда есть новый таск/нода то поток его подхватывает и выполняет

кроме этого запросов на обработку деревьев может прийти несколько и обработка должна происходить параллельно.
это значит что один и тот же инстанс потока будет обработает таски от разных деревьев.

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

[valchkou]

дуп

[M. Ridcully]

хотел как проще... хорошо сами напросились
numRunningThreads == 0 нельзя

количество работающих потоков фиксировано и ранятся они всегда (Flyweight pattern).

Тогда пусть будет numBusyThreads - ну то есть любая обработка (куска) дерева обёрнута в инкремент / декремент этой переменной.

кроме этого запросов на обработку деревьев может прийти несколько и обработка должна происходить параллельно.
это значит что один и тот же инстанс потока будет обработает таски от разных деревьев.

Значит, эта переменная не одна, а per-tree.

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

Так так и будет - отдельный thread раскидывает задачи, другие молотят деревья. То есть оригинальный псевдокод чуть меняется, чтобы дерево всегда начинать обрабатывать не в главном потоке, но суть не меняется. Чтобы понять, когда дерево обработано - заведите condition variable (опять per-tree, как и numBusyThreads), или чего там у вас в Java есть, чтобы мог сигналиться при декременте numBusyThreads. Отдельный thread мониторит эти condition variables и отмечает обработанные деревья.

[valchkou]

Так так и будет - отдельный thread раскидывает задачи, другие молотят деревья. То есть оригинальный псевдокод чуть меняется, чтобы дерево всегда начинать обрабатывать не в главном потоке, но суть не меняется. Чтобы понять, когда дерево обработано - заведите condition variable (опять per-tree, как и numBusyThreads), или чего там у вас в Java есть, чтобы мог сигналиться при декременте numBusyThreads. Отдельный thread мониторит эти condition variables и отмечает обработанные деревья.

да State для каждого дерева/запроса будет трекаться отдельно.
в жаве у нас есть все!
собственно это решения я и предложил в самом первом топике, надеюсь вы его не читали и пришли к данному решению самостоятельно
оно логично. Интересно было бы рассмотреть альтернативы

[M. Ridcully]

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

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

[M. Ridcully]

Добавлю.
У вас там async unconditional. Я не знаю, что это значит, но если вы полагаетесь на то, что thread pool будет ставить в очередь при достижении предела параллелизма, то мне это не нравится.
В моем подходе никаких очередей нет, будь то явных или неявных.

[valchkou]

Добавлю.
У вас там async unconditional. Я не знаю, что это значит, но если вы полагаетесь на то, что thread pool будет ставить в очередь при достижении предела параллелизма, то мне это не нравится.
В моем подходе никаких очередей нет, будь то явных или неявных.

в контектсте приведенного мною пример это всего лишь означает что операция асинхронна. Это судо код для наглядности концепции
там действительно опущены детали имплиментации.
в реальности количество потоков которые дозволено использовать будет ограничено конфигурацией,
потому иногда придется ждать. Например с помощью BlockingQueue либо АККА backoff

Я не стал приводить полную имплементацию чтобы не запутывать народ еще больше. Она слишком жава специфик.

[Palych]

А узлы дерева суть файлы?
И в чем цель параллельности? Если чисто вычислительные задачи (без сетевых и прочих удаленных вызовов) - может Streams использовать?

[valchkou]

А узлы дерева суть файлы?
И в чем цель параллельности? Если чисто вычислительные задачи (без сетевых и прочих удаленных вызовов) - может Streams использовать?

суть похожа но не файлы.
дерево физически в базе данных, а ноды достаются через rest api причем используя pagination, потому как количество элементов даже на одном уровне может быть огромно.
parallelStream прост но он преднозначен для уже полученной коллекции, а тут рекурсия по сути, то есть внутри стримов придется запустить еще стримы.
впринципе даже это не проблема, проблема в том чтобы количество потоков можно было переконфигурить на лету.
поэтому вместо Streams можно как вариант заюзать ForkJoin напрямую.