Merge pull request #501 from SciML/ensemble

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

@@ -1,14 +1,47 @@
 using Distributed
 addprocs(2)
 println("There are $(nprocs()) processes")
-@everywhere using OrdinaryDiffEq
 
-@everywhere prob = ODEProblem((u,p,t)->1.01u,0.5,(0.0,1.0))
-@everywhere u0s = [rand()*prob.u0 for i in 1:2]
-@everywhere function prob_func(prob,i,repeat)
-    println("Running trajectory $i")
-    ODEProblem(prob.f,u0s[i],prob.tspan)
+@everywhere begin
+    using Pkg
+    Pkg.activate("downstream")
+    Pkg.develop(PackageSpec(path=joinpath(pwd(), "..")))
+    Pkg.instantiate()
+    using OrdinaryDiffEq
+    prob = ODEProblem((u,p,t)->1.01u,0.5,(0.0,1.0))
+    u0s = [rand()*prob.u0 for i in 1:2]
+    function prob_func(prob,i,repeat)
+        println("Running trajectory $i")
+        ODEProblem(prob.f,u0s[i],prob.tspan)
+    end
 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