To learn about concurrency in Clojure, I wrote a function that accomplishes the following:

• Takes a (possibly lazy or infinite) sequence of functions representing tasks to be run asynchronously
• Runs the tasks asynchronously and in parallel
• Immediately returns a lazy sequence of promises that can be deref'd to get the return values of these tasks, in the same order
• Only starts a task once its promise has been requested from the lazy sequence returned
• Uses only up to a given number of threads at a time

My implementation, below, is based on the following algorithm:

• When an item in the lazy sequence is requested, create a promise.
• Place the task that needs to be executed in a ref containing a queue, and also place the promise in a separate ref containing another queue.
• Check the task queue, which entails the following:

• If there is a task on the queue ref, and the number of threads currently in use is less than the allowed number of threads, then:

• Take the task off the queue, and increase the number of threads in use.
• Run the task in a future; once it is done, deliver the return value to the corresponding promise, decrease the number of threads in use, and check the task queue again.
• Return the promise.

(ns temp.core)

[& more]
(locking *out* (apply println more)))

(def empty-queue clojure.lang.PersistentQueue/EMPTY)

(defn check-queue
(dosync
the-promise (first @promise-queue)]
(alter promise-queue pop)
; is it ok to use a future in a transaction?
; or does it count as a side-effect and it
; might be run multiple times (undesirable)?
(future
(deliver the-promise rv)
(dosync

(defn throttled-futures
"Takes (possibly lazy or infinite) sequence of tasks and runs them
in parallel using up to a specified number of threads at a time.
Returns a lazy sequence of objects that can be deref'd to get the
corresponding objects in the lazy sequence are requested."
(throttled-futures
(ref empty-queue)
(ref empty-queue)))
(lazy-seq
(cons (let [the-promise (promise)]
(dosync
(alter promise-queue conj the-promise)))
the-promise)
(throttled-futures
promise-queue))))))

(defn get-futures
[number-of-futures]
(println "Realizing" number-of-futures "futures")
(let [futures (doall
(take number-of-futures
(throttled-futures
3
; using a higher-arity form of map prevents chunking
; see http://stackoverflow.com/a/16567104/3538165
(map (fn [index _]
(fn []
index))
(range 10)
(range 10)))))]
(dorun (map deref futures))


I included get-futures to easily test throttled-futures, like so:

temp.core=> (get-futures 5)
Realizing 5 futures
(#object[clojure.core$promise$reify__6779 0x708d9ee0 {:status :pending, :val nil}] #object[clojure.core$promise$reify__6779 0x68964df5 {:status :pending, :val nil}] #object[clojure.core$promise$reify__6779 0x4876340d {:status :pending, :val nil}] #object[clojure.core$promise$reify__6779 0x72737be0 {:status :pending, :val nil}] #object[clojure.core$promise$reify__6779 0x6e1581d0 {:status :pending, :val nil}])
(#object[clojure.core$promise$reify__6779 0x708d9ee0 {:status :ready, :val 0}] #object[clojure.core$promise$reify__6779 0x68964df5 {:status :ready, :val 1}] #object[clojure.core$promise$reify__6779 0x4876340d {:status :ready, :val 2}] #object[clojure.core$promise$reify__6779 0x72737be0 {:status :ready, :val 3}] #object[clojure.core$promise$reify__6779 0x6e1581d0 {:status :ready, :val 4}])


As you can see, it starts three (value of max-threads) tasks, creates all of the promises, and then returns them. Then, once a task is finished, it starts a new task. In this case, it turned out that task 0 was long, so it finished last even though it was started first. Also, only five tasks are run because only the first five elements of the lazy sequence throttled-futures returned were realized.

Here are some of my concerns with this code:

• Are my uses of ref and dosync correct and/or appropriate?
• Is this code thread-safe in all situations, or is it possible for my refs to get out of sync?
• Are there any major performance concerns?
• Is there a more elegant way to approach this problem (for instance, by using a fixed number of threads instead of creating new ones and terminating old ones)?
• Welcome to Code Review! Good job on your first question. – Phrancis Mar 19 '16 at 16:45

Using future inside the dosync is not thread safe as future will create a thread and start running the task, so if the transaction is retried you will end up with two futures running the same task.

That said, I think that the refs cannot get out of sync.

Performance is quite relative. If your task are IO bound, the performance will be ok. If your task are CPU bound, probably reducers will be a better option.

Some more elegant approaches:

1. if io bound: (map #(future (%)) tasks). Clojure futures implement exactly the functionality that you describe.
2. if cpu bound: (pmap #(%) tasks)

Have a look at the Claypoole library for some inspiration.

Nice work!