Ray ActorPool to operate on a fixed pool of actors

README.md

Ray ActorPool

An actor pool is a utility class similar to multiprocessing.Pool that lets you schedule Ray tasks over a fixed pool of actors.

References:

• ray-project/ray#2105 (comment)

• https://github.com/ray-project/ray/blob/5df801605e12de297fb660b360500d17bccdbd3b/python/ray/util/actor_pool.py

   Create an Actor pool from a list of existing actors.
   An actor pool is a utility class similar to multiprocessing.Pool that
   lets you schedule Ray tasks over a fixed pool of actors.
   Arguments:
       actors (list): List of Ray actor handles to use in this pool.
   Examples:
       >>> a1, a2 = Actor.remote(), Actor.remote()
       >>> pool = ActorPool([a1, a2])
       >>> print(pool.map(lambda a, v: a.double.remote(v), [1, 2, 3, 4]))
       [2, 4, 6, 8]
  

actor_pool.py

import ray

# num_cpus = psutil.cpu_count(logical=False)
num_cpus = len(psutil.Process().cpu_affinity())
print('Initializing Ray on {} cpus.'.format(num_cpus))
ray.init(num_cpus=num_cpus)
print('Ray initialization {} on {} cores.'.format('successful' if ray.is_initialized() else 'failed', num_cpus))

@ray.remote
class MyActor(object):
    def __init__(self):
        pass
    def func(self, x):
        return x * 2

actors = [MyActor.remote() for _ in range(num_cpus)]
actor_pool = ray.util.ActorPool(actors)
    
def ray_batch(args_list):
    res = actor_pool.map(lambda a, v: a.func.remote(v), args_list)
    return res
    
args_list = [range(10)]
results_list = ray_batch(args_list)