Fix other parallelization issue with switch to @spawnat over @pmap

eureqa.jl

@@ -17,6 +17,8 @@ const parsimony = 0.01
 # How much to scale temperature by (T between 0 and 1)
 const alpha = 10.0
 
+const maxsize = 20
+
 
 
 
@@ -188,7 +190,8 @@ function mutateConstant(
         node = randomNode(tree)
     end
     
-    maxChange = T + 1.0
+    bottom = 0.1
+    maxChange = T + 1.0 + bottom
     factor = maxChange^rand()
     makeConstBigger = rand() > 0.5
     
@@ -310,12 +313,13 @@ function iterate(
     weights = [8, 1, 1, 1]
     weights /= sum(weights)
     cweights = cumsum(weights)
+    n = countNodes(tree)
     
     if mutationChoice < cweights[1]
         tree = mutateConstant(tree, T)
     elseif mutationChoice < cweights[2]
         tree = mutateOperator(tree)
-    elseif mutationChoice < cweights[3]
+    elseif mutationChoice < cweights[3] && n < maxsize
         tree = appendRandomOp(tree)
     elseif mutationChoice < cweights[4]
         tree = deleteRandomOp(tree)
@@ -407,6 +411,7 @@ end
 function regEvolCycle(pop::Population, T::Float64)::Population
     for i=1:Int(pop.n/10)
         baby = iterateSample(pop, T)
+        #printTree(baby.tree)
         oldest = argmin([pop.members[member].birth for member=1:pop.n])
         pop.members[oldest] = baby
     end
@@ -427,8 +432,9 @@ function run(
         if annealing
             pop = regEvolCycle(pop, allT[iT])
         else
-            pop = regEvolCycle(pop, 0.0)
+            pop = regEvolCycle(pop, 1.0)
         end
     end
     return pop
 end
+

paralleleureqa.jl

@@ -1,14 +1,14 @@
 using Distributed
-const nthreads = 8
-addprocs(nthreads)
+addprocs(8)
+@everywhere const nthreads = 8
 
 @everywhere include("eureqa.jl")
 
 println("Lets try to learn (x2^2 + cos(x3) + 5) using regularized evolution from scratch")
-const npop = 100
-const annealing = false
-const niterations = 10
-const ncyclesperiteration = 1000
+@everywhere const npop = 100
+@everywhere const annealing = false
+@everywhere const niterations = 30
+@everywhere const ncyclesperiteration = 10000
 
 # Generate random initial populations
 
@@ -16,43 +16,96 @@ const ncyclesperiteration = 1000
 @everywhere f = (pop,)->run(pop, ncyclesperiteration, annealing)
 
 
-function update(allPops::Array{Population, 1}, bestScore::Float64, pool::AbstractWorkerPool)
-    # Map it over our workers
-    #global allPops = deepcopy(pmap(f, deepcopy(allPops)))
-    curAllPops = deepcopy(pmap(f, allPops))
-    for j=1:nthreads
-        allPops[j] = curAllPops[j]
-    end
+allPops = [Population(npop, 3) for j=1:nthreads]
+bestScore = Inf
+# Repeat this many evolutions; we collect and migrate the best
+# each time.
+for k=1:4
+    # Spawn independent evolutions
+    futures = [@spawnat :any f(allPops[i]) for i=1:nthreads]
 
-    # Get best 10 models for each processes
-    bestPops = Population([member for pop in allPops for member in bestSubPop(pop).members])
+    # Gather them
+    for i=1:nthreads
+        allPops[i] = fetch(futures[i])
+    end
+    # Get best 10 models for each processes. Copy because we re-assign later.
+    bestPops = deepcopy(Population([member for pop in allPops for member in bestSubPop(pop).members]))
     bestCurScoreIdx = argmin([bestPops.members[member].score for member=1:bestPops.n])
     bestCurScore = bestPops.members[bestCurScoreIdx].score
-    if bestCurScore < bestScore
-        bestScore = bestCurScore
-        println(bestScore, " is the score for ", stringTree(bestPops.members[bestCurScoreIdx].tree))
-    end
+    println(bestCurScore, " is the score for ", stringTree(bestPops.members[bestCurScoreIdx].tree))
 
     # Migration
     for j=1:nthreads
-        allPops[j].members[1:50] = deepcopy(bestPops.members[rand(1:bestPops.n, 50)])
+        for k in rand(1:npop, 50)
+            # Copy in case one gets copied twice
+            allPops[j].members[k] = deepcopy(bestPops.members[rand(1:size(bestPops.members)[1])])
+        end
     end
-    return allPops, bestScore
 end
 
 
-function runExperiment()
-    # Do niterations cycles
-    allPops = [Population(npop, 3) for j=1:nthreads]
-    bestScore = Inf
-    #pool = CachingPool(workers())
-    pool = WorkerPool(workers())
-    for i=1:niterations
-        allPops, bestScore = update(allPops, bestScore, pool)
-    end
 
-    return bestScore
-end
 
-runExperiment()
 
+# julia> @everywhere include_string(Main, $(read("count_heads.jl", String)), "count_heads.jl")
+
+# julia> a = @spawnat :any count_heads(100000000)
+# Future(2, 1, 6, nothing)
+
+# julia> b = @spawnat :any count_heads(100000000)
+# Future(3, 1, 7, nothing)
+
+# julia> fetch(a)+fetch(b)
+# 100001564
+
+# allPops = [Population(npop, 3) for j=1:nthreads]
+# bestScore = Inf
+# for i=1:10
+    # tmpPops = fetch(pmap(f, allPops))
+    # allPops[1:nthreads] = tmpPops[1:nthreads]
+    # # Get best 11 models for each processes
+    # bestPops = Population([member for pop in allPops for member in bestSubPop(pop).members])
+    # bestCurScoreIdx = argmin([bestPops.members[member].score for member=1:bestPops.n])
+    # bestCurScore = bestPops.members[bestCurScoreIdx].score
+    # println(bestCurScore, " is the score for ", stringTree(bestPops.members[bestCurScoreIdx].tree))
+# end
+
+
+# function update(allPops::Array{Population, 1}, bestScore::Float64)
+    # # Map it over our workers
+    # #global allPops = deepcopy(pmap(f, deepcopy(allPops)))
+    # #curAllPops = deepcopy(pmap(f, allPops))
+    # curAllPops = pmap(f, allPops)
+    # for j=1:nthreads
+        # allPops[j] = curAllPops[j]
+    # end
+
+    # # Get best 10 models for each processes
+    # bestPops = Population([member for pop in allPops for member in bestSubPop(pop).members])
+    # bestCurScoreIdx = argmin([bestPops.members[member].score for member=1:bestPops.n])
+    # bestCurScore = bestPops.members[bestCurScoreIdx].score
+    # if bestCurScore < bestScore
+        # bestScore = bestCurScore
+        # println(bestScore, " is the score for ", stringTree(bestPops.members[bestCurScoreIdx].tree))
+    # end
+
+    # # Migration
+    # for j=1:nthreads
+        # allPops[j].members[1:50] = deepcopy(bestPops.members[rand(1:bestPops.n, 50)])
+    # end
+    # return allPops, bestScore
+# end
+
+
+# function runExperiment()
+    # # Do niterations cycles
+    # allPops = [Population(npop, 3) for j=1:nthreads]
+    # bestScore = Inf
+    # for i=1:niterations
+        # allPops, bestScore = update(allPops, bestScore)
+    # end
+
+    # return bestScore
+# end
+
+# runExperiment()