Отфильтровать ленивую последовательность в параллельном

Question

У меня есть проблема, в которой я ищу номера с определенными свойствами в очень большом пространстве поиска (возможно, бесконечно, но определенно слишком большой для того, чтобы все пространство соответствовало памяти). Поэтому мне нужна ленивая последовательность, которую я фильтрую. Мой наивный подход состоял в том, чтобы использовать список для определения всего (for) для всего поиска, но это выполняет поиск в одном потоке. Есть несколько действительно простых способов обрезать пространство поиска, и есть также некоторые части поиска, которые являются более интенсивными с точки зрения вычислительной мощности.

Мой немного менее наивный подход заключался в том, чтобы легко обрезать выражение for и сделать функцию search, которая сделала более трудную работу. Затем: (filter search (for [..... :when (prune)])). В библиотеке редукторов есть функция filter, но that won't work on lazy seqs. Я не могу преобразовать из ленивого seq из-за ограничения памяти.

Итак, что является хорошим способом фильтрации ленивой последовательности параллельно? Мой последний наивный подход будет что-то вроде наклеивания последовательности в атоме:

(defn accessor-gen [lazys] 
    (let [s (atom [nil lazys])] 
    (fn [] 
     (first (swap! s (fn [[_ s]] [(first s) (rest s)])))))) 

Тогда я мог бы иметь пул потоков в шести или так, используя эту функцию для поиска места.

Вопрос: У меня возникло непростое чувство, что я делаю это сложнее, чем нужно. Кроме того, меня беспокоит утверждение об атоме. Есть ли более простой метод одновременного использования ленивой последовательности? Наконец, весь мой подход в корне ошибочен? Есть ли лучший способ, возможно, тот, который не требует ленивых последовательностей?

Answer 1

Первое, что я хотел бы попробовать, чтобы фильтровать pmapped SEQ:

(defn search [i] 
     (println (Thread/currentThread) i) 
     (when (zero? (rem i 10)) 
      i)) 

(take 10 (filter identity (pmap search (range)))) 

Фильтрация будет происходить в одном потоке, но поиск будет вычисляться параллельно.

Если то, что вы на самом деле хотите сделать параллельно является фильтрацией, вам нужно будет разделить ленивый и SEQ CONCAT результатов:

(defn search [numbers] 
     (doall (filter (fn [i] (zero? (rem i 10))) numbers))) 

(take 10 (apply concat (pmap search (partition-all 1000 (range)))))