[docs]classClosure(NamedTuple):""" A pickle-able lambda; note that f, args, and kwargs must be pickleable """f:Callable[...,None]args:List[Any]kwargs:Dict[str,Any]
[docs]classInitializer:""" A singleton class with global state used to initialize a multiprocessing.Process """_single:Optional["Initializer"]=None
[docs]@classmethoddefget(cls)->"Initializer":""" A wrapper around init since this class is a singleton """ifcls._singleisNone:cls._single=cls(_manual=False)returncls._single
[docs]def__init__(self,*,_manual:bool=True):if_manual:raiseRuntimeError("This is a singleton; call .get() instead")self.initializers:List[Closure]=[]
[docs]defregister(self,f:Callable[...,None],*args:Any,**kwargs:Any)->None:""" A shortcut for adding Closures as initializers """self.initializers.append(Closure(f,args,kwargs))
[docs]definitialize(self)->None:""" Initialize a multiprocessing.Process Set the current global initializer to the same state as this initializer, then calls each initializer """self._single=selfforiinself.initializers:i.f(*i.args,**i.kwargs)
[docs]defmp_context():system=platform.system()spawn_methods={"Windows":"spawn","Linux":"fork",# Python<3.8 defaults to fork# https://bugs.python.org/issue33725"Darwin":"spawn",}spawn_method=spawn_methods.get(system,"fork")# default to fork on other platformsreturnmultiprocessing.get_context(spawn_method)