Clojure core.async 用于数据计算答案

【问题标题】：Clojure core.async for data computationClojure core.async 用于数据计算
【发布时间】：2015-10-19 00:35:26
【问题描述】：

我已经开始使用 clojure core.async 库。我发现 CSP、channels、go 块的概念真的很容易使用。但是，我不确定我是否正确使用它们。我有以下代码-

(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 块，我从通道中获取值并将它们存储到符号中。他们都在使用阻塞 <!! 函数，因为它们在主线程上。

我要做的基本上是通过将程序片段拆分为异步进程来加快计算时间。这是正确的做法吗？

【问题讨论】：

为什么要在这里使用异步代码？当您必须阻塞和等待时，异步代码是有效的。否则它什么也不做。当您只需要进行数学运算时，您应该直接或并行计算它，例如将来。
谢谢。在我的代码中，我同时开始 4 个计算并阻止/等待这些值，直到它们返回一些东西。这与异步阻塞/等待有何不同？但是，就像尼古拉斯和你指出的那样，我将为此使用未来。我不能在 go 块中使用 core.reducers 吗？或者这是一个可怕的想法？
您不会同时启动它们。一切都是按顺序执行的，因为您在 go 块中没有有价值的阻塞操作。我定义了函数“素数？”测试数字 i 是否为素数。看这里："(time (do (count (filter true? (map prime? (range 2 20000)))))" 给出 3 秒和 "(time (count (filter true? (map prime? (range 2 20000) )))))" 给出 3 秒。你只使用一个线程，这里的一切都是顺序的。测试一下！
谢谢。已经将我的代码切换为使用未来。

标签： performance concurrency clojure core.async

【解决方案1】：

对于这种处理方式，future 可能稍微合适一些。

链接中的例子很容易掌握：

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

处理，future块立即启动，所以上面确实启动了一个线程，等待10s，完成后打印“done”。

当您需要该值时，您可以使用：

(deref f)
; or @f

这将阻塞并返回未来代码块的值。

在同一个示例中，如果您在 10 秒过去之前调用 deref，则调用将阻塞，直到计算完成。

在您的示例中，由于您只是在等待计算完成，并且不太关心频道参与者之间的消息和交互未来是我推荐的。所以：

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

【讨论】：

谢谢尼古拉斯。我会试试看。不过，如果我不得不使用 core.async，你对我的实现有什么看法？我正在尝试掌握 CSP。

【解决方案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
  )

【讨论】：

【解决方案3】：

如果您希望通过并行运行代码来更普遍地加速您的程序，那么您可以考虑使用 Clojure 的 Reducers 或 Aphyr 的 Tesser。这些工作通过将单个计算的工作分成可并行化的部分，然后将它们组合在一起。这些将有效地在与您的计算机一样多的内核上运行工作。如果您使用 future 或在 go 块中运行每个计算，那么每个计算将在单个线程上运行，有些可能会在其他线程之前完成，并且这些核心将处于空闲状态。

【讨论】：