Clojure core.async用于数据计算

Question

我已经开始使用clojure core.async库了.我发现CSP,channel,go blocks的概念非常容易使用.但是,我不确定我是否正确使用它们.我有以下代码 -

(def x-ch (chan))
(def y-ch (chan))
(def w1-ch (chan))
(def w2-ch (chan))

; they all return matrices
(go (>! x-ch (Mat/* x (map #(/ 1.0 %) (max-fold x)))))
(go (>! y-ch (Mat/* y (map #(/ 1.0 %) (max-fold y)))))
(go (>! w1-ch (gen-matrix 200 300)))
(go (>! w2-ch (gen-matrix 300 100)))

(let [x1 (<!! (go (<! x-ch)))
        y1 (<!! (go (<! y-ch)))
        w1 (<!! (go (<! w1-ch)))
        w2 (<!! (go (<! w2-ch)))]

    ;; do stuff w/ x1 y1 w1 w2
)

我在符号有预定义的(矩阵)载体x和y.在使用之前我需要修改两个向量.那些向量非常大.我还需要生成两个随机矩阵.由于go宏以异步方式启动计算,因此我将所有四个计算任务拆分为单独的go块,并将后续结果放入通道中.然后我有一个let块,我从通道中获取值并将它们存储到符号中.它们都使用阻塞<!!函数,因为它们位于主线程上.

我正在尝试做的基本上是通过将程序片段分成异步进程来加快我的计算时间.这是正确的方法吗？

Answer 1

对于这种处理,未来可能稍微充分.

链接中的示例很容易掌握:

 (def f 
   (future 
     (Thread/sleep 10000) 
     (println "done") 
     100))

处理,将来的块立即启动,所以上面确实启动一个线程,等待10秒并在完成时打印"完成".

当您需要该值时,您可以使用:

(deref f)
; or @f

这将阻止并返回未来代码块的值.

在同一个示例中,如果在10秒之前调用deref,则调用将阻塞,直到计算完成.

在你的榜样,因为你只是等待计算完成,而不是对渠道参与者之间的信息和交互这么关心未来是什么,我会推荐.所以:

 (future 
    (Mat/* x (map #(/ 1.0 %) (max-fold x))))

Answer 2

go块返回包含表达式结果的通道，因此您无需为其结果创建中间通道。下面的代码可让您同时启动所有 4 个计算，然后阻止这些值直到它们返回。如果您不需要立即使用某些结果，则可以仅在实际使用该值时才阻止该值。

(let [x1-ch (go (Mat/* x (map #(/ 1.0 %) (max-fold x))))
      y1-ch (go (Mat/* y (map #(/ 1.0 %) (max-fold y))))
      w1-ch (go (gen-matrix 200 300))
      w2-ch (go (gen-matrix 300 100))
      x1 (<!! x1-ch)
      y1 (<!! y1-ch)
      w1 (<!! w1-ch)
      w2 (<!! w2-ch)]
  ;; do stuff w/ x1 y1 w1 w2
  )