Idiomatic Clojure: Mixing Parallel Side Effects and Iteration

This is what gets printed out:

1
2019
27
14
5
16
10
80
6
4
13
7
11
29
12
31
15
9
17
2
28
18
21
3
25

24
23
26
22
30

What are we seeing?

First, it's important to understand an implementation detail which concerns lazy sequences in Clojure: They are realized in chunks of 32 elements at a time.

Since pmap is lazy, a chunk was realized. But if you'll notice, it didn't take 64 seconds, although each api call takes ~2 seconds.

That is because pmap creates a future for every element in the sequence:

(map #(future (f %)) coll)

Another subtlety is that pmap is "semi" lazy, in that it tries to stay realize availableProcessors +2 elements ahead.

Hard to tell, pmap will give it a good try to stay ahead of your computation, but what happens when you use pmap in too many places in your code in parallel? The answer to that is not so deterministic.

pmap uses future which uses agents' soloExecutor service, which is a cached thread pool.

What does it mean in layman's terms? We can't know on which thread it happens, or on how many. In I think it's safe to assume each pmap call will use availableProcessors +2 threads. What happens if you call it twice in close succession?

This also complicates exception handling and the option of providing a default handler.

The value of xs will change in time and in execution, as it will block while the rest of the lazy sequence materializes.

pmap is great for computation since it provides a high degree of semi-lazy parallelism, however, since it is lazy, lacks control over execution context (threads number, exception handling, back pressure) it is unsuitable for performing side effects where these issues can make or break a system.

(import '[java.util.concurrent Executors ExecutorService])

(defn fixed-pool
  ([n]
   (Executors/newFixedThreadPool n))
  ([n factory]
   (Executors/newFixedThreadPool n factory)))

(defonce default-pool (delay (Executors/newFixedThreadPool 2)))

(defn submit*
  [pool f]
  (.submit ^ExecutorService pool ^Callable f))

(defmacro submit
  [pool & body]
  `(submit* ~pool (fn* [] ~@body)))

(defn pmap*
  ([f xs]
   (pmap* @default-pool f xs))
  ([pool f xs]
   (->> xs
        (mapv #(submit pool (f %)))
        (mapv deref))))

(pmap* (fixed-pool 2) api-call! (vec (range 10)))

For a slightly different example, see core.async's implementation.

One glaring deficiency in the previous example is the lack of back pressure, which could lead to us blowing the heap up with enqueued tasks.

Using others' implementation is no shame, so let's have a look at tolitius/lasync

As stated in the README, the purpose of the library is to be able to block on .submit and .execute, which is exactly the issue we saw in the previous implementation.

lasync/pool returns a ThreadPoolExecutor so it plugs right in to our pmap* implementation.

(require '[lasync.core :as lasync])

(def pool (lasync/pool :threads 2))

(pmap* pool api-call! (range 10))

If you don't want to implement anything yourself, including pmap, and perhaps need a richer API, which for example doesn't guarantee ordering, or supports parallel list comprehensions (for), take a look at TheClimateCorporation/claypoole

Usage is very straightforward:

(require '[com.climate.claypoole :as cp])
(def pool (cp/threadpool 4))
(def output (cp/pmap pool api-call! (range 64)))

(require '[clojure.core.async :as async])

(defn parallel
  "Returns a channel which will contain a transient vector of results"
  ([n f xs]
   (parallel nil n f xs))
  ([buf-or-n n f xs]
   (let [out (async/chan)]
     (async/pipeline-blocking
      n
      out
      (map f)
      (async/to-chan xs))
     (async/reduce conj! (transient []) out))))


(def ch (parallel 4 api-call! (range 16)))
(persistent! (async/<!! ch))
;; => [0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15]

If api-call! had been asynchronous pipeline-blocking can be swapped for pipeline-async, and instead of a transducer:

(fn af [v c]
  (api-call!
   v
   (fn cb [res] (async/put! c res) (async/close! c))))

A slight problem with the above implementation is that it has higher overhead than previous ones. It allocates plenty of channels, new async/thread s for each call, which isn't completely terrible as it uses a cached thread pool, and requires working with channels all the way.

It's probably a suitable solution if your entire code base is already written in the abstraction, otherwise I'm not sure I'd go with it.

If you're using core.async and haven't read Things I Wish I knew about core.async take a few extra minutes of your day to read it. Building off the final example in the post:

(defn async-wrapper [pool f]
  (let [ch (async/chan 1)]
    (.submit
     pool
     (fn []
       (try (async/put! ch (f))
            (catch Exception e (async/put! ch (ex-info "some error" {} e)))
            (finally (async/close! ch)))))
    ch))

(defmacro asyncly
  [pool & body]
  `(async-wrapper ~pool (fn* [] ~@body)))


(->> (range 8)
     (mapv #(asyncly @default-pool (api-call! %)))
     async/merge
     (async/reduce conj [])
     async/<!!)
;; => [1 0 3 2 4 5 6 7]

This example nicely ties together everything we've seen until now. Pay attention that it does not preserve order.