Python Process Pool non-daemonic?

I had the necessity to employ a non-daemonic pool in Python 3.7 and ended up adapting the code posted in the accepted answer. Below there's the snippet that creates the non-daemonic pool:

import multiprocessing.pool class NoDaemonProcess(multiprocessing.Process): @property def daemon(self): return False @daemon.setter def daemon(self, value): pass class NoDaemonContext(type(multiprocessing.get_context())): Process = NoDaemonProcess # We sub-class multiprocessing.pool.Pool instead of multiprocessing.Pool # because the latter is only a wrapper function, not a proper class. class NestablePool(multiprocessing.pool.Pool): def __init__(self, *args, **kwargs): kwargs['context'] = NoDaemonContext() super(NestablePool, self).__init__(*args, **kwargs) 

As the current implementation of multiprocessing has been extensively refactored to be based on contexts, we need to provide a NoDaemonContext class that has our NoDaemonProcess as attribute. NestablePool will then use that context instead of the default one.

That said, I should warn that there are at least two caveats to this approach:

1. It still depends on implementation details of the multiprocessing package, and could therefore break at any time.
2. There are valid reasons why multiprocessing made it so hard to use non-daemonic processes, many of which are explained here. The most compelling in my opinion is:

As for allowing children threads to spawn off children of its own using subprocess runs the risk of creating a little army of zombie 'grandchildren' if either the parent or child threads terminate before the subprocess completes and returns.

As of Python 3.8, concurrent.futures.ProcessPoolExecutor doesn't have this limitation. It can have a nested process pool with no problem at all:

from concurrent.futures import ProcessPoolExecutor as Pool from itertools import repeat from multiprocessing import current_process import time def pid(): return current_process().pid def _square(i): # Runs in inner_pool square = i ** 2 time.sleep(i / 10) print(f'{pid()=} {i=} {square=}') return square def _sum_squares(i, j): # Runs in outer_pool with Pool(max_workers=2) as inner_pool: squares = inner_pool.map(_square, (i, j)) sum_squares = sum(squares) time.sleep(sum_squares ** .5) print(f'{pid()=}, {i=}, {j=} {sum_squares=}') return sum_squares def main(): with Pool(max_workers=3) as outer_pool: for sum_squares in outer_pool.map(_sum_squares, range(5), repeat(3)): print(f'{pid()=} {sum_squares=}') if __name__ == "__main__": main() 

The above demonstration code was tested with Python 3.8.

A limitation of ProcessPoolExecutor, however, is that it doesn't have maxtasksperchild. If you need this, consider the answer by Massimiliano instead.

Credit: answer by jfs

The multiprocessing module has a nice interface to use pools with processes or threads. Depending on your current use case, you might consider using multiprocessing.pool.ThreadPool for your outer Pool, which will result in threads (that allow to spawn processes from within) as opposed to processes.

It might be limited by the GIL, but in my particular case (I tested both), the startup time for the processes from the outer Pool as created here far outweighed the solution with ThreadPool.

It's really easy to swap Processes for Threads. Read more about how to use a ThreadPool solution here or here.

On some Python versions replacing standard Pool to custom can raise error: AssertionError: group argument must be None for now.

Here I found a solution that can help:

class NoDaemonProcess(multiprocessing.Process): # make 'daemon' attribute always return False @property def daemon(self): return False @daemon.setter def daemon(self, val): pass class NoDaemonProcessPool(multiprocessing.pool.Pool): def Process(self, *args, **kwds): proc = super(NoDaemonProcessPool, self).Process(*args, **kwds) proc.__class__ = NoDaemonProcess return proc 

I have seen people dealing with this issue by using celery's fork of multiprocessing called billiard (multiprocessing pool extensions), which allows daemonic processes to spawn children. The walkaround is to simply replace the multiprocessing module by:

import billiard as multiprocessing 

The issue I encountered was in trying to import globals between modules, causing the ProcessPool() line to get evaluated multiple times.

globals.py

