B d1@sdZddlZddlZddlZddlZddlmZddlmZddlmZddl m Z Gdd d eZ d d Z d d Z ddZddZddZGdddZGddde ZGddde ZdS)zImplements MPIPoolExecutor.N)Future)Executor) as_completed)_libc@szeZdZdZeZdddZeejZ ddZ e dd Z dd d Z d dZddddddZdddZdddddZdS)MPIPoolExecutorz MPI-based asynchronous executor.NcKsv|dk r(t|}|dkr td||d<|dk rPt|s@td||d<||d<||_d|_d|_t|_ d|_ dS) aInitialize a new MPIPoolExecutor instance. Args: max_workers: The maximum number of MPI processes that can be used to execute the given calls. If ``None`` or not given then the number of worker processes will be determined from the MPI universe size attribute if defined, otherwise a single worker process will be spawned. initializer: An callable used to initialize workers processes. initargs: A tuple of arguments to pass to the initializer. Keyword Args: python_exe: Path to Python executable used to spawn workers. python_args: Command line arguments to pass to Python executable. mpi_info: Dict or iterable with ``(key, value)`` pairs. globals: Dict or iterable with global variables to set in workers. main: If ``False``, do not import ``__main__`` in workers. path: List of paths to append to ``sys.path`` in workers. wdir: Path to set current working directory in workers. env: Environment variables to update ``os.environ`` in workers. Nrz"max_workers must be greater than 0 max_workerszinitializer must be a callable initializerinitargsF) int ValueErrorcallable TypeError_options _shutdown_broken threadingLock_lock_pool)selfr r r kwargsrr`/work/yifan.wang/ringdown/master-ringdown-env/lib/python3.7/site-packages/mpi4py/futures/pool.py__init__s zMPIPoolExecutor.__init__cCs|jdkr|||_dS)N)r _make_pool)rrrr _bootstrapAs zMPIPoolExecutor._bootstrapc Cs@|j0|jrdS|jrdS||j|jjSQRXdS)N)rrrrrwaitsize)rrrr _max_workersEs zMPIPoolExecutor._max_workersTc Cs:|j*|jrtd||r,|j|SQRXdS)zAllocate executor resources eagerly. Args: wait: If ``True`` then bootup will not return until the executor resources are ready to process submissions. zcannot bootup after shutdownN)rr RuntimeErrorrrr)rrrrrbootupPs zMPIPoolExecutor.bootupc Os`|jP|jrt|j|jr(td||}|||f}|j ||f|SQRXdS)a&Submit a callable to be executed with the given arguments. Schedule the callable to be executed as ``fn(*args, **kwargs)`` and return a `Future` instance representing the execution of the callable. Returns: A `Future` representing the given call. zcannot submit after shutdownN) rrrBrokenExecutorrr rrrpush)rfnargsrfuturetaskrrrsubmit`s   zMPIPoolExecutor.submitrF)timeout chunksize unorderedcGs||t||||S)aReturn an iterator equivalent to ``map(fn, *iterables)``. Args: fn: A callable that will take as many arguments as there are passed iterables. iterables: Iterables yielding positional arguments to be passed to the callable. timeout: The maximum number of seconds to wait. If ``None``, then there is no limit on the wait time. chunksize: The size of the chunks the iterable will be broken into before being passed to a worker process. unordered: If ``True``, yield results out-of-order, as completed. Returns: An iterator equivalent to built-in ``map(func, *iterables)`` but the calls may be evaluated out-of-order. Raises: TimeoutError: If the entire result iterator could not be generated before the given timeout. Exception: If ``fn(*args)`` raises for any values. )starmapzip)rr$r)r*r+ iterablesrrrmapvszMPIPoolExecutor.mapcCsB|dkrtd|dkr*t|j||||St|j|||||SdS)aReturn an iterator equivalent to ``itertools.starmap(...)``. Args: fn: A callable that will take positional argument from iterable. iterable: An iterable yielding ``args`` tuples to be used as positional arguments to call ``fn(*args)``. timeout: The maximum number of seconds to wait. If ``None``, then there is no limit on the wait time. chunksize: The size of the chunks the iterable will be broken into before being passed to a worker process. unordered: If ``True``, yield results out-of-order, as completed. Returns: An iterator equivalent to ``itertools.starmap(fn, iterable)`` but the calls may be evaluated out-of-order. Raises: TimeoutError: If the entire result iterator could not be generated before the given timeout. Exception: If ``fn(*args)`` raises for any values. rzchunksize must be >= 1.N)r _starmap_helperr(_starmap_chunks)rr$iterabler)r*r+rrrr,s  zMPIPoolExecutor.starmap)cancel_futuresc Csr|jR|js(d|_|jdk r(|j|r@|jdk r@|jd}|rT|j}d|_WdQRX|dk rn|dS)aDClean-up the resources associated with the executor. It is safe to call this method several times. Otherwise, no other methods can be called after this one. Args: wait: If ``True`` then shutdown will not return until all running futures have finished executing and the resources used by the executor have been reclaimed. cancel_futures: If ``True`` then shutdown will cancel all pending futures. Futures that are completed or running will not be cancelled. TN)rrrdonecanceljoin)rrr3poolrrrshutdowns    zMPIPoolExecutor.shutdown)NNr)T)NrF)T)__name__ __module__ __qualname____doc__rr staticmethodrZ WorkerPoolrrpropertyrr!r(r/r,r8rrrrrs (   !rcsZdk r ttdtjfdd|Dr@tfdd}|S)N monotoniccsg|]}f|qSrr).0r%)functionr(rr sz#_starmap_helper..c3syrVdkrt}nt}xv|D]"}||g}|Vq.WnJdkrx8r|VqhWn xrVqWWn$xrqWYnXdS)N)rremovepopresultreverser5)iteratorr&)end_timefuturesr)timerr+rrresult_iterators&   z(_starmap_helper..result_iterator)getattrtimeset)r(rAr2r)r+rKr)rHrArIr(r)rJr+rr0s r0csfdd|DS)Ncsg|] }|qSrr)r@r%)rArrrBsz!_apply_chunks..r)rAchunkr)rAr _apply_chunkssrPccs2t|}x$tt||}|s"dS|fVq WdS)N)itertuple itertoolsislice)r*r2rOrrr _build_chunkss rUccs.x(|D] }|x|r$|VqWqWdS)N)rFrD)r2itemrrr_chain_from_iterable_of_listss rWcCs.tt|}t||}t|||||}t|S)N) functoolspartialrPrUr0rW)r(rAr2r)r+r*rErrrr1 s   r1c@s*eZdZdZd ddZddZdd ZdS) MPICommExecutoraWContext manager for `MPIPoolExecutor`. This context manager splits a MPI (intra)communicator in two disjoint sets: a single master process and the remaining worker processes. These sets are then connected through an intercommunicator. The target of the ``with`` statement is assigned either an `MPIPoolExecutor` instance (at the master) or ``None`` (at the workers). Example:: with MPICommExecutor(MPI.COMM_WORLD, root=0) as executor: if executor is not None: # master process executor.submit(...) executor.map(...) NrcKsX|dkrt}|r td|dks4||krs&        zMPICommExecutor.__enter__cGs,|j}d|_|dk r$|jdddSdSdS)z0Shutdown `MPIPoolExecutor` instance at the root.NT)rF)r^r8)rr%rerrr__exit__Ys  zMPICommExecutor.__exit__)Nr)r9r:r;r<rrarfrrrrrZs rZc@seZdZdZeejZdS)ThreadPoolExecutorz3`MPIPoolExecutor` subclass using a pool of threads.N)r9r:r;r<r=rrcrrrrrrgesrgc@seZdZdZeejZdS)ProcessPoolExecutorz5`MPIPoolExecutor` subclass using a pool of processes.N)r9r:r;r<r=rZ SpawnPoolrrrrrrhjsrh)r<rMrXrSrZ_corerrrrrr0rPrUrWr1rZrgrhrrrrs$    B#  R