Make sure that all the cores are occupied for the entire duration of the...

Make sure that all the cores are occupied for the entire duration of the computation of the Bayesian and best-fit estimates. We do that with a chunksize of 1. This is not beneficial for the computation of the models as it would induce a large number of costly context switches and it would affect negatively the model cache as blocks would not be attributed consecutively to a process.

CHANGELOG.md

@@ -36,6 +36,7 @@
 - The computation of the weighted means and standard deviations has been made ~50% faster by normalising the likelihood. (Médéric Boquien)
 - The the fritz2006 module should now run faster thanks to an optimisation of the computation of the luminosity of the various AGN components (Médéric Boquien & Guang Yang)
 - Various optimisations have been made regarding shared arrays to make their access faster. The overall effect is a speedup of 3-4% for the computation of the models. (Médéric Boquien)
+- All the cores should now be used over the entire duration of the computation of the Bayesian and best-fit estimates. Before the number of active cores could drop towards the end of the computation. (Médéric Boquien)
 
 ## 2018.0 (2018-11-06)
 ### Added

pcigale/analysis_modules/pdf_analysis/__init__.py

@@ -121,16 +121,17 @@ class PdfAnalysis(AnalysisModule):
 
         initargs = (models, results, Counter(len(obs)))
         self._parallel_job(worker_analysis, obs, initargs,
-                           init_worker_analysis, conf['cores'])
+                           init_worker_analysis, conf['cores'], 1)
 
         return results
 
     def _compute_best(self, conf, obs, params, results):
         initargs = (conf, params, obs, results, Counter(len(obs)))
         self._parallel_job(worker_bestfit, obs, initargs,
-                           init_worker_bestfit, conf['cores'])
+                           init_worker_bestfit, conf['cores'], 1)
 
-    def _parallel_job(self, worker, items, initargs, initializer, ncores):
+    def _parallel_job(self, worker, items, initargs, initializer, ncores,
+                      chunksize=None):
         if ncores == 1:  # Do not create a new process
             initializer(*initargs)
             for idx, item in enumerate(items):
@@ -138,7 +139,7 @@ class PdfAnalysis(AnalysisModule):
         else:  # run in parallel
             with mp.Pool(processes=ncores, initializer=initializer,
                          initargs=initargs) as pool:
-                pool.starmap(worker, enumerate(items))
+                pool.starmap(worker, enumerate(items), chunksize)
 
     def _compute(self, conf, obs, params):
         results = []