deanwampler

import numpy as np
import ray

@ray.remote
class ParameterServer(object):
    def __init__(self, dim):
        # Alternatively, params could be a dictionary
        # mapping keys to arrays.
        self.params = np.zeros(dim)

deanwampler

>>> ray.init()

deanwampler

>>> ps = ParameterServer.remote(10)
>>> ps
Actor(ParameterServer, 7d58f41501000000)

deanwampler

>>> params_id = ps.get_params.remote()  # This returns a future.
>>> params_id
ObjectID(7268cb8d345ef26632430df6f18cc9690eb6b300)

deanwampler

>>> ray.get(params_id)  # This blocks until the task finishes.
array([0., 0., 0., 0., 0., 0., 0., 0., 0., 0.])

deanwampler

import time

@ray.remote
def worker(ps):
    for _ in range(100):  # Arbitrary number (100) of updates
        # First, get the latest parameters. The following
        # method call is non-blocking; it returns a future
        # effectively immediately.
        params_id = ps.get_params.remote()
        # As before, this is a blocking call that waits for

deanwampler

for _ in range(2):
    worker.remote(ps)

deanwampler

>>> ray.get(ps.get_params.remote())
array([64., 64., 64., 64., 64., 64., 64., 64., 64., 64.])
>>> ray.get(ps.get_params.remote())
array([78., 78., 78., 78., 78., 78., 78., 78., 78., 78.])
...

deanwampler

import ray, time
ray.init()

# A function that "stays busy" for some number of seconds
@ray.remote
def busy(i):
    time.sleep(i)
    return i

deanwampler

times = range(5,30,5) # 5, 10, ...
ids = [busy.remote(i) for i in times]  # Each waits progressively longer
for x in list(zip(times, ids)):        # Show what we have, a list of ids and the times each one will take 
    print(x)