使用 Clojure 线程化长时间运行的进程并比较它们的回报 [英] with Clojure threading long running processes and comparing their returns

问题描述

我对需要处理的两个非常大的数据集有两个不同的函数，最终归结为两个布尔值.然后需要将这些值并在一起以获得最终结果.我的问题是创建线程以便两个长函数可以同时运行的最佳方法是什么.我的想法是这样的，

I have two different function on two very large sets of data that need to be processed, in the end coming down to two Boolean values. those values need to then be anded together for the final result. my question is what is the best way to create threads so the two long functions can run at the same time. my thoughts were something like,

(def f (future longProcessOne(data_one)))
(def g (future longProcessTwo(data_two)))
(and @f @g)

但我一直在寻找更好的方法来解决这个问题.

but I was looking for input on a better way to go about this.

推荐答案

(顶部基于 Promise 的方法，底部基于 core.async 的方法.两者都在第一个 falsey 值上短路.)

这是一个利用单个承诺可以多次交付这一事实的版本(尽管只有第一次交付会成功设置其值；随后的交付仅返回 nil 而没有副作用).

Here's a version making use of the fact that a single promise can be delivered multiple times (although only the first delivery will succeed in setting its value; subsequent deliveries simply return nil with no side effects).

(defn thread-and
  "Computes logical conjunction of return values of fs, each of which
  is called in a future. Short-circuits (cancelling the remaining
  futures) on first falsey value."
  [& fs]
  (let [done (promise)
        ret  (atom true)
        fps  (promise)]
    (deliver fps (doall (for [f fs]
                          (let [p (promise)]
                            [(future
                               (if-not (swap! ret #(and %1 %2) (f))
                                 (deliver done true))
                               (locking fps
                                 (deliver p true)
                                 (when (every? realized? (map peek @fps))
                                   (deliver done true))))
                             p]))))
    @done
    (doseq [[fut] @fps]
      (future-cancel fut))
    @ret))

一些测试:

(thread-and (constantly true) (constantly true))
;;= true

(thread-and (constantly true) (constantly false))
;;= false

(every? false?
        (repeatedly 100000
                    #(thread-and (constantly true) (constantly false))))
;;= true

;; prints :foo, but not :bar
(thread-and #(do (Thread/sleep 1000) (println :foo))
            #(do (Thread/sleep 3000) (println :bar)))

<小时>

将 Arthur 和 A. Webb 的想法放在一起，您可以使用 core.async 和结果，同时对返回的第一个 falsey 值进行短路:

---

Putting Arthur's and A. Webb's ideas together, you could use core.async to and the results together while short-circuiting on the first falsey value returned:

(defn thread-and
  "Call each of the fs on a separate thread. Return logical
  conjunction of the results. Short-circuit (and cancel the calls
  to remaining fs) on first falsey value returned."
  [& fs]
  (let [futs-and-cs
        (doall (for [f fs]
                 (let [c (chan)]
                   [(future (>!! c (f))) c])))]
    (loop [futs-and-cs futs-and-cs]
      (if (seq futs-and-cs)
        (let [[result c] (alts!! (map peek futs-and-cs))]
          (if result
            (recur (remove #(identical? (peek %) c)
                           futs-and-cs))
            (do (doseq [fut (map first futs-and-cs)]
                  (future-cancel fut))
                false)))
        true))))

使用(constantly false)和(constantly true)进行测试:

(thread-and (constantly true) (constantly true))
;= true
(thread-and (constantly true) (constantly false))
;= false

;;; etc.

另请注意，短路确实有效:

Also note that short-circuiting does indeed work:

;;; prints :foo before returning false
(thread-and #(do (Thread/sleep 3000) false)
            #(do (Thread/sleep 1000) (println :foo)))

;;; does not print :foo
(thread-and #(do (Thread/sleep 3000) false)
            #(do (Thread/sleep 7000) (println :foo)))

这篇关于使用 Clojure 线程化长时间运行的进程并比较它们的回报的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！