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?

Assuan

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

Kial

Thanks a lot! A lot of Clojure to study but this shall definitely help me a lot, not only your version but also your comments as to what is wrong in my code. Weirdly some "a/" appeared in your last line of code, I edited it :) – Assuan

Of course slower than just using a reducer, but that's the point of the benchmark. I tried to write a version of the skynet function that would return a channel containing the result (instead of taking one as argument), and use async/reduce instead of a manual reduce and could not find a way to do it. I would be interested in such a version. Anyway, Kudos ! – Bethsaida