如何使用core.async实现Skynet 1m微基准测试？

Question

为了尝试理解core.async,我不成功地试图实现"天网100万微基准",即:

创造一个演员(goroutine,无论如何),它产生10个新演员,每个演员产生10个演员等,直到在最后一级创建了100万个演员.然后,它们中的每一个都返回它的序数(从0到999999),它们在前一级别上求和并向上游发送,直到到达根演员.(答案应该是499999500000).

这里有许多语言的实现:

https://github.com/atemerev/skynet

这是我完全破碎的尝试:

(defn skynet [chan num size div]
  (if (= 1 size)
    (>! chan num)
    (>! chan (reduce + (let [rc  (async/chan)
                             n   (/ size div)]
                          (doall (for [i [0 div]]
                                   (skynet rc (+ num (* i n)) n div))
                                 (for [i [0 div]] (<! rc))))))))

我试图在REPL的一个go块内调用它:

  (time (go (<!! (skynet (async/chan) 0 1000000 10))))

我可能会对很多关于core.async(以及懒惰评估)的事情感到困惑.

我该如何解决这个问题？为什么？

Answer 1

core.async能够做什么有一些限制,所以你不能使用map或for函数.

您的实现非常接近正确的实现.一些要点:

1. go ==一个进程,所以你只是创建一个进程,而不是1米
2. <!! 是在外面使用块
3. <! 将被用于内部块
4. 您正在使用的不正确
5. doall 只接受一个参数

可能可以改进的工作实现:

(defn skynet [parent num size div]
  (go ;; We create a new process each time skynet is called
    (if (= 1 size)
      (>! parent num)
      (let [self (chan)
            new-size (/ size div)]
        (dotimes [i div] ;; dotimes is more explicit for side effects 
          (skynet self (+ num (* i new-size)) new-size div))
    (loop [i div ;; Manual reduce 
           t   0]
      (if (zero? i)
        (>! parent t)
        (recur (dec i)
               (+ t (<! self)))))))))

并称之为:

 (time
   (do
     (def result (chan))
     (def x (skynet result 0 1000000 10))
     (<!! result)))