improve process pid management

slapos/recipe/erp5.recipe.testnode/setup.py

 from setuptools import setup, find_packages
 
 name = "erp5.recipe.testnode"
-version = '1.0.16'
+version = '1.0.19'
 
 def read(name):
   return open(name).read()

slapos/recipe/erp5.recipe.testnode/src/erp5/recipe/testnode/SlapOSControler.py

@@ -3,7 +3,7 @@ from xml_marshaller import xml_marshaller
 
 class SlapOSControler(object):
 
-  def __init__(self, config, process_group_pid_list=None):
+  def __init__(self, config, process_group_pid_set=None):
     self.config = config
     # By erasing everything, we make sure that we are able to "update"
     # existing profiles. This is quite dirty way to do updates...
@@ -11,7 +11,7 @@ class SlapOSControler(object):
       os.unlink(config['proxy_database'])
     proxy = subprocess.Popen([config['slapproxy_binary'],
       config['slapos_config']], close_fds=True, preexec_fn=os.setsid)
-    process_group_pid_list.append(proxy.pid)
+    process_group_pid_set.add(proxy.pid)
     # XXX: dirty, giving some time for proxy to being able to accept
     # connections
     time.sleep(2)
@@ -46,7 +46,7 @@ class SlapOSControler(object):
  'reference': config['computer_id'],
  'software_root': config['software_root']}))
 
-  def runSoftwareRelease(self, config, environment, process_group_pid_list=None):
+  def runSoftwareRelease(self, config, environment, process_group_pid_set=None):
     print "SlapOSControler.runSoftwareRelease"
     while True:
       cpu_count = os.sysconf("SC_NPROCESSORS_ONLN")
@@ -63,10 +63,11 @@ class SlapOSControler(object):
         config['slapos_config']],
         stdout=stdout, stderr=stderr,
         close_fds=True, preexec_fn=os.setsid)
-      process_group_pid_list.append(slapgrid.pid)
+      process_group_pid_set.add(slapgrid.pid)
       slapgrid.wait()
       stdout.seek(0)
       stderr.seek(0)
+      process_group_pid_set.remove(slapgrid.pid)
       status_dict = {'status_code':slapgrid.returncode,
                      'stdout':stdout.read(),
                      'stderr':stderr.read()}
@@ -74,7 +75,7 @@ class SlapOSControler(object):
       stderr.close()
       return status_dict
 
-  def runComputerPartition(self, config, process_group_pid_list=None):
+  def runComputerPartition(self, config, process_group_pid_set=None):
     print "SlapOSControler.runSoftwareRelease"
     slap = slapos.slap.slap()
     slap.registerOpenOrder().request(self.software_profile,
@@ -82,7 +83,8 @@ class SlapOSControler(object):
         partition_parameter_kw=config['instance_dict'])
     slapgrid = subprocess.Popen([config['slapgrid_partition_binary'],
       config['slapos_config'], '-c', '-v'], close_fds=True, preexec_fn=os.setsid)
-    process_group_pid_list.append(slapgrid.pid)
+    process_group_pid_set.add(slapgrid.pid)
     slapgrid.wait()
+    process_group_pid_set.remove(slapgrid.pid)
     if slapgrid.returncode != 0:
       raise ValueError('Slapgrid instance failed')

slapos/recipe/erp5.recipe.testnode/src/erp5/recipe/testnode/testnode.py

@@ -21,11 +21,11 @@ class SubprocessError(EnvironmentError):
 
 from Updater import Updater
 
-process_group_pid_list = []
+process_group_pid_set = set()
 process_pid_file_list = []
 process_command_list = []
 def sigterm_handler(signal, frame):
-  for pgpid in process_group_pid_list:
+  for pgpid in process_group_pid_set:
     try:
       os.killpg(pgpid, signal.SIGTERM)
     except:
@@ -96,6 +96,13 @@ repository = %(repository_path)s
   retry_software = False
   try:
     while True:
+      # kill processes from previous loop if any
+      for pgpid in process_group_pid_set:
+        try:
+          process_group_pid_set.remove(pgpid)
+          os.killpg(pgpid, signal.SIGTERM)
+        except:
+          pass
       # Make sure we have local repository
       if not os.path.exists(repository_path):
         parameter_list = [config['git_binary'], 'clone',
@@ -143,11 +150,11 @@ repository = %(repository_path)s
 
         # Now prepare the installation of SlapOS
         slapos_controler = SlapOSControler(config,
-          process_group_pid_list=process_group_pid_list)
+          process_group_pid_set=process_group_pid_set)
         # this should be always true later, but it is too slow for now
         status_dict = slapos_controler.runSoftwareRelease(config,
           environment=config['environment'],
-          process_group_pid_list=process_group_pid_list,
+          process_group_pid_set=process_group_pid_set,
           )
         if status_dict['status_code'] != 0:
           safeRpcCall(master.reportTaskFailure,
@@ -157,7 +164,7 @@ repository = %(repository_path)s
 
         # create instances, it should take some seconds only
         slapos_controler.runComputerPartition(config,
-                process_group_pid_list=process_group_pid_list)
+                process_group_pid_set=process_group_pid_set)
 
         # update repositories downloaded by buildout. Later we should get
         # from master a list of repositories
@@ -196,15 +203,16 @@ repository = %(repository_path)s
                          '--node_quantity', config['node_quantity'],
                          '--master_url', config['test_suite_master_url'],
                          ], )
-        process_group_pid_list.append(run_test_suite.pid)
+        process_group_pid_set.add(run_test_suite.pid)
         run_test_suite.wait()
+        process_group_pid_set.remove(run_test_suite.pid)
 
   finally:
     # Nice way to kill *everything* generated by run process -- process
     # groups working only in POSIX compilant systems
     # Exceptions are swallowed during cleanup phase
-    print "going to kill %r" % (process_group_pid_list,)
-    for pgpid in process_group_pid_list:
+    print "going to kill %r" % (process_group_pid_set,)
+    for pgpid in process_group_pid_set:
       try:
         os.killpg(pgpid, signal.SIGTERM)
       except: