Fix slot manager #884

Author: Bo Peng

src/sos/step_executor.py

@@ -1256,11 +1256,13 @@ def run(self):
                 else:
                     sm = SlotManager()
                     # because the master process pool will count one worker in (step)
-                    requested = sm.acquire(len(self._groups) - 1, env.config.get('max_procs', max(int(os.cpu_count() / 2), 1)))
-                    if requested == 0:
+                    gotten = sm.acquire(len(self._groups) - 1, env.config.get('max_procs', max(int(os.cpu_count() / 2), 1)))
+                    if gotten == 0:
+                        env.logger.debug(f'Input group executed sequencially due to -j constraint')
                         self.concurrent_input_group = False
                     else:
-                        self.worker_pool = Pool(requested + 1)
+                        env.logger.debug(f'Using process pool with size {gotten+1}')
+                        self.worker_pool = Pool(gotten + 1)
 
         try:
             for idx, (g, v) in enumerate(zip(self._groups, self._vars)):

src/sos/utils.py

@@ -917,14 +917,16 @@ def sos_handle_parameter_(key, defvalue):
 #        return True
 
 class SlotManager(object):
-    def __init__(self):
+    def __init__(self, reset=False):
         manager_id = env.config.get('master_md5', 'general')
         self.lock_file = os.path.join(os.path.expanduser('~'), '.sos', f'manager_{manager_id}.lck')
         self.slot_file = os.path.join(os.path.expanduser('~'), '.sos', f'manager_{manager_id}.slot')
+        if reset and os.path.isfile(self.slot_file):
+            os.remove(self.slot_file)
 
     def acquire(self, num=None, max_slots=10, force=False):
         # if num == None, request as many as possible slots
-        if not num:
+        if num is None:
             num = max_slots
         with fasteners.InterProcessLock(self.lock_file) as lock:
             if not os.path.isfile(self.slot_file):
@@ -934,14 +936,14 @@ def acquire(self, num=None, max_slots=10, force=False):
                     with open(self.slot_file, 'r') as slot:
                         slots = int(slot.read())
                 except Exception as e:
-                    print(e)
+                    env.logger.error(e)
                     slots = 0
             # return all available slots
             avail = max_slots - slots
-            ret = num if force else min(num, avail)
+            ret = num if force else max(min(num, avail), 0)
             with open(self.slot_file, 'w') as slot:
                 slot.write(str(ret + slots))
-            env.logger.debug(f'{ret} slots requested from {avail} avail')
+            env.logger.debug(f'{num} slots requested  {ret} returned ({slots} active, force={force})')
             return ret
 
     def release(self, num):
@@ -951,7 +953,7 @@ def release(self, num):
             # return all available slots
             with open(self.slot_file, 'w') as slot:
                 slot.write(str(max(0, slots - num)))
-            env.logger.debug(f'{num} slots released')
+            env.logger.debug(f'{num} slots released from {slots} active, {slots - num} remain')
             return max(0, slots - num)
 
 

src/sos/workflow_executor.py

@@ -238,7 +238,7 @@ def is_pending(self):
 
 class ExecutionManager(object):
     # this class managers workers and their status ...
-    def __init__(self, max_workers):
+    def __init__(self, max_workers, master=True):
                 #
         # running processes. It consisists of
         #
@@ -255,7 +255,7 @@ def __init__(self, max_workers):
         # process pool that is used to pool temporarily unused processed.
         self.pool = []
 
-        self.slot_manager = SlotManager()
+        self.slot_manager = SlotManager(reset=master)
         self.last_num_procs = None
 
         self.max_workers = max_workers
@@ -281,8 +281,8 @@ def add_placeholder_worker(self, runnable, pipe):
     def all_busy(self):
         n = len([x for x in self.procs if x and not x.is_pending()])
         if self.last_num_procs is None:
-            # we force the increase of numbers because `n` is observed
-            self.slot_manager.acquire(n, self.max_workers, force=True)
+            # clear counter file if already exists
+            self.slot_manager.acquire(n, self.max_workers)
             self.last_num_procs = n
         elif n != self.last_num_procs:
             if self.last_num_procs > n:
@@ -832,7 +832,7 @@ def i_am():
         #   node: node that is being executed, which is a dummy node
         #       created on the fly for steps passed from nested workflow
         #
-        manager = ExecutionManager(env.config['max_procs'])
+        manager = ExecutionManager(env.config['max_procs'], master=not nested)
         #
         wf_result = {'__workflow_id__': my_workflow_id, 'shared': {}}
         #