from processing import Manager, Lock from pathos.multiprocessing import ProcessPool from pathos.threading import ThreadPool class SingletonMeta(type): def __new__(cls, name, bases, dict): dict['__deepcopy__'] = dict['__copy__'] = lambda self, *args: self return super(SingletonMeta, cls).__new__(cls, name, bases, dict) def __init__(cls, name, bases, dict): super(SingletonMeta, cls).__init__(name, bases, dict) cls.instance = None def __call__(cls,*args,**kw): if cls.instance is None: cls.instance = super(SingletonMeta, cls).__call__(*args, **kw) return cls.instance def __deepcopy__(self, item): return item.__class__.instance class Globals(object): __metaclass__ = SingletonMeta """ This class is a workaround to the bug: AssertionError: daemonic processes are not allowed to have children The root cause is that importing this file from different modules causes this file to be reevalutated each time, thus ProcessPool() gets reexecuted inside that child thread, thus causing the daemonic processes bug """ def __init__(self): print "%s::__init__()" % (self.__class__.__name__) self.shared_manager = Manager() self.shared_process_pool = ProcessPool() self.shared_thread_pool = ThreadPool() self.shared_lock = Lock() # BUG: Windows: global name 'lock' is not defined | doesn't affect cygwin 

Then import safely from elsewhere in your code

from globals import Globals Globals().shared_manager Globals().shared_process_pool Globals().shared_thread_pool Globals().shared_lock 

I have written a more expanded wrapper class around pathos.multiprocessing here:

• https://github.com/JamesMcGuigan/python2-timeseries-datapipeline/blob/master/src/util/MultiProcessing.py

As a side note, if your usecase just requires async multiprocess map as a performance optimization, then joblib will manage all your process pools behind the scenes and allow this very simple syntax:

squares = Parallel(-1)( delayed(lambda num: num**2)(x) for x in range(100) ) 

• https://joblib.readthedocs.io/

Here is how you can start a pool, even if you are in a daemonic process already. This was tested in python 3.8.5

First, define the Undaemonize context manager, which temporarily deletes the daemon state of the current process.

class Undaemonize(object): '''Context Manager to resolve AssertionError: daemonic processes are not allowed to have children Tested in python 3.8.5''' def __init__(self): self.p = multiprocessing.process.current_process() if 'daemon' in self.p._config: self.daemon_status_set = True else: self.daemon_status_set = False self.daemon_status_value = self.p._config.get('daemon') def __enter__(self): if self.daemon_status_set: del self.p._config['daemon'] def __exit__(self, type, value, traceback): if self.daemon_status_set: self.p._config['daemon'] = self.daemon_status_value 

Now you can start a pool as follows, even from within a daemon process:

with Undaemonize(): pool = multiprocessing.Pool(1) pool.map(... # you can do something with the pool outside of the context manager 

While the other approaches here aim to create pool that is not daemonic in the first place, this approach allows you to start a pool even if you are in a daemonic process already.

This presents a workaround for when the error is seemingly a false-positive. As also noted by James, this can happen to an unintentional import from a daemonic process.

For example, if you have the following simple code, WORKER_POOL can inadvertently be imported from a worker, leading to the error.

import multiprocessing WORKER_POOL = multiprocessing.Pool() 

A simple but reliable approach for a workaround is:

import multiprocessing import multiprocessing.pool class MyClass: @property def worker_pool(self) -> multiprocessing.pool.Pool: # Ref: https://stackoverflow.com/a/63984747/ try: return self._worker_pool # type: ignore except AttributeError: # pylint: disable=protected-access self.__class__._worker_pool = multiprocessing.Pool() # type: ignore return self.__class__._worker_pool # type: ignore # pylint: enable=protected-access 

In the above workaround, MyClass.worker_pool can be used without the error. If you think this approach can be improved upon, let me know.

Since Python version 3.7 we can create non-daemonic ProcessPoolExecutor

Using if __name__ == "__main__": is necessary while using multiprocessing.

from concurrent.futures import ProcessPoolExecutor as Pool num_pool = 10 def main_pool(num): print(num) strings_write = (f'{num}-{i}' for i in range(num)) with Pool(num) as subp: subp.map(sub_pool,strings_write) return None def sub_pool(x): print(f'{x}') return None if __name__ == "__main__": with Pool(num_pool) as p: p.map(main_pool,list(range(1,num_pool+1))) 

For me it was that answer that helped: https://stackoverflow.com/a/71929459/14715428

Although the question itself is a little bit different, the speed remained the same and I didn't have to rewrite anything