core.async Buffering Behavior

For this write up I am considering buffering to be the difference between the input to a process and the output of a process, at the point where the process stops accepting new input. This definition is on the rough and ready side, which should prepare you for the nature of things to come.

Generally, any complicated system will act like a buffer when put between two queues, because of batching behaviors, copying behaviors etc. So as part of understanding how to compose systems with each other you need to understand buffers.

More concretely a situation may arise where you might have a buffer of size five in front of your logic to persist data to disk. That buffer of size five is critical to understanding the risk of data loss in your system.

Communicating in core.async happens over channels. Channels are FIFO queues. When you create a channel you have the ability to specify a buffer size, for example: (async/chan 1). This explicit buffer is a useful way to determine when the back pressure of a full queue should kick in.

We can write some code that measures the buffering of a channel:

(defn buffer-size [c]
  (let [accepted? (atom false)]
    (loop [i 0]
      (reset! accepted? false)
      (async/put! c (rand) (fn [& _] (reset! accepted? true)))
      (Thread/sleep 500) ;; this is a race
      (if @accepted?
        (recur (inc i))
        i))))

If we run our buffer-size function over a bunch of channels we can see it guess some buffer sizes:

(dotimes [i 5]
  (println i " " (buffer-size (async/chan i))))

All very straight forward, no surprises, the buffering behavior we get is exactly what we ask for. That would be the end of it, but buffering arises elsewhere too.

Channels are provided for communication, what communicates over these channels are these machines, or processes, or threads, or whatever you want to call them. I will call them processes. Most of these processes will be reading from a channel and writing to a channel, and will often end up acting as a buffer between these channels.

A very simple process is just copy:

;; this is almost the same implementation wise as 
;; core.async's builtin pipe
(defn copy [in out]
  (async/go-loop []
    (let [x (async/<! in)]
      (if (nil? x)
        (async/close! out)
        (do
          (async/>! out x)
          (recur))))))

We can use buffer-size to examine the buffering behavior of this process.

(dotimes [i 2]
  (dotimes [ii 2]
    (let [in (async/chan i)
          out (async/chan ii)]
      (copy in out)
      (println i ii (buffer-size in)))))

We can model this buffering with x + y + 1 = z where x is the input buffer size, y is the output buffer size, z is the buffer-size measurement, and the extra one added is the buffering of the value from the channel in the local variable x in the copy process. With the current protocol for channels, I don't think it is possible to implement a copying process without introducing extra buffering for at least one datum. You can't get rid of the plus one.

A slightly more complicated process might be one that outputs the sum of every two inputs:

(defn sum [in out]
  (async/go-loop []
    (let [a (async/<! in)
          b (async/<! in)]
      (do
        (async/>! out (+ a b))
        (recur)))))

Let's take a look at its buffer table:

(dotimes [i 3]
  (dotimes [ii 3]
    (let [in (async/chan i)
          out (async/chan ii)]
      (sum in out)
      (println i ii (buffer-size in)))))

This table is much more interesting than the previous ones because the behavior it describes seems more complex. An equation to model this buffering would be something like x + 2y + 2 = z. The output buffer y is effectively doubles the size, because between the input and the output every two elements becomes one, and you have two variables in the process so add two.

Calling this sum process a buffer is kind of hand wavy. The sum process is consuming more than one item from the input to produce a single output, it really is performing some calculation. But if you stick this in the middle of a data flow you get behavior that is sort of buffer like, even if comparing it apples to apples with other buffers doesn't entirely make sense. I wave my hand at it for the sake of informal reasoning.

Now lets take a look at some processes that have a good deal more going on.

My interest in measuring buffering in core.async started with looking at the trio of pipeline functions. The pipeline functions create a group of worker processes taking input from a given channel and sending output to a given channel. I think they are the most complicated process creating functions that ship with core.async. Lets wire them to buffer-size and see what happens.

(dotimes [_ 5]
  (let [i (rand-int 100)
        ii (rand-int 100)
        iii (inc (rand-int 100))]
    (let [in (async/chan i)
          out (async/chan ii)]
      (async/pipeline iii out (map identity) in)
      (println "pipeline" i ii iii (buffer-size in)))
    (let [in (async/chan i)
          out (async/chan ii)]
      (async/pipeline-blocking iii out (map identity) in)
      (println "pipeline-blocking" i ii iii (buffer-size in)))
    (let [in (async/chan i)
          out (async/chan ii)]
      (async/pipeline-async iii out (fn [x o]
                                      (async/>! o x)
                                      (async/close! o)) in)
      (println "pipeline-async" i ii iii (buffer-size in)))))

pipeline and pipeline-blocking follow an equation like x + y + p = z where x is the input buffer size, y is the output buffer size, and p is the parallelism.

pipeline-async is the odd man out (and not just in buffering behavior). It can be modeled with x + p + 2 = z (ASYNC-163 apropos of that 2).

This is not a huge surprise because pipeline and pipeline-blocking are largely the same, they only differ on how the workers are run. pipeline-async differs in the kind of function it takes, and how the results of those functions are fed to the output. The fact that the buffering behavior of pipeline-async is not tied to the size of the output buffer is kind of troubling. I don't know of a concrete situation where that would be a problem, but it seems like that is at least a yellow flag.

• core.async
• these are very simplified examples
• Everything Will Flow
• https://en.wikipedia.org/wiki/Queueing_theory is as good a place to find further reading as any
• I am looking for a remote Clojure job, my email is hiredman@thelastcitadel.com
• Reductionem ad finem is another thing I wrote about some Clojure