Какой метод параллелизма clojure использовать при поиске растущего пространства решений?

Question

Как правильно в Clojure выполнять параллельную обработку, когда каждое задание обработки может выполняться в полной изоляции и может генерировать список дополнительных заданий, которые необходимо оценить?

Моя настоящая проблема - это проблема расчета питательности, но я обозначу ее в форме шахмат, которая имеет те же черты пробела, что и мои вычисления.

Предположим, например, что я пытаюсь найти все ходы для шаха в игре в шахматы. При поиске по состояниям доски я начинал с 20 возможных состояний, каждое из которых представляло различный возможный ход открытия. Каждое из них необходимо будет оценить, принять или отклонить, а затем для каждого принятого хода будет создан новый список заданий, представляющий все возможные последующие шаги. Работа будет выглядеть так:

initial: '([] proposed-move)
accepted: '([move] proposed-response)
          '([move move] proposed-response)

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

Решение, с которым я играю, выглядит так:

; a list of all final solutions, each of which is a sequence of moves
(def solutions (agent []))
; a list of all jobs pending evaluation
(def jobs (agent []))

С учетом этих определений у меня будет пул потоков Java, и каждый поток будет запрашивать задание у агента заданий (и ожидать выполнения этого запроса). Затем он запустит расчет, сгенерирует список решений и возможных решений. Наконец, он отправляет решения агенту решений, а возможные решения агенту заданий.

Является ли использование комбинации агентов и потоков наиболее идиоматичным способом в этом случае? Могу ли я даже получить данные из очереди заданий так, как я предлагаю?

Или мои задания должны быть java.util.concurrent.LinkedBlockingQueue, как описано в Производитель-потребитель с квалификацией ?

Answer 1

Вы можете сделать это следующим способом:

• Повторные применения pmap (что обеспечивает параллельную обработку всех элементов в коллекции)
• Функция, используемая в pmap, возвращает список элементов. Может быть ноль, один или несколько элементов, которые затем будут обработаны в следующей итерации
• Результаты объединяются с concat
• Вы повторяете обработку списка столько раз, сколько хотите, возможно, сохраняя результат в атоме.

Пример кода может быть что-то вроде следующего

(def jobs (atom '(1 10 100)))

(defn process-element [value]
  (if (< (rand) 0.8)
    [(inc value)]
    []))

(defn do-processing []
  (swap! jobs 
         (fn [job-list] (apply concat (pmap process-element job-list)))))

(while (seq @jobs)
  (prn @jobs)
  (do-processing))

Whick может производить вывод, например:

(1 10 100)
(2 11 101)
(3 12 102)
(4 13 103)
(5 14 104)
(6 15 105)
(7 106)
(107)
(108)
(109)
nil

Обратите внимание, что вы должны быть немного осторожны, чтобы убедиться, что ваш алгоритм завершается! В этом примере это гарантируется тем, что элементы со временем исчезают, но если ваше пространство поиска увеличивается, вы, вероятно, захотите применить ограничение по времени, а не просто использовать (while ...) цикл .

Answer 2

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

единственное, что я хотел бы изменить, чтобы сделать его более "похожим на clojure", это использовать pmap для перебора очереди, хранящейся в агенте. использование pmap вместо собственного пула потоков избавит вас от необходимости управлять пулом потоков, поскольку pmap уже использует пул потоков clojure, который правильно инициализирован для текущего числа процессоров. это также поможет вам воспользоваться последовательностью фрагментов (что может помочь).

Answer 3

Вы также можете использовать каналы. Может быть, что-то вроде этого:

(def jobs (chan))
(def solutions (chan))
(def accepted-solutions (atom (vector)))

(go (loop [job (<! jobs)]
      (when job
        (go (doseq [solution (process-job-into-solutions job)]
              (>! solutions)))
        (recur (<! jobs)))))

(go (loop [solution (<! solutions)]
      (when (acceptable? solution)
        (swap! accepted-solutions conj solution)
        (doseq [new-job (generate-new-jobs solution)]
          (>! jobs))
        (recur (<! solutions)))))

(>!! jobs initial-job)