Can call distributed processing from python

julia/loop.jl

@@ -1,117 +0,0 @@
-using Distributed
-const nprocs = 4
-addprocs(4)
-@everywhere include(".dataset_28330894764081783777.jl")
-@everywhere include(".hyperparams_28330894764081783777.jl")
-@everywhere include("sr.jl")
-
-
-# 1. Start a population on every process
-allPops = Future[]
-bestSubPops = [Population(1) for j=1:nprocs]
-hallOfFame = HallOfFame()
-
-for i=1:nprocs
-    npop=300
-    future = @spawnat :any Population(npop, 3)
-    push!(allPops, future)
-end
-
-npop=300
-ncyclesperiteration=3000
-fractionReplaced=0.1f0
-verbosity=convert(Int, 1e9)
-topn=10
-niterations=10
-
-
-# # 2. Start the cycle on every process:
-for i=1:nprocs
-    allPops[i] = @spawnat :any run(fetch(allPops[i]), ncyclesperiteration, verbosity=verbosity)
-end
-println("Started!")
-cycles_complete = nprocs * 10
-while cycles_complete > 0
-    for i=1:nprocs
-        if isready(allPops[i])
-            cur_pop = fetch(allPops[i])
-            bestSubPops[i] = bestSubPop(cur_pop, topn=topn)
-
-            #Try normal copy...
-            bestPops = Population([member for pop in bestSubPops for member in pop.members])
-
-            for member in cur_pop.members
-                size = countNodes(member.tree)
-                if member.score < hallOfFame.members[size].score
-                    hallOfFame.members[size] = deepcopy(member)
-                    hallOfFame.exists[size] = true
-                end
-            end
-
-            # Dominating pareto curve - must be better than all simpler equations
-            dominating = PopMember[]
-            open(hofFile, "w") do io
-                debug(verbosity, "\n")
-                debug(verbosity, "Hall of Fame:")
-                debug(verbosity, "-----------------------------------------")
-                debug(verbosity, "Complexity \t MSE \t Equation")
-                println(io,"Complexity|MSE|Equation")
-                for size=1:actualMaxsize
-                    if hallOfFame.exists[size]
-                        member = hallOfFame.members[size]
-                        curMSE = MSE(evalTreeArray(member.tree), y)
-                        numberSmallerAndBetter = sum([curMSE > MSE(evalTreeArray(hallOfFame.members[i].tree), y) for i=1:(size-1)])
-                        betterThanAllSmaller = (numberSmallerAndBetter == 0)
-                        if betterThanAllSmaller
-                            debug(verbosity, "$size \t $(curMSE) \t $(stringTree(member.tree))")
-                            println(io, "$size|$(curMSE)|$(stringTree(member.tree))")
-                            push!(dominating, member)
-                        end
-                    end
-                end
-                debug(verbosity, "")
-            end
-
-            # Try normal copy otherwise.
-            if migration
-                for k in rand(1:npop, round(Integer, npop*fractionReplaced))
-                    to_copy = rand(1:size(bestPops.members)[1])
-                    cur_pop.members[k] = PopMember(
-                        copyNode(bestPops.members[to_copy].tree),
-                        bestPops.members[to_copy].score)
-                end
-            end
-
-            if hofMigration && size(dominating)[1] > 0
-                for k in rand(1:npop, round(Integer, npop*fractionReplacedHof))
-                    # Copy in case one gets used twice
-                    to_copy = rand(1:size(dominating)[1])
-                    cur_pop.members[k] = PopMember(
-                       copyNode(dominating[to_copy].tree)
-                    )
-                end
-            end
-
-            allPops[i] = @spawnat :any let
-                tmp_pop = run(cur_pop, ncyclesperiteration, verbosity=verbosity)
-                for j=1:tmp_pop.n
-                    if rand() < 0.1
-                        tmp_pop.members[j].tree = simplifyTree(tmp_pop.members[j].tree)
-                        tmp_pop.members[j].tree = combineOperators(tmp_pop.members[j].tree)
-                        if shouldOptimizeConstants
-                            tmp_pop.members[j] = optimizeConstants(tmp_pop.members[j])
-                        end
-                    end
-                end
-                tmp_pop
-            end
-
-            global cycles_complete -= 1
-        end
-    end
-    sleep(1e-3)
-end
-
-rmprocs(nprocs)
-
-

