如何在 Clojure 中实现并行逻辑或提前终止

Question

我想定义一个谓词，将一些谓词作为输入 具有相应的输入（它们可以作为惰性调用序列给出）， 并行运行它们并计算结果的逻辑或， 这样，当谓词调用终止返回true时， 整个计算也终止（返回true）。

除了提供时间优化，这也有助于避免 在某些情况下不终止（某些谓词调用可能不会终止）。 实际上，将不终止解释为第三个undefined 值， 此谓词模拟 Kleene's K3 logic 中的 or 操作 （初始居中Kleene algebra中的join）。

here 为 Haskell 家族提供了类似的东西。 在 Clojure 中是否有任何（最好是简单的）方法可以做到这一点？

编辑：我决定在阅读 cmets 后添加一些说明。

(a) 首先，线程池耗尽后发生的事情并不重要。我认为创建一个足够大的线程池来满足我们的需求是一个合理的约定。

(b) 最关键的要求是谓词调用开始并行运行，一旦谓词调用终止返回true，所有其他运行的线程都会中断。预期的行为是：

• 如果有谓词调用返回true：并行或返回true
• else 如果存在未终止的谓词调用：并行或未终止
• else：并行或返回false

换句话说，它的行为类似于falseundefinedtrue 给出的三元素格中的连接，undefined 表示非终止。

(c) 并行的 or 应该能够将许多谓词和许多谓词输入（每个都对应一个谓词）作为输入。但如果将惰性序列作为输入会更好。然后，命名并行或pany（用于“并行任何”），我们可以进行如下调用：

• (pany (map (comp eval list) predicates inputs))
• (pany (map (comp eval list) predicates (repeat input)))
• (pany (map (comp eval list) (repeat predicate) inputs)) 相当于 (pany (map predicate (unchunk inputs)))

作为最后的评论，我认为要求 pany、双重 pall 或用于构建此类早期终止并行缩减以易于实施甚至内置的机制是很自然的Clojure 等面向并行的语言。

Answer 1

我将根据归约函数来定义我们的谓词。实际上，我们可以重新实现所有 Clojure 迭代函数来支持这种并行操作，但我将仅使用 reduce 作为示例。

我将定义一个计算函数。我会用同一个，但没有什么能阻止你拥有很多。如果累积 1000，则该函数为“真”。

(defn computor [acc val]
        (let [new (+' acc val)] (if (> new 1000) (reduced new) new)))

(reduce computor 0 (range))
;; =>
1035

(reduce computor 0 (range Long/MIN_VALUE 0))
;; =>
;; ...this is a proxy for a non-returning computation

;; wrap these up in a form suitable for application of reduction
(def predicates [[computor 0 (range)] 
                 [computor 0 (range Long/MIN_VALUE 0)]])

现在让我们开始讨论这个问题。我想在每次计算中迈出一步，如果其中一个计算完成，我想返回它。实际上，一次一步使用 pmap 非常慢 - 工作单元太小，不值得线程化。在这里，在继续之前，我已经更改了每个工作单元的 1000 次迭代。您可能会根据您的工作量和步骤成本来调整它。

(defn p-or-reducer* [reductions]
        (let [splits (map #(split-at 1000 %) reductions) ;; do at least 1000 iterations per cycle
              complete (some #(if (empty? (second %)) (last (first %))) splits)]
          (or complete (recur (map second splits)))))

然后我将其包装在驱动程序中。

(defn p-or [s]
  (p-or-reducer* (map #(apply reductions %) s)))

(p-or predicates)
;; =>
1035

在哪里插入 CPU 并行度？ p-or-reducer* 中的 s/map/pmap/ 应该这样做。我建议只并行化第一个操作，因为这将驱动归约序列进行计算。

(defn p-or-reducer* [reductions]
        (let [splits (pmap #(split-at 1000 %) reductions) ;; do at least 1000 iterations per cycle
              complete (some #(if (empty? (second %)) (last (first %))) splits)]
          (or complete (recur (map second splits)))))

(def parallelism-tester (conj (vec (repeat 40000 [computor 0 (range Long/MIN_VALUE 0)]))
                             [computor 0 (range)]))

(p-or parallelism-tester) ;; terminates even though the first 40K predicates will not

很难定义一个高性能的通用版本。在不知道每次迭代的成本的情况下，很难得出有效的并行策略——如果一次迭代需要 10 秒，那么我们可能一次只走一步。如果需要 100ns，那么我们需要一次采取许多步骤。

Answer 2

您是否会考虑采用core.async 处理与async/go 或async/thread 的并行任务，并与async/alts! 提前返回？

例如，将核心or 函数从串行变为并行。我们可以创建一个宏（我称之为por）将输入函数（或谓词）包装到async/thread 中，然后在它们之上执行套接字选择async/alts!：

(defmacro por [& fns]
  `(let [[v# c#] (async/alts!!
                  [~@(for [f fns]
                       (list `async/thread f))])]
     v#))

(time
 (por (do (println "running a") (Thread/sleep 30) :a)
      (do (println "running b") (Thread/sleep 20) :b)
      (do (println "running c") (Thread/sleep 10) :c)))
;; running a
;; running b
;; running c
;; "Elapsed time: 11.919169 msecs"
;; => :c

与原来的or（串行运行）相比：

(time
 (or (do (println "running a") (Thread/sleep 30) :a)
     (do (println "running b") (Thread/sleep 20) :b)
     (do (println "running c") (Thread/sleep 10) :c)))
;; running a
;; => :a
;; "Elapsed time: 31.642506 msecs"