extended pmap with kw args

base/multi.jl

@@ -1271,27 +1271,43 @@ pmap(f) = f()
 # rsym(n) = (a=rand(n,n);a*a')
 # L = {rsym(200),rsym(1000),rsym(200),rsym(1000),rsym(200),rsym(1000),rsym(200),rsym(1000)};
 # pmap(eig, L);
-function pmap(f, lsts...)
+function pmap(f, lsts...; err_retry=true, err_stop=false)
     np = nprocs()
     n = length(lsts[1])
     results = cell(n)
     queue = [1:n]
-    i = 1
+
+    task_in_err = false
+    is_task_in_error() = task_in_err
+    function set_task_in_error() 
+        task_in_err = true
+    end
+
     @sync begin
         for p=1:np
             wpid = PGRP.workers[p].id
             if wpid != myid() || np == 1
                 @async begin
                     while true
-                        if isempty(queue)
+                        if isempty(queue) || (is_task_in_error() && err_stop)
                             break
                         end
                         idx = shift!(queue)
                         try
                             results[idx] = remotecall_fetch(wpid, f,
                                                             map(L->L[idx], lsts)...)
-                        catch
-                            push!(queue, idx)
+                            if isa(results[idx], Exception)
+                                rethrow(results[idx])
+                            end
+                        catch e
+                            if err_retry
+                                push!(queue, idx) 
+                            else
+                                results[idx] = e
+                            end
+                            set_task_in_error()
+
+                            # remove this worker from accepting any more tasks 
                             break
                         end
                     end

test/runtests.jl

@@ -17,7 +17,7 @@ blas_set_num_threads(1)
 
 @everywhere include("testdefs.jl")
 
-reduce(propagate_errors, nothing, pmap(runtests, tests))
+reduce(propagate_errors, nothing, pmap(runtests, tests; err_retry=false, err_stop=true))
 
 @unix_only n > 1 && rmprocs(workers())
 println("    \033[32;1mSUCCESS\033[0m")