fix thread-safety of splitthreads

Author: ChrisRackauckas

src/ensemble/basic_ensemble_solve.jl

@@ -208,9 +208,8 @@ end
 function thread_monte(prob,I,alg,procid,kwargs...)
   batch_data = Vector{Any}(undef,length(I))
   let
-    j = 0
-    Threads.@threads for i in I
-      j += 1
+    Threads.@threads for j in 1:length(I)
+      i = I[j]
       iter = 1
       _prob = prob.safetycopy ? deepcopy(prob.prob) : prob.prob
       new_prob = prob.prob_func(_prob,i,iter)

test/downstream/distributed_ensemble.jl

@@ -12,3 +12,30 @@ end
 
 ensemble_prob = EnsembleProblem(prob, prob_func=prob_func)
 sim = solve(ensemble_prob,Tsit5(),EnsembleSplitThreads(),trajectories=2)
+
+@everywhere function lorenz!(du,u,p,t)
+    du[1] = 10.0*(u[2]-u[1])
+    du[2] = u[1]*(28.0-u[3]) - u[2]
+    du[3] = u[1]*u[2] - (8/3)*u[3]
+end
+
+u0 = [1.0, 0.0, 0.0]
+tspan = (0.0, 100.0)
+p = [1, 2.0, 3]
+prob = ODEProblem(lorenz!, u0, tspan, p)
+
+@everywhere function prob_func(prob,i,repeat)
+    prob = remake(prob, tspan=(rand(), 100.0), p=rand(3))
+    return prob
+end
+
+ensemble_prob = EnsembleProblem(prob, prob_func = prob_func, safetycopy=true)
+
+println("Running EnsembleSerial()")
+@test length(solve(ensemble_prob, Tsit5(), EnsembleSerial(), trajectories=100)) == 100
+println("Running EnsembleThreads()")
+@test length(solve(ensemble_prob, Tsit5(), EnsembleThreads(), trajectories=100)) == 100
+println("Running EnsembleDistributed()")
+@test length(solve(ensemble_prob, Tsit5(), EnsembleDistributed(), trajectories=100)) == 100
+println("Running EnsembleSplitThreads()")
+@test length(solve(ensemble_prob, Tsit5(), EnsembleSplitThreads(), trajectories=100)) == 100