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

[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 напрямую.

[Palych]

проблема в том чтобы количество потоков можно было переконфигурить на лету.

А зачем? Чтобы не перегружать удалённую систему?

[valchkou]

проблема в том чтобы количество потоков можно было переконфигурить на лету.

А зачем? Чтобы не перегружать удалённую систему?

да, как удаленную так и локальную.

[Ann4Ann]

я сейчас очень похожую задачу пишу, только у меня не дерево, а граф, так что стейт сложнее. вначале один-в-один как у автора написала, но очень уж оно не сбалансровано все. в итоге по времени и по перформансу мне легче вышло вначале сделать легкий траверс графа (в лоб) собирая инфу для heavy lifting в какой-то мнее.. execution plan. а потом этот план уже параллельно запустить. так получилось больше контроля за балансом нагрузки на потоки. по времени выигрыш ~ в 4 раза в моем случае. но, понятно, что не всегда можно легкий траверс сделать, там как раз дьявол в деталях.

[valchkou]

автор еще ничего не написал,
если я пойду первым путем то работа будет распределяться и распараллеливаться через
ArrayBlockingQueue<Node> queue;

с одной стороны я буду доставать и добавлять в эту очередь ноды
queue.put(node);

с другой же будет фиксированное количество потоков читающих из очереди
Node n = queue.take();
process(n)

что то типа этого

Code: Select all

package com.valchkou;

import javax.annotation.PostConstruct;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TaskPubSub {

    private int queueSize = 100;
    private int workerSize = 4;

    private ArrayBlockingQueue<Task> queue;
    private ExecutorService taskExecutor;

    public TaskPubSub () {
        queue = new ArrayBlockingQueue<Task>(queueSize);
        taskExecutor = Executors.newFixedThreadPool(workerSize);
        subscribe();
    }


    /**
     * Submit Task for processing or block if queue is full.
     */
    public void submit(Task t) {
    	queue.put(t);
    }


    /**
     * initialize working threads to process the queued items
    */
    private void subscribe() {
        for (int i=0; i< workerSize; i++) {
            taskExecutor.submit(() -> {
                 while (true) {
                         Task t = queue.take();
                          process(t);
                 }
            });
        }
    }

}

[ALV00]

я сейчас очень похожую задачу пишу, только у меня не дерево, а граф, так что стейт сложнее. вначале один-в-один как у автора написала, но очень уж оно не сбалансровано все. в итоге по времени и по перформансу мне легче вышло вначале сделать легкий траверс графа (в лоб) собирая инфу для heavy lifting в какой-то мнее.. execution plan.

Интересно, как этот план запоминается? Тоже дерево?

[Palych]

я сейчас очень похожую задачу пишу, только у меня не дерево, а граф, так что стейт сложнее. вначале один-в-один как у автора написала, но очень уж оно не сбалансровано все. в итоге по времени и по перформансу мне легче вышло вначале сделать легкий траверс графа (в лоб) собирая инфу для heavy lifting в какой-то мнее.. execution plan.

Интересно, как этот план запоминается? Тоже дерево?

По идее, если между обработкой листьев дерева нет зависимостей - план окажется простой: линейный список, или очередь...

[ALV00]

я сейчас очень похожую задачу пишу, только у меня не дерево, а граф, так что стейт сложнее. вначале один-в-один как у автора написала, но очень уж оно не сбалансровано все. в итоге по времени и по перформансу мне легче вышло вначале сделать легкий траверс графа (в лоб) собирая инфу для heavy lifting в какой-то мнее.. execution plan.

Интересно, как этот план запоминается? Тоже дерево?

По идее, если между обработкой листьев дерева нет зависимостей - план окажется простой: линейный список, или очередь...

Что-то не осилил. Как из дерева получился список? В базах данных план выполнения это дерево.

[Palych]

я сейчас очень похожую задачу пишу, только у меня не дерево, а граф, так что стейт сложнее. вначале один-в-один как у автора написала, но очень уж оно не сбалансровано все. в итоге по времени и по перформансу мне легче вышло вначале сделать легкий траверс графа (в лоб) собирая инфу для heavy lifting в какой-то мнее.. execution plan.

Интересно, как этот план запоминается? Тоже дерево?

По идее, если между обработкой листьев дерева нет зависимостей - план окажется простой: линейный список, или очередь...

Что-то не осилил. Как из дерева получился список? В базах данных план выполнения это дерево.

Допустим дерево - это файлы в под-директориях. Нужно каждый переименовать.
Тогда построение плана: нахождение каждого файла и отправка его во очередь на переименование.

[DropAndDrag]

странно ... вроде бы написал здесь, а ничего нет

[Boriskin]

Напоминает задачу определения порядка компиляции множественных проектов с зависимостями друг на друга.