Hack to make submit work

Author: nchristensen

charm4py/pool.py

@@ -165,6 +165,7 @@ def startSingleTask(self, func, future, *args):
             job.remote = self.workers.runTask_star_th
         else:
             job.remote = self.workers.runTask_star
+        #future.set_running_or_notify_cancel()
         self.schedule()
 
     def start(self, func, tasks, result, ncores, chunksize):
@@ -218,6 +219,7 @@ def schedule(self):
         job = self.job_next
         prev = self
         while job is not None:
+            #job.future.set_running_or_notify_cancel()
             if len(self.idle_workers) == 0:
                 return
             while True:
@@ -482,6 +484,7 @@ def Task(self, func, args, ret=False, awaitable=False):
             f = Future()
         # unpack the arguments for sending to allow benefiting from direct copy
         #print(f)
+        #f.set_running_or_notify_cancel()
         self.pool_scheduler.startSingleTask(func, f, *args)
         return f
 
@@ -556,13 +559,16 @@ def submit(self, fn, /, *args, **kwargs):
                 "charm4py.pool.PoolExecutor object has been shut down")
 
         if kwargs is None or len(kwargs) == 0:
-            return self.pool.Task(fn, args, ret=True)
+            future = self.pool.Task(fn, args, ret=True)
             #return self.pool.map_async(_StarmappedFunction(fn), (args,),
             #                       chunksize=1, ncores=-1)
         else:
             # Task doesn't support kwargs so this sneaks them in with a tuple
             iterable_arg = tuple([tuple([args, frozendict(kwargs)])])
-            return self.pool.Task(_WrappedFunction(fn), iterable_arg, ret=True)
+            future = self.pool.Task(_WrappedFunction(fn), iterable_arg, ret=True)
+
+        future.set_running_or_notify_cancel()
+        return future
 
     def map(self, func, *iterables, timeout=None, chunksize=1, ncores=-1):
         if self.is_shutdown:
@@ -586,7 +592,7 @@ def shutdown(self, wait=True, *, cancel_futures=False):
                     job.future.cancel()
 
         # Is this necessary?
-        self.pool.pool_scheduler.schedule()
+        #self.pool.pool_scheduler.schedule()
 
         if wait:
             for job in self.pool.pool_scheduler.jobs:

charm4py/threads.py

@@ -106,7 +106,6 @@ def getTargetProxyEntryMethod(self):
 
     def deposit(self, result):
         """ Deposit a value for this future. """
-
         retval = False
         with self._condition:
             if self._state in {CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED}:
@@ -283,11 +282,23 @@ def set_running_or_notify_cancel(self):
         """
 
         # How to force this future to start running asynchronously?
+        # Can't really control which greenlet executes next so
+        # a greenlet might complete before it is marked to run.
+        # Need to make Concurrent.wait() work here.
+        #if self._state == FINISHED:
+        #    return True
         retval = super().set_running_or_notify_cancel()
         if retval:
             #self.waitReady(None)
+            # Pause this thread (and allow other threads to execute)
+            threadMgr.pauseThread()
+            #self.blocked = True
+            # Resume this thread
             #self.resume(threadMgr)
-            threadMgr.start()
+            #threadMgr.start()
+            #self.gr = getcurrent()
+            #threadMgr.resumeThread(self.gr, self)
+            #threadMgr.resumeThread(threadMgr.main_gr, self)
 
         return retval