initial commit

Author: mpolinowski

01_ray-wait.py

@@ -0,0 +1,81 @@
+import ray
+import time
+import timeit
+
+# simulate remote functions with
+# different execution times
+@ray.remote
+def remote_task(x):
+    time.sleep(x)
+    return x
+
+# create a list of things
+things = list(range(10))
+# ensure that the futures won’t complete in order
+things.sort(reverse=True)
+
+
+# # GET
+
+# ## get results when all results are available
+# def in_order():
+#     ### use remote function to retrieve futures
+#     futures = list(map(lambda x: remote_task.remote(x), things))
+#     ### ray.get will block your function until all futures are returned
+#     values = ray.get(futures)
+#     ### loop over results and print
+#     for v in values:
+#         print(f" Completed {v}")
+#         ### simulate some business logic
+#         time.sleep(1)
+
+
+# ## call order and see how long it takes to complete
+# print("GET took: ", timeit.timeit(lambda: in_order(), number=1))
+
+
+# WAIT
+
+# ## process as results become available
+# def as_available():
+#     ### use remote function to retrieve futures
+#     futures = list(map(lambda x: remote_task.remote(x), things))
+#     ### while still futures left
+#     while len(futures) > 0:
+#         ### call ray.wait to get the next future
+#         ready_futures, rest_futures = ray.wait(futures)
+#         ### show progress
+#         print(f"Ready {len(ready_futures)} rest {len(rest_futures)}")
+#         ### print results
+#         for id in ready_futures:
+#             print(f'completed value {id}, result {ray.get(id)}')
+#             ### simulate some business logic
+#             time.sleep(1)
+#         ### wait on the ones that are not yet available
+#         futures = rest_futures
+
+# ## call order and see how long it takes to complete
+# print("WAIT took: ", timeit.timeit(lambda: as_available(), number=1))
+
+
+
+## same as above but with timeout and remote cancel
+def as_available():
+    futures = list(map(lambda x: remote_task.remote(x), things))
+    ### while still futures left
+    while len(futures) > 0:
+        ### call ray.wait to get the next future
+        ### but with a 10s timeout and always collect 5 results before returning anything
+        ready_futures, rest_futures = ray.wait(futures, timeout=10, num_returns=5)
+        # if we get back less than num_returns 
+        if len(ready_futures) < 1:
+            print(f"Timed out on {rest_futures}")
+            # cancel remote function, e.g. if task is using a lot of resources
+            ray.cancel(*rest_futures)
+            break
+        for id in ready_futures:
+            print(f'completed value {id}, result {ray.get(id)}')
+            futures = rest_futures
+
+## call order and see how long it takes to complete
+print("WAIT took: ", timeit.timeit(lambda: as_available(), number=1))

02_ray-pipeline.py

@@ -0,0 +1,24 @@
+import random
+import ray
+import time
+
+# first remote function
+## retrieve a value from somewhere
+@ray.remote
+def generate_number(s: int, limit: int, sl: float) -> int :
+    # create a random seed from value of 's'
+    random.seed(s)
+    # wait for value of 'sl'
+    time.sleep(sl)
+    # create a random number between '0' and the value of 'limit'
+    return random.randint(0, limit)
+
+# second remote function
+## take values and do some math
+@ray.remote
+def sum_values(v1: int, v2: int, v3: int) -> int :
+    # take three integer and add them up
+    return v1+v2+v3
+
+# use first remote function return  as input for second remote function
+print(ray.get(sum_values.remote(generate_number.remote(1, 10, .1), generate_number.remote(5, 20, .2), generate_number.remote(7, 15, .3))))

03_ray_nested_parallelism.py

@@ -0,0 +1,47 @@
+import random
+import ray
+import time
+
+# retrieve a value from somewhere
+@ray.remote
+def generate_number(s: int, limit: int) -> int :
+    # create a random seed from value of 's'
+    random.seed(s)
+    # simulate processing time
+    time.sleep(.1)
+    # create a random number between '0' and the value of 'limit'
+    return random.randint(0, limit)
+
+# invoke multiple remote functions and return the 'ObjectRef's
+@ray.remote
+def remote_objrefs():
+    results = []
+    for n in range(4):
+        results.append(generate_number.remote(n, 4*n))
+    return results
+
+# store returned 'ObjectRef's from remote_objrefs()
+futures = ray.get(remote_objrefs.remote())
+
+# invoke multiple remote functions and return the resulting values directly
+@ray.remote
+def remote_values():
+    results = []
+    for n in range(4):
+        results.append(generate_number.remote(n, 4*n))
+    return ray.get(results)
+
+# print returned results
+print(ray.get(remote_values.remote()))
+
+
+# take returned 'ObjectRef's and get results
+while len(futures) > 0:
+    ready_futures, rest_futures = ray.wait(futures, timeout=600, num_returns=1)
+    # break when the return is smaller than num_returns
+    if len(ready_futures) < 1:
+        ray.cancel(*rest_futures)
+        break
+    for id in ready_futures:
+        print(f'completed result {ray.get(id)}')
+        futures = rest_futures