Переписывание последовательного кода для параллельного выполнения

Question

Что такое лучший идиоматический подход для перезаписи обычного lisp-последовательного кода для параллельного выполнения?

Существует несколько хороших библиотек, таких как lparallel, помогая простым случаям. Например, если у нас был какой-то каркас в длинном списке, мы можем заменить его на lparallel: mapcar, и он будет выполнять эту работу в большинстве случаев. Теперь у меня есть некоторый вызов цикла, принимая результат некоторого удаленного JSON API и nconcing его в списке:

(loop :for offset :from 0 :by 100 
      :for result = (get-remote-data offset) 
      :until (null result) :nconc result) 

Как заменить его, чтобы позвонить get-remote-data параллельно без нужды менять get-remote-data сам? Существуют ли какие-либо стандартные и идиоматические способы? Любое хорошее чтение на эту тему также поможет. Благодарю.

Answer 1

Я использовал chanl для такого использования, чтобы настроить очередь сообщений. I начал n рабочих потоков, которые сделали удаленные вызовы и отправили их в очередь. Агрегатор получил результаты из очереди, и объединил их.

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

EDIT: Чтобы получить неизвестное количество страниц, вы можете использовать атомный счетчик и атомный флаг. Рабочие потоки (из фиксированного пула ) затем проверяют флаг, получают следующее смещение от счетчика, делают удаленный вызов, наконец, либо отправляем результат в очередь, либо , если результат пуст, отбросьте флаг , Если флаг перевернут , любой рабочий поток проверяет, что он выключается. Когда пул потоков пуст, вы закончите.

Answer 2

я придумал следующее, используя lparallel:

(defun get-datum (n) 
    (sleep (random 2)) 
    (if (> n 1000) 
     () 
     (list n))) 

(defun get-data() 
    (let ((channel (lparallel:make-channel)) 
     results) 
    (flet ((collect (item) 
      (setq results (append item results)))) 

     ;; Ask lparallel to schedule the first 8 requests 
     (loop for i from 0 to 70 by 10 
      do (lparallel:submit-task channel #'get-datum i)) 

     ;; Schedule an additional request each time one returns 
     ;; until we get a null result 
     (loop for i from 80 by 10 
      for result = (lparallel:receive-result channel) 
      while result 
      do (lparallel:submit-task channel #'get-datum i) 
      (collect result)) 

     ;; wait for all outstanding requests 
     (loop repeat 7 
      do (collect (lparallel:receive-result channel))) 

     results)))