I noticed with the working pdoseq I posted earlier that sometimes the
threads on one core get ahead of those on the others, for some reason,
and then that core is idle for the rest of a job -- Windows, at least,
doesn't seem to reassign one or more threads to the freed core. So I
wrote this version:
(defmacro pdoseq
"Bindings as for for, but parallel execution as per pmap, pcalls,
pvalues; returns
nil."
[seq-exprs & body]
`(let [procs# (+ 2 (.availableProcessors (Runtime/getRuntime)))
calls# (for ~seq-exprs (fn [] ~@body))
chunks# (atom [nil (partition 100 calls#)])
threads# (for [i# (range procs#)]
(Thread.
#(loop []
(let [chunk# (first
(swap! chunks# (fn [[_# [a# &
b#]]] [a# b#])))]
(when chunk#
(doall
(map (fn [x#] (x#)) chunk#))
(recur))))))]
(doseq [t# threads#]
(.start t#))
(doseq [t# threads#]
(.join t#))))
This one replaces the interleaving with a job queue, which each thread
atomically takes from. The input items are chunked into hundreds in
the code above, but that number can easily be changed (or even made
into a parameter of the macro). I think this should divide the work up
more evenly among the available cores until nearly the end (if there
are thousands or more of calls in calls#) when there will eventually
be no chunks left unassigned and worker threads becoming idle.
In case anyone might find this useful, I'm relinquishing any copyright
in either version and placing both into the public domain for others
to freely reuse.
--
You received this message because you are subscribed to the Google
Groups "Clojure" group.
To post to this group, send email to clojure@googlegroups.com
Note that posts from new members are moderated - please be patient with your
first post.
To unsubscribe from this group, send email to
clojure+unsubscr...@googlegroups.com
For more options, visit this group at
http://groups.google.com/group/clojure?hl=en
