Merge remote-tracking branch 'origin/main' into ritvik/ray_and_mac_test

Author: Ritvik Rao

auto_test.py

@@ -30,7 +30,7 @@ def searchForPython(python_implementations):
 
 
 # ----------------------------------------------------------------------------------
-TIMEOUT = 60  # timeout for each test (in seconds)
+TIMEOUT = 120  # timeout for each test (in seconds)
 CHARM_QUIET_AFTER_NUM_TESTS = 5
 
 commonArgs = ['++local']

charm4py/c_object_store.pyx

@@ -224,7 +224,7 @@ cdef class CObjectStore:
     cpdef void request_location_object(self, ObjectId obj_id, int requesting_pe):
         # this function is intended to be called on home pes of the object id
         cdef object obj = self.lookup_object(obj_id)
-        if obj != None:
+        if not (obj is None):
             self.proxy[requesting_pe].receive_remote_object(obj_id, obj)
             return
         cdef int pe = self.lookup_location(obj_id, fetch=False)
@@ -248,7 +248,7 @@ cdef class CObjectStore:
 
     cpdef void request_object(self, ObjectId obj_id, int requesting_pe):
         cdef object obj = self.lookup_object(obj_id)
-        if obj == None:
+        if obj is None:
             self.buffer_obj_request(obj_id, requesting_pe)
         else:
             self.proxy[requesting_pe].receive_remote_object(obj_id, obj)

charm4py/charm.py

@@ -156,6 +156,9 @@ def __init__(self):
         self.threadMgr = threads.EntryMethodThreadManager(self)
         self.createFuture = self.Future = self.threadMgr.createFuture
 
+        # The send buffer is not currently used since we do not buffer messages on
+        # the sender. This should be used later when scheduling decisions are to be
+        # based on locations of arguments
         self.send_buffer = MessageBuffer()
         self.receive_buffer = MessageBuffer()
         # TODO: maybe implement this buffer in c++
@@ -182,7 +185,7 @@ def print_dbg(self, *args, **kwargs):
     def get_future_value(self, fut):
         #self.print_dbg("Getting data for object", fut.id)
         obj = fut.lookup_object()
-        if obj == None:
+        if obj is None:
             local_f = LocalFuture()
             self.future_get_buffer[fut.store_id] = (local_f, fut)
             fut.request_object()
@@ -370,11 +373,11 @@ def recvArrayMsg(self, aid, index, ep, msg, dcopy_start):
                 for i, arg in enumerate(args[:-1]):
                     if isinstance(arg, Future):
                         dep_obj = arg.lookup_object()
-                        if dep_obj != None:
-                            args[i] = dep_obj
-                        else:
+                        if dep_obj is None:
                             dep_ids.append(arg.store_id)
                             arg.request_object()
+                        else:
+                            args[i] = dep_obj
             if len(dep_ids) > 0:
                 charm.receive_buffer.insert(dep_ids, (obj, ep, header, args))
             else:

charm4py/entry_method.py

@@ -47,7 +47,7 @@ def _run(self, obj, header, args, ret_fut=False):
                 fut = args[-1]
                 args = args[:-1]
             ret = getattr(obj, self.name)(*args)
-            if ret_fut and ret != None:
+            if ret_fut and not (ret is None):
                 fut.create_object(ret)
         except SystemExit:
             exit_code = sys.exc_info()[1].code

charm4py/threads.py

@@ -136,7 +136,7 @@ def is_local(self):
         if not self.store:
             raise ValueError("Operation not supported for future not"
                              " stored in the object store")
-        return self.lookup_object() != None
+        return not (self.lookup_object() is None)
 
     def create_object(self, obj):
         from .charm import charm

examples/ray/parameter_server.py

@@ -125,7 +125,7 @@ def compute_gradients(self, weights):
 def sync_train(args):
     ray.init()
 
-    iterations = 101
+    iterations = 11
     num_workers = 4
 
     model = ConvNet()
@@ -159,7 +159,7 @@ def async_train(args):
 
     model = ConvNet()
     test_loader = get_data_loader()[1]
-    iterations = 101
+    iterations = 11
     num_workers = 4
 
     print("Running asynchronous parameter server training.")

examples/ray/simple.py

@@ -1,5 +1,6 @@
 from charm4py import charm, coro, Chare, Array, ray
 from time import sleep
+import numpy as np
 
 
 @ray.remote
@@ -26,14 +27,19 @@ def main(args):
     ray.init()
     # create 3 instances of MyChare, distributed among cores by the runtime
     arr = [Compute.remote(i) for i in range(4)]
-
+    
+    obj1 = np.arange(100)
+    obj2 = np.arange(100)
+    a = ray.put(obj1)
+    b = ray.put(obj2)
     c = arr[0].add.remote(1, 2) # fut id 0
     d = arr[1].add.remote(3, c) # fut id 1
     e = arr[2].add.remote(2, d)
     f = arr[3].add.remote(c, 4)
-    g = add_task.remote(e, f)
+    g = arr[3].add.remote(a, b)
+    h = add_task.remote(e, f)
 
-    not_ready = [c, d, e, f, g]
+    not_ready = [c, d, e, f, g, h]
     while len(not_ready) > 0:
         ready, not_ready = ray.wait(not_ready)
         print("Fetched value: ", ray.get(ready))