ctools

The parallelization should be done using the multiprocessing module that is available from Python 2.6 on. Ideally the code should be written in a way so that it is still working even if the multiprocessing module is not available.

Add cssens to the list

First attempt: make cslightcrv class itself pickalable
(accepted solution at https://stackoverflow.com/questions/9310053/how-to-make-my-swig-extension-module-work-with-pickle#9325185)
so that I can pass it with all its attributes (observations, times, etc) to an external function that carries out the operation to be parallelised. Fails for unclear reasons: the process hangs for a while and then ends with Killed: 9.

Jürgen pointed out this thread
https://stackoverflow.com/questions/27318290/why-can-i-pass-an-instance-method-to-multiprocessing-process-but-not-a-multipro
with some possible solutions to parallelize a class member rather than an external function.
Answer 4, the easiest, fails, because the wrapper function contains a new instance of the class, not the one with all the attributes etc.

Jürgen started working on the pickle-ability of gammalib/ctools objects (#1938), this issue will be on hold until that effort is completed.

# ============================================ #
# Global functions for multiprocessing support #
# ============================================ #
def _multiprocessing_func_wrapper(args):
   return _multiprocessing_func(*args)
def _multiprocessing_func(cls, i):
    return cls._timebin(i)

# Create pool of workers
from multiprocessing import Pool
pool = Pool(processes = self._nthreads)

# Run time bin analysis in parallel with map
args    = [(self, i) for i in range(self._tbins.size())]
results = pool.map(_multiprocessing_func_wrapper, args)

# Close pool and join
pool.close()
pool.join()

Note that the CPU clock seems to measure only the time spent outside the _timebin method in case of multiprocessing (nthreads != 1). The wall clock is of course subject to fluctuations, but decreases with increasing number of threads. There is a small increase of the CPU clock with number of threads, which may be related to the overhead of pickling the cslightcrv instance.

• still works fine
• log file is same as before

Do you really need the following piece?

# Close pool and join
pool.close()
pool.join() 

I thought this was not needed for the map method.

Tibaldo Luigi wrote:

I tested in Python 3 with your modifications:

• still works fine
• log file is same as before

Do you really need the following piece?
[...]

I thought this was not needed for the map method.

I’m not a multiprocessing expert, but it seems to be good to do this (see https://stackoverflow.com/questions/38271547/when-should-we-call-multiprocessing-pool-join)

I added a GLog::buffer() method to recover the buffer of a logger into a string, and I used this method in cslightcrv (see below). Before calling the _timebin method the logger is cleared and the buffer size is set to a large value so that normally the buffer will never be flushed into a file (I guess even that the clearing unsets the link to the log file). After _timebin the output of the buffer is recovered in a string, which is then passed together with other computing information to the client.

# ============================================ #
# Global functions for multiprocessing support #
# ============================================ #
def _multiprocessing_func_wrapper(args):
   return _multiprocessing_func(*args)
def _multiprocessing_func(cls, i):

    # Initialise thread logger
    cls._log.clear()
    cls._log.buffer_size(100000)

    # Compute light curve bin
    cstart  = cls.celapse()
    result  = cls._timebin(i)
    celapse = cls.celapse() - cstart
    buffer  = cls._log.buffer()

    # Close logger
    cls._log.close()

    # Collect thread information
    info = {'celapse': celapse, 'log': buffer}

    # Return light curve bin result and thread information
    return result, info

# Create pool of workers
from multiprocessing import Pool
pool = Pool(processes = self._nthreads)

# Run time bin analysis in parallel with map
args        = [(self, i) for i in range(self._tbins.size())]
poolresults = pool.map(_multiprocessing_func_wrapper, args)

# Close pool and join
pool.close()
pool.join()

# Construct results
results = []
for i in range(self._tbins.size()):
    results.append(poolresults[i][0])
    self._log_string(gammalib.TERSE, poolresults[i][1]['log'], False)

Forgot: all code is now in devel (GammaLib/ctools), hence you can start a fresh branch from devel to continue your developments.

I fixed the log file mangling issue. The code is in the devel branch.

Working on the parallelization of csspec I discovered a typo in gammalib GEbounds.i
The piece of code

state = tuple([self.emin(i) for i in range(self.size())]), \
        tuple([self.emin(i) for i in range(self.size())])

state = tuple([self.emin(i) for i in range(self.size())]), \
        tuple([self.emax(i) for i in range(self.size())])

Thanks for catching that one. The fix is merged into devel.

I implemented a way to handle the usage of csphagen through obsutils which seems reasonable to me. obsutils.get_onoff_obs has an additional parameter nthreads set by default to 0. This implies the default behavior for csphagen w.r.t. multiprocessing. However, in cslightsrv and csphagen I pass nthreads=1, so that the original number of processes set in those scripts is preserved.
For the future this is flexible enough that other possibilites are open/easy to implement.

Tibaldo Luigi wrote:

I implemented a way to handle the usage of csphagen through obsutils which seems reasonable to me. obsutils.get_onoff_obs has an additional parameter nthreads set by default to 0. This implies the default behavior for csphagen w.r.t. multiprocessing. However, in cslightsrv and csphagen I pass nthreads=1, so that the original number of processes set in those scripts is preserved.
For the future this is flexible enough that other possibilites are open/easy to implement.

That sounds like a good solution.

Good, I implemented the same logic for obsutils.sim, which is used in cspull (and, formally, in cssens and cstsdist, but there the On/Off analysis is not implemented).
I also updated the reference manual for all tools. Do you think we need a general introduction to multiprocessing in ctolls/cscripts?

Since for the end user things are pretty transparent I’m not sure that an “introduction” is needed. But maybe a page describing how multiprocessing is handled in ctools and cscripts would probably be valuable.

The question is whether before doing this we should add a nthreads parameter to all OpenMP parallelised ctools so that they behave from the User perspective exactly like cscripts. We should also note that OpenMP supports depends on the compiler that is used, and that currently, no OpenMP support comes by default on Mac OS X. We could even describe how to make OpenMP available on a Mac.

Yes, it runs successfully in about 10 minutes in my Python 3.5.4 installation. I just run it again: here you are the results:

PASS: test_python_ctools.sh
PASS: test_python_cscripts.sh
PASS: test_examples.py
============================================================================
Testsuite summary for ctools 1.6.0.dev1
============================================================================
# TOTAL: 3
# PASS:  3
# SKIP:  0
# XFAIL: 0
# FAIL:  0
# XPASS: 0
# ERROR: 0
============================================================================

The code hangs in the following loop in GObservations::likelihood::eval

// Loop over all observations. The omp for directive will deal
// with the iterations on the differents threads.
#pragma omp for
for (int i = 0; i < m_this->size(); ++i) {

    // Compute likelihood
    *cpy_value += m_this->m_obs[i]->likelihood(cpy_model,
                                               cpy_gradient,
                                               cpy_curvature,
                                               cpy_npred);

} // endfor: looped over observations

Looks like OpenMP is not compatible with fork(): https://stackoverflow.com/questions/49049388/understanding-openmp-shortcomings-regarding-fork

It looks like the modifications did it, but I will wait until a full night continuous integrations runs through. In addition, I will add unit tests for pickling for all tools, as they exists now for cslightcrv (since this was the unit test that posed problems).

I found a bug in the main loop of cspull and cstsdist which was first executed in parallel, and then another time in series, which made the overall execution time longer. An else statement was in fact missing for the case that nthreads is not larger than 1.

I merged the bug fix into devel.