Managing Complexity with Concurrent Programming - Clojure Data Analysis - page 78

Getting ready

We'll need to use the same dependencies as we did for Managing program complexity
with agents , and we'll use two values and functions from that recipe: data-files and
read-file-amounts .

How to do it…

For this recipe, we need to deine a few functions to work through a queue of input chunks
and then block until all of the processing is complete:

1. Most of what we use will be from the previous recipe, including read-file-

amounts . However, we'll wrap it in another function that takes its output and uses
commute to update the shared-count hashmap with the counts it has just read:
(defn read-update-amounts [m filename count-ref]
(dosync
(let [file-amounts (read-file-amounts m filename)]
(commute count-ref
#(merge-with + % file-amounts)))))

2. We're now ready for the main function. This creates the agents and the shared

reference, sends tasks to the agents, and waits for the results before returning them:
(defn main [data-files agent-count]
(let [counts (ref {})
agents (map agent (repeat agent-count {}))]
(dorun
(map #(send %1 read-update-amounts %2 counts)
(cycle agents)
data-files))
(doseq [a agents]
(await a))
@counts))

How it works…

Using this code looks exactly like how we'd expect:

user=> (def amounts (main data-files 8))
user=> (take 5 amounts)
(["" 106549]
["|N00032245|" 22750]
["|N00027812|" 150]