julia/sr.jl

@@ -738,3 +738,109 @@ mutable struct HallOfFame
 end
 
 
+function fullRun(niterations::Integer;
+                npop::Integer=300,
+                ncyclesperiteration::Integer=3000,
+                fractionReplaced::Float32=0.1f0,
+                verbosity::Integer=0,
+                topn::Integer=10
+               )
+    # 1. Start a population on every process
+    allPops = Future[]
+    bestSubPops = [Population(1) for j=1:nprocs]
+    hallOfFame = HallOfFame()
+
+    for i=1:nprocs
+        npop=300
+        future = @spawnat :any Population(npop, 3)
+        push!(allPops, future)
+    end
+
+    # # 2. Start the cycle on every process:
+    for i=1:nprocs
+        allPops[i] = @spawnat :any run(fetch(allPops[i]), ncyclesperiteration, verbosity=verbosity)
+    end
+    println("Started!")
+    cycles_complete = nprocs * 10
+    while cycles_complete > 0
+        for i=1:nprocs
+            if isready(allPops[i])
+                cur_pop = fetch(allPops[i])
+                bestSubPops[i] = bestSubPop(cur_pop, topn=topn)
+
+                #Try normal copy...
+                bestPops = Population([member for pop in bestSubPops for member in pop.members])
+
+                for member in cur_pop.members
+                    size = countNodes(member.tree)
+                    if member.score < hallOfFame.members[size].score
+                        hallOfFame.members[size] = deepcopy(member)
+                        hallOfFame.exists[size] = true
+                    end
+                end
+
+                # Dominating pareto curve - must be better than all simpler equations
+                dominating = PopMember[]
+                open(hofFile, "w") do io
+                    debug(verbosity, "\n")
+                    debug(verbosity, "Hall of Fame:")
+                    debug(verbosity, "-----------------------------------------")
+                    debug(verbosity, "Complexity \t MSE \t Equation")
+                    println(io,"Complexity|MSE|Equation")
+                    for size=1:actualMaxsize
+                        if hallOfFame.exists[size]
+                            member = hallOfFame.members[size]
+                            curMSE = MSE(evalTreeArray(member.tree), y)
+                            numberSmallerAndBetter = sum([curMSE > MSE(evalTreeArray(hallOfFame.members[i].tree), y) for i=1:(size-1)])
+                            betterThanAllSmaller = (numberSmallerAndBetter == 0)
+                            if betterThanAllSmaller
+                                debug(verbosity, "$size \t $(curMSE) \t $(stringTree(member.tree))")
+                                println(io, "$size|$(curMSE)|$(stringTree(member.tree))")
+                                push!(dominating, member)
+                            end
+                        end
+                    end
+                    debug(verbosity, "")
+                end
+
+                # Try normal copy otherwise.
+                if migration
+                    for k in rand(1:npop, round(Integer, npop*fractionReplaced))
+                        to_copy = rand(1:size(bestPops.members)[1])
+                        cur_pop.members[k] = PopMember(
+                            copyNode(bestPops.members[to_copy].tree),
+                            bestPops.members[to_copy].score)
+                    end
+                end
+
+                if hofMigration && size(dominating)[1] > 0
+                    for k in rand(1:npop, round(Integer, npop*fractionReplacedHof))
+                        # Copy in case one gets used twice
+                        to_copy = rand(1:size(dominating)[1])
+                        cur_pop.members[k] = PopMember(
+                           copyNode(dominating[to_copy].tree)
+                        )
+                    end
+                end
+
+                allPops[i] = @spawnat :any let
+                    tmp_pop = run(cur_pop, ncyclesperiteration, verbosity=verbosity)
+                    for j=1:tmp_pop.n
+                        if rand() < 0.1
+                            tmp_pop.members[j].tree = simplifyTree(tmp_pop.members[j].tree)
+                            tmp_pop.members[j].tree = combineOperators(tmp_pop.members[j].tree)
+                            if shouldOptimizeConstants
+                                tmp_pop.members[j] = optimizeConstants(tmp_pop.members[j])
+                            end
+                        end
+                    end
+                    tmp_pop
+                end
+
+                cycles_complete -= 1
+            end
+        end
+        sleep(1e-3)
+    end
+end
+

