Wishful Coding

For the July Amsterdam Clojure meetup, a lot of people where curious what core.async is all about, so I proposed tackling the dining philosophers problem with core.async.

The problem is explained and a solution proposed in the CSP book section 2.5, but using events rather than channels.

We worked on the problem dojo style, switching the driver seat every few minutes. But with no one really knowing the library very well, progress was slow, and by the end of the meetup we could make one philosopher eat.

One problem we ran into was that go blocks are lexical, so you can’t easily write helper functions that use <!

(defn putdown [ch] (>! ch :fork))
(go (putdown (chan)))
Exception in thread "async-dispatch-1" java.lang.AssertionError: Assert failed: >! used not in (go ...) block

So this morning I sat down to make this thing work.

During the meetup we had a function that would do something silly to setup a vector with 5 forks represented by channels, which I replaced by some equally silly, until I just came up with this.

I use channels with a buffer of one and put a fork in that buffer. This makes sure a fork can but picked up and put down without blocking, but only once.

(defn set-table []
  (let [table [(chan 1) (chan 1) (chan 1) (chan 1) (chan 1)]]
    (doseq [ch table]
      (>!! ch :fork))
    table))

The butler making sure only 4 philosophers are seated is simply represented as

(chan 4)

This leads to the definition of the basic actions

(def pickup <!!)
(def putdown #(>!! % :fork))

(def sit-down #(>!! % :sit))
(def get-up <!!)

The simplified behaviour of a philosopher then becomes

(sit-down butler)
(alts!! my-forks)
(alts!! my-forks)
(println "eating")
(map putdown my-forks)
(get-up butler)
(println "thinking")
(recur)

And finally we can run 5 philos threads.

(doseq [p philosophers]
  (thread (philosopher table butler p)))

Be sure to check out the code and output and join the next meetup if you’re in the Netherlands.

I found a nice place to put the RCX and NXT and got together some remotes to play with it.

The video and slideshow might not work on the RSS feed or mailing list. Check the website.

I am reading about Clojure’s core.async and the CSP book it is based on. A few things that stood out from the announcing blog post and the book:

It should be noted that Clojure’s mechanisms for concurrent use of state remain viable, and channels are oriented towards the flow aspects of a system.

This law which permits a system defined in terms of concurrency to be given an alternative description without concurrency.

So I would say core.async is a library for reasoning about sequential control flow in concurrent programs. That’s a great idea.

A curious thing is that the first 3 chapters of the book discuss events and only in chapter 4 are channels introduced. Core.async just includes the channels.

Compare:

An event is an action without duration, whose occurrence may require simultaneous participation by more than one independently described process.

We shall observe the convention that channels are used for communication in only one direction and between only two processes.

So channels have one sender and one receiver, while events just happen if all parties involved agree.

Observe the following orchestra

(defn player [state]
  (assert (= :swing (<!! state)))
  (>!! state :note)
  (println "fuut")
  (recur state))

(defn conductor [state]
  (Thread/sleep 1000)
  (>!! state :swing)
  (assert (= :note (<!! state)))
  (recur state))

(def state (chan))
(thread (conductor state))
(thread (player state))
;> fuut
;> fuut
;> fuut
;> ...

I would really like to have a pattern on receiving, so that instead of assert you could just say “receive X” and you won’t get a Y you can’t deal with.

The book describes channels as an event of the pair c.v where c is the name of the channel and v is the value of the message. Like any event, the communication only happens when both ends agree on the event.

(defn player [state]
  (<!! state :swing)
  (>!! state :note)
  (println "fuut")
  (recur state))

(defn conductor [state]
  (Thread/sleep 1000)
  (>!! state :swing)
  (<!! state :note)
  (recur state))

(def state (chan))
(thread (conductor state))
(thread (player state))

But the bigger problem with this program is that it doesn’t work for more than one player. This could be solved with a broadcast channel, or by using events.

(defn player [state]
  (!! state :swing)
  (!! state :note)
  (println "fuut")
  (recur state))

(defn conductor [state]
  (Thread/sleep 1000)
  (!! state :swing)
  (!! state :note)
  (recur state))

(def state (events))
(thread (conductor state))
(thread (player state))
(thread (player state))
(thread (player state))

Maybe I misunderstood the book, or maybe this is just out of scope for core.async, but I think it’d be neat to have events and channels.

(defn conductor [swing]
  (loop []
    (Thread/sleep 1000)
    (.await swing)
    (recur)))

(defn player [swing sound]
  (loop []
    (.await swing)
    (println sound)
    (recur)))

(def swing (java.util.concurrent.CyclicBarrier. 4))
(future (conductor swing))
(future (player swing "fuut"))
(future (player swing "fwoop"))
(future (player swing "bang"))
;> fuut
;> fwoop
;> bang
;> ...