dmtsource / deap_ray Goto Github PK
View Code? Open in Web Editor NEWRay based map function for Deap with examples. Automatically batches iterator to multiple workers for many processors or clusters.
Ray based map function for Deap with examples. Automatically batches iterator to multiple workers for many processors or clusters.
Investigating odd behavior when using the DeltaPenalty decorator in onemax_ray.py.
The eval function is not being used at all, this was noticed when working on #1 and the arbitrary speedup had no affect. Occurs with standard use of map and Ray batching...so I seem to have messed up implementing DeltaPenalty. Trying to fix asap.
when i tried to run symbreg_ray.py in my ray cluster, below errors occures:
AttributeError: Can't get attribute 'rand101' on <module 'deap.gp' from '/home/ray/anaconda3/lib/python3.8/site-packages/deap/gp.py'>
any suggestions? @DMTSource
It's mentioned in worker. get that this method will issue a warning if it's running inside async context. The problem I encounter is likely to be caused by this reason.
import ray
import time
from ray.util import ActorPool
from deap import algorithms, base, creator, gp, tools
......
@ray.remote(num_cpus=1)
class Ray_Deap_Map():
def __init__(self, creator_setup=None, pset_creator=None):
# issue 946? Ensure non trivial startup to prevent bad load balance across a cluster
time.sleep(0.01)
# recreate scope from global
# For GA no need to provide pset_creator. Both needed for GP
self.creator_setup = creator_setup
if creator_setup is not None:
self.creator_setup()
self.pset_creator = pset_creator
if pset_creator is not None:
self.pset_creator()
def ray_remote_eval_batch(self, f, zipped_input):
iterable, id_ = zipped_input
# attach id so we can reorder the batches
return [(f(i), id_) for i in iterable]
class Ray_Deap_Map_Manager():
def __init__(self, creator_setup=None, pset_creator=None):
# Can adjust the number of processes in ray.init or when launching cluster
self.n_workers = int(ray.cluster_resources()['CPU'])
# recreate scope from global (for ex need toolbox in gp too)
self.creator_setup = creator_setup
self.pset_creator = pset_creator
def map(self, func, iterable):
if self.n_workers == 1:
# only 1 worker, normal listcomp/map will work fine. Useful for testing code?
##results = [func(item) for item in iterable]
results = list(map(func, iterable)) # forced eval to time it
else:
# many workers, lets use ActorPool
if len(iterable) < self.n_workers:
n_workers = len(iterable)
else:
n_workers = self.n_workers
n_per_batch = int(len(iterable) / n_workers) + 1
batches = [iterable[i:i + n_per_batch] for i in range(0, len(iterable), n_per_batch)]
id_for_reorder = range(len(batches))
eval_pool = ActorPool(
[Ray_Deap_Map.remote(self.creator_setup, self.pset_creator) for _ in range(n_workers)])
unordered_results = list(eval_pool.map_unordered(
lambda actor, input_tuple: actor.ray_remote_eval_batch.remote(func, input_tuple),
zip(batches, id_for_reorder)))
# ensure order of batches
ordered_batch_results = [batch for batch_id in id_for_reorder for batch in unordered_results if
batch_id == batch[0][1]]
# flatten batches to list of fitnes
results = [item[0] for sublist in ordered_batch_results for item in sublist]
return results
# This is what we register as map in deap toolbox.
# For GA no need to provide pset_creator. Both needed for GP
def ray_deap_map(func, pop, creator_setup=None, pset_creator=None):
# Manager will determine if batching is needed and crate remote actors to do work
map_ray_manager = Ray_Deap_Map_Manager(creator_setup, pset_creator)
results = map_ray_manager.map(func, pop)
return results
def Tool(pset, max_c, ids):
toolbox = base.Toolbox()
......
toolbox.register("evaluate", EvaluationBacktest, toolbox=toolbox, max_c=max_c, ids=ids)
toolbox.register("map", ray_deap_map, creator_setup=Creator_Setup,
pset_creator=Pset_Creator)
return toolbox
def EvaluationBacktest(individual, toolbox, max_c, ids):
func = toolbox.compile(expr=individual)
......
try:
df = func(......)
......
except:
result = -100
return result,
fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)
Here is the error message:
2021-06-10 21:12:18,290 WARNING worker.py:1115 -- The log monitor on node ALPHA failed with the following error:
OSError: [WinError 87] 参数错误。
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\_private\log_monitor.py", line 364, in <module>
log_monitor.run()
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\_private\log_monitor.py", line 285, in run
self.open_closed_files()
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\_private\log_monitor.py", line 172, in open_closed_files
self.close_all_files()
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\_private\log_monitor.py", line 107, in close_all_files
os.kill(file_info.worker_pid, 0)
SystemError: <built-in function kill> returned a result with an error set
2021-06-10 21:12:18,295 WARNING worker.py:1115 -- A worker died or was killed while executing task ffffffffffffffff151a7c83ccb29b66d20c565901000000.
Traceback (most recent call last):
File "D:\Factor\GA\GA_LongShortStrategy.py", line 569, in <module>
pop = algorithms.eaMuPlusLambda(pop, toolbox, mu=mu, lambda_=lambda_,
File "C:\Users\Administrator\anaconda3\lib\site-packages\deap\algorithms.py", line 320, in eaMuPlusLambda
fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)
File "D:\Factor\GA\RayMap.py", line 89, in ray_deap_map
results = map_ray_manager.map(func, pop)
File "D:\Factor\GA\RayMap.py", line 70, in map
unordered_results = list(eval_pool.map_unordered(
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\util\actor_pool.py", line 91, in map_unordered
yield self.get_next_unordered()
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\util\actor_pool.py", line 210, in get_next_unordered
return ray.get(future)
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\_private\client_mode_hook.py", line 47, in wrapper
return func(*args, **kwargs)
File "C:\Users\Administrator\anaconda3\lib\site-packages\ray\worker.py", line 1483, in get
raise value
RayActorError: The actor died unexpectedly before finishing this task.
Any help would be appreciated.
Testing the 2 examples, comparing to standard map vs 4 ray workers. Map is much faster with default example(as expected due to overhear?).
Arbitrability adding a loop to the eval function to ensure each individuals eval takes more time. Testing if batching out via ray ActorPool vs standard python map is improving things to ensure the implementation in ray_map.py is not inefficient.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.