Q: upmap - need to wait until all the tasks are done when an exception happens in one of the computations

[jumarko]

I'm using the upmap function like this

(vec (cp/upmap 5 clone-git-repo-fn repos))

vec is used explicitly to make sure everything is cloned before proceeding with other logic.

But sometimes this fails with an exception thrown by clone-git-repo-fn for one of the repos (e.g. because of incorrect repository information/access).

In that case, I do want to abort the process but also delete all the other repos that may have been cloned on the disk.
The trouble is that cloning of some of the other repositories may still be in progress - there's no guarantee that all the "git clone" processes have finished when the exception is thrown.

Therefore, I would need to wait until all the processes/threads started by upmap finished before proceeding with the cleanup.
I'm not sure how to do that.

Consider this contrived example:

 ;; 1. fire the tasks
  (def results
    (cp/upmap 5
              (fn [s] (Thread/sleep s)
                (when (or (= s 3000)
                          (= s 6000)) (throw (ex-info "error" {})))
                (println s) s)
              (shuffle [1000 2000 3000 4000 5000 6000 5000 4000 5000 7000])))

;; 2. evaluate this immediately after the expression above
  (try
    (vec results)
    (catch Exception e
      (println "Got an exception" e)
      (println "Waiting till all of the results are finished")
      (vec results)
      (println "FINISHED ?!?")
      ))

It prints

1000
Got an exception #error {
 :cause error
 :data {}
 :via
 [{:type clojure.lang.ExceptionInfo
...
Waiting till all of the results are finished
FINISHED ?!?
4000
5000
2000
7000
5000
5000
4000

Funny enough, if I try to evaluate the second expression again, I get only this without any exception:

[1000]

[karlmikko]

As per docs: https://github.com/clj-commons/claypoole?tab=readme-ov-file#how-are-exceptions-handled

Claypoole (as of 0.4.0) works just like core pmap: it will not kill queued tasks, but it will stop adding tasks to the queue. Like core pmap, Claypoole's functions have a buffer of tasks enqueued to keep the thread pool busy.

If you were using pmap, or claypoole version, if you don't want to abort and want the f called for all items, you need to ensure your f doesn't throw.

[jumarko]

@karlmikko thanks for the answer.
I do want to wait but only for the operations that are already in progress (= repos being cloned) - I do not want to wait for all the repos to be cloned (including ones for which git clone haven't started yet)