pysr/sr.py

@@ -5,7 +5,8 @@ import pathlib
 import numpy as np
 import pandas as pd
 
-def pysr(X=None, y=None, weights=None, threads=4,
+def pysr(X=None, y=None, weights=None,
+            procs=4,
             niterations=100,
             ncyclesperiteration=300,
             binary_operators=["plus", "mult"],
@@ -35,6 +36,7 @@ def pysr(X=None, y=None, weights=None, threads=4,
             test='simple1',
             verbosity=1e9,
             maxsize=20,
+            threads=None, #deprecated
         ):
     """Run symbolic regression to fit f(X[i, :]) ~ y[i] for all i.
     Note: most default parameters have been tuned over several example
@@ -43,9 +45,7 @@ def pysr(X=None, y=None, weights=None, threads=4,
 
     :param X: np.ndarray, 2D array. Rows are examples, columns are features.
     :param y: np.ndarray, 1D array. Rows are examples.
-    :param threads: int, Number of threads (=number of populations running).
-        You can have more threads than cores - it actually makes it more
-        efficient.
+    :param procs: int, Number of processes (=number of populations running).
     :param niterations: int, Number of iterations of the algorithm to run. The best
         equations are printed, and migrate between populations, at the
         end of each.
@@ -91,6 +91,8 @@ def pysr(X=None, y=None, weights=None, threads=4,
         (as strings).
 
     """
+    if threads is not None:
+        raise ValueError("The threads kwarg is deprecated. Use procs.")
 
     # Check for potential errors before they happen
     assert len(binary_operators) > 0
@@ -155,7 +157,7 @@ const hofMigration = {'true' if hofMigration else 'false'}
 const fractionReplacedHof = {fractionReplacedHof}f0
 const shouldOptimizeConstants = {'true' if shouldOptimizeConstants else 'false'}
 const hofFile = "{equation_file}"
-const nthreads = {threads:d}
+const nprocs = {procs:d}
 const nrestarts = {nrestarts:d}
 const perturbationFactor = {perturbationFactor:f}f0
 const annealing = {"true" if annealing else "false"}
@@ -192,12 +194,18 @@ const weights = convert(Array{Float32, 1}, """f"{weight_str})"
     with open(f'/tmp/.dataset_{rand_string}.jl', 'w') as f:
         print(def_datasets, file=f)
 
+    with open(f'/tmp/.runfile_{rand_string}.jl', 'w') as f:
+        print(f'@everywhere include("/tmp/.hyperparams_{rand_string}.jl")', file=f)
+        print(f'@everywhere include("/tmp/.dataset_{rand_string}.jl")', file=f)
+        print(f'include("{pkg_directory}/sr.jl")', file=f)
+        print(f'fullRun({niterations:d}, npop={npop:d}, ncyclesperiteration={ncyclesperiteration:d}, fractionReplaced={fractionReplaced:f}f0, verbosity=round(Int32, {verbosity:f}), topn={topn:d})', file=f)
+        print(f'rmprocs(nprocs)', file=f)
+
 
     command = [
         'julia -O3',
-        '--threads auto',
-        '-e',
-        f'\'include("/tmp/.hyperparams_{rand_string}.jl"); include("/tmp/.dataset_{rand_string}.jl"); include("{pkg_directory}/sr.jl"); fullRun({niterations:d}, npop={npop:d}, ncyclesperiteration={ncyclesperiteration:d}, fractionReplaced={fractionReplaced:f}f0, verbosity=round(Int32, {verbosity:f}), topn={topn:d})\'',
+        f'-p {procs}',
+        f'/tmp/.runfile_{rand_string}.jl',
         ]
     if timeout is not None:
         command = [f'timeout {timeout}'] + command