How to implement the Skynet 1m microbenchmark with core.async?

Question

In order to try to understand core.async, I unsuccesfully tried to implement the "Skynet 1 million microbenchmark", which is:

Creates an actor (goroutine, whatever), which spawns 10 new actors, each of them spawns 10 more actors, etc. until one million actors are created on the final level. Then, each of them returns back its ordinal number (from 0 to 999999), which are summed on the previous level and sent back upstream, until reaching the root actor. (The answer should be 499999500000).

There are implementation in many languages here:

https://github.com/atemerev/skynet

Here's my totally broken attempt:

(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))))))))

And I was trying to call it all from inside a go block at the REPL:

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

I'm probably seriously confused about many things concerning core.async (and lazy evaluation too).

How should I go about solving this problem and why?

Answer 1

There are some limitations on what core.async is able to do, so you cannot use the map or for functions.

Your implementation is pretty close to the correct one. Some points:

1. go == one process, so you are just creating one process, not 1m
2. <!! is to be used outside go block
3. <! is to be used inside go blocks
4. You are using for incorrectly
5. doall accepts just one parameter

A working implementation that probably can be improved:

(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)))))))))

And to call it:

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