Using core.async for Producer-consumer Systems

README.md

Using core.async for Producer-consumer Systems

# Set up input file
repeat 10000 echo "." >> input

# Run inline
time lein run inline < input > output

# Run async
time lein run inline < input > output

Copyright © 2014 Elben Shira

Distributed under the Eclipse Public License either version 1.0 or (at your option) any later version.

core.clj

(ns test-speed.core
  (:gen-class)
  (:require [clojure.core.async :as async]))

(defn println-err [& args]
  (binding [*out* *err*]
    (apply println args)))

(defn process
  "Do 'work'"
  [line]
  (Thread/sleep 10)
  line)

(def stdin-reader
  (java.io.BufferedReader. *in*))

(def in-chan (async/chan))
(def out-chan (async/chan))

(defn start-async-consumers
  "Start num-consumers threads that will consume work
  from the in-chan and put it into the out-chan."
  [num-consumers]
  (dotimes [_ num-consumers]
    (async/thread
      (while true
        (let [line (async/<!! in-chan)
              data (process line)]
          (async/>!! out-chan data))))))

(defn start-async-aggregator
  "Take items from the out-chan and print it."
  []
  (async/thread
    (while true
      (let [data (async/<!! out-chan)]
        (println data)))))

(defn -main
  ""
  [& args]
  (let [[mode num-times] args]
    (println-err args)
    (case mode

      "async"
      (do
        (start-async-consumers 8)
        (start-async-aggregator)
        (doseq [line (line-seq stdin-reader)]
          (async/>!! in-chan line)))

      "inline"
      ;; Read each line from stdin
      (doseq [line (line-seq stdin-reader)]
        (let [data (process line)]
          (println data))))))

project.clj

(defproject test-speed "1.0.0"
  :description "core.async producer-consumer workflow"
  :url "http://example.com/FIXME"
  :license {:name "Eclipse Public License"
            :url "http://elbenshira.com/blog/using-core-async-for-producer-consumer-workflows/"}
  :dependencies [[org.clojure/clojure "1.7.0-alpha1"]
                 [org.clojure/core.async "0.1.338.0-5c5012-alpha"]]
  :main ^:skip-aot test-speed.core
  :target-path "target/%s"
  :profiles {:uberjar {:aot :all}})

Had some fun altering this a bit so it runs within clojure without using the terminal; overall it's similar. It also has a 3rd choice with streams: https://gist.github.com/viperscape/f0e13ab6227e8fe1de4d

Timing it in Clojure seems like a better idea than my use of time :)

Things I learned getting this code to run locally:

1. lein new test-speed is easiest way to drop in someone else's project.clj & core.clj (match their namespace, drop the core & project .clj's on top of the templated ones)
2. zsh has a 'repeat' shell builtin
3. Elben's Macbook Air is 10x faster than mine :(