Revised signal implementation and added PULSE for determining if a duplicate job is running.

Author: Nathan Lee

pyrunner/core/config.py

@@ -19,7 +19,6 @@
 from subprocess import Popen, PIPE
 from collections import deque
 
-
 class Config:
   """
   Captures framework-level configurations.
@@ -165,6 +164,8 @@ def __setitem__(self, key, value):
           self._attr[key]['value'] = True
         else:
           self._attr[key]['value'] = self._attr[key]['type'](value)
+      elif key == 'tickrate' and value < 1:
+        self._attr['tickrate']['value'] = 1
       else:
         self._attr[key]['value'] = self._attr[key]['type'](value)
 
@@ -238,17 +239,6 @@ def ctx_file(self):
     else:
       return '{}/{}.ctx'.format(self['temp_dir'], self['app_name'])
 
-  @property
-  def abort_sig_file(self):
-    """
-    Path/filename of job's .sig file.
-    This file should only appear in case any signal is to be sent to a running job.
-    """
-    if not self['temp_dir'] or not self['app_name']:
-      return None
-    else:
-      return '{}/.{}.sig.abort'.format(self['temp_dir'], self['app_name'])
-  
   def source_config_file(self, config_file):
     """
     Sources config file to export environment variables.

pyrunner/core/engine.py

@@ -17,6 +17,7 @@
 import pyrunner.core.constants as constants
 from pyrunner.core.config import Config
 from pyrunner.core.context import Context
+from pyrunner.core.signal import SignalHandler, SIG_ABORT, SIG_PAUSE
 from multiprocessing import Manager
 
 import os, sys, glob
@@ -45,6 +46,7 @@ def __init__(self):
   def initiate(self, **kwargs):
     """Begins the execution loop."""
 
+    signal_handler = SignalHandler(self.config)
     sys.path.append(self.config['worker_dir'])
     self.start_time = time.time()
     wait_interval = 1.0/self.config['tickrate'] if self.config['tickrate'] > 0 else 0
@@ -56,9 +58,10 @@ def initiate(self, **kwargs):
     # Execution loop
     try:
       while self.register.running_nodes or self.register.pending_nodes:
+        sig_set = signal_handler.consume()
 
         # Check for abort signals
-        if os.path.exists(self.config.abort_sig_file):
+        if SIG_ABORT in sig_set:
           print('ABORT signal received! Terminating all running Workers.')
           self._abort_all_workers()
           return -1

pyrunner/core/pyrunner.py

@@ -29,6 +29,7 @@
 from pyrunner.core.engine import ExecutionEngine
 from pyrunner.core.config import Config
 from pyrunner.core.register import NodeRegister
+from pyrunner.core.signal import SignalHandler, SIG_ABORT, SIG_PAUSE, SIG_PULSE
 from pyrunner.version import __version__
 
 from datetime import datetime as datetime
@@ -41,23 +42,25 @@ def __init__(self, **kwargs):
     self._environ = os.environ.copy()
     self.config = Config()
     self.notification = notification.EmailNotification()
+    self.signal_handler = SignalHandler(self.config)
 
     self.serde_obj = serde.ListSerDe()
     self.register = NodeRegister()
     self.engine = ExecutionEngine()
 
     self._init_params = {
-      'config_file'       : kwargs.get('config_file'),
-      'proc_file'         : kwargs.get('proc_file'),
-      'restart'           : kwargs.get('restart', False),
-      'cvar_list'         : [],
-      'exec_proc_name'    : None,
-      'exec_only_list'    : [],
-      'exec_disable_list' : [],
-      'exec_from_id'      : None,
-      'exec_to_id'        : None
+      'config_file'          : kwargs.get('config_file'),
+      'proc_file'            : kwargs.get('proc_file'),
+      'restart'              : kwargs.get('restart', False),
+      'cvar_list'            : [],
+      'exec_proc_name'       : None,
+      'exec_only_list'       : [],
+      'exec_disable_list'    : [],
+      'exec_from_id'         : None,
+      'exec_to_id'           : None
     }
 
+    # Lifecycle hooks
     self._on_create_func = None
     self._on_start_func = None
     self._on_restart_func = None
@@ -66,11 +69,23 @@ def __init__(self, **kwargs):
     self._on_destroy_func = None
 
     self.parse_args()
+    
+    if self.dup_proc_is_running():
+      raise OSError('Another process for "{}" is already running!'.format(self.config['app_name']))
 
   def reset_env(self):
     os.environ.clear()
     os.environ.update(self._environ)
 
+  def dup_proc_is_running(self):
+    self.signal_handler.emit(SIG_PULSE)
+    time.sleep(1.1)
+    if SIG_PULSE not in self.signal_handler.peek():
+      print(self.signal_handler.peek())
+      return True
+    else:
+      return False
+  
   def load_proc_file(self, proc_file, restart=False):
     if not proc_file or not os.path.isfile(proc_file):
       return False
@@ -367,7 +382,7 @@ def parse_args(self):
     longopt_list = [
       'setup', 'help', 'nozip', 'interactive', 'abort',
       'restart', 'version', 'dryrun', 'debug',
-      'preserve-context', 'dump-logs', 'disable-exclusive-jobs',
+      'preserve-context', 'dump-logs', 'allow-duplicate-jobs',
       'email=', 'email-on-fail=', 'email-on-success=', 'ef=', 'es=',
       'env=', 'cvar=', 'context=',
       'to=', 'from=', 'descendants=', 'ancestors=',
@@ -424,8 +439,8 @@ def parse_args(self):
         self.config['tickrate'] = int(arg)
       elif opt in ['--preserve-context']:
         self.preserve_context = True
-      elif opt in ['--disable-exclusive-jobs']:
-        self.disable_exclusive_jobs = True
+      elif opt in ['--allow-duplicate-jobs']:
+        self._init_params['allow_duplicate_jobs'] = True
       elif opt in ['--exec-proc-name']:
         self._init_params['exec_proc_name'] = arg
       elif opt == '--abort':
@@ -452,7 +467,7 @@ def parse_args(self):
 
     if abort:
       print('Submitting ABORT signal to running job for: {}'.format(self.config['app_name']))
-      open(self.config.abort_sig_file, 'a').close()
+      self.signal_handler.emit(SIG_ABORT)
       sys.exit(0)
 
     # Check if restart is possible (ctllog/ctx files exist)

pyrunner/core/signal.py

@@ -0,0 +1,44 @@
+# Copyright 2019 Comcast Cable Communications Management, LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+
+SIG_ABORT = 'sig.abort'
+SIG_PAUSE = 'sig.pause'
+SIG_PULSE = 'sig.pulse'
+
+_valid_signals = (SIG_ABORT, SIG_PAUSE, SIG_PULSE)
+
+class SignalHandler:
+  
+  def __init__(self, config):
+    self.config = config
+  
+  def sig_file(self, sig):
+    return '{}/.{}.{}'.format(self.config['temp_dir'], self.config['app_name'], sig)
+  
+  def emit(self, sig):
+    if sig not in _valid_signals: return ValueError('Unknown signal type: {}'.format(sig))
+    open(self.sig_file(sig), 'a').close()
+  
+  def consume(self):
+    sig_set = self.peek()
+    for sig in sig_set:
+      os.remove(self.sig_file(sig))
+    return sig_set
+  
+  def peek(self):
+    return set([ s for s in _valid_signals if os.path.exists(self.sig_file(s)) ])