Merge pull request #1103 from matthewrmshin/init-process-pool-early

Initialise process pool early

doc/siterc.tex

@@ -29,6 +29,16 @@ \subsubsection{temporary directory}
 \item {\em example:} \lstinline@temporary directory = /tmp/$USER/cylc@
 \end{myitemize}
 
+\subsubsection{process pool size}
+
+Number of process pool worker processes used to execute shell commands
+(job submission, event handlers, job poll and kill commands).
+
+\begin{myitemize}
+\item {\em type:} integer
+\item {\em default:} None (number of processor cores on the suite host)
+\end{myitemize}
+
 \subsubsection{state dump rolling archive length}
 
 A rolling archive of suite state dumps is maintained under the suite run

doc/suiterc.tex

@@ -168,14 +168,6 @@ \subsection{[cylc]}
     \item {\em default:} False
 \end{myitemize}
 
-\subsubsection[process pool size]{[cylc] $\rightarrow$ process pool size}
-Number of process pool worker processes used to execute shell commands
-(job submission, event handlers, job poll and kill commands).
-\begin{myitemize}
-    \item {\em type:} integer
-    \item {\em default:} None (number of processor cores on the suite host)
-\end{myitemize}
-
 \subsubsection[{[[}event hooks{]]}]{[cylc] $\rightarrow$ [[event hooks]]}
 \label{SuiteEventHandling}
 

lib/cylc/cfgspec/site.py

@@ -34,6 +34,7 @@
 USER_FILE = os.path.join( os.environ['HOME'], '.cylc', 'user.rc' )
 
 SPEC = {
+    'process pool size'                   : vdr( vtype='integer', default=None ),
     'temporary directory'                 : vdr( vtype='string' ),
     'state dump rolling archive length'   : vdr( vtype='integer', vmin=1, default=10 ),
     'disable interactive command prompts' : vdr( vtype='boolean', default=True ),

lib/cylc/cfgspec/suite.py

@@ -208,7 +208,6 @@ def _coerce_interval_list( value, keys, args, back_comp_unit_factor=1 ):
         'force run mode'                      : vdr( vtype='string', options=['live','dummy','simulation'] ),
         'abort if any task fails'             : vdr( vtype='boolean', default=False ),
         'log resolved dependencies'           : vdr( vtype='boolean', default=False ),
-        'process pool size'                   : vdr( vtype='integer', default=None ),
         'environment' : {
             '__MANY__'                        : vdr( vtype='string' ),
             },

lib/cylc/mp_pool.py

@@ -21,6 +21,7 @@
 import subprocess
 import multiprocessing
 
+from cylc.cfgspec.site import sitecfg
 import flags
 
 """Process pool to execute shell commands for the suite daemon.
@@ -101,7 +102,10 @@ class mp_pool(object):
     """Use a process pool to execute shell commands."""
 
     def __init__(self, pool_size=None):
-        self.pool_size = pool_size or multiprocessing.cpu_count()
+        self.pool_size = (
+            pool_size or
+            sitecfg.get(["process pool size"]) or
+            multiprocessing.cpu_count())
         # (The Pool class defaults to cpu_count anyway, but does not
         # expose the result via its public interface).
         self.log = logging.getLogger("main")

lib/cylc/scheduler.py

@@ -159,6 +159,7 @@ def __init__( self, is_restart=False ):
         self.parse_commandline()
 
     def configure( self ):
+        self.proc_pool = mp_pool()
         # read-only commands to expose directly to the network
         self.info_commands = {
                 'ping suite'        : self.info_ping_suite,
@@ -693,7 +694,6 @@ def configure_suite( self, reconfigure=False ):
             self.command_queue = comqueue( self.control_commands.keys() )
             self.pyro.connect( self.command_queue, 'command-interface' )
 
-            self.proc_pool = mp_pool( self.config.cfg['cylc']['process pool size'])
             task.task.proc_pool = self.proc_pool
 
             self.info_interface = info_interface( self.info_commands )