B dj!@sddlmZmZmZmZdgZyddlmZWnek rFdZYnXGddde Z Gddde Z d d Z Gd dde Z dS) )divisionprint_functionabsolute_importunicode_literalsMPIPool)MPINc@seZdZddZdS)_close_pool_messagecCsdS)Nz)selfr r [/work/yifan.wang/ringdown/master-ringdown-env/lib/python3.7/site-packages/emcee/mpi_pool.py__repr__sz_close_pool_message.__repr__N)__name__ __module__ __qualname__r r r r r rsrc@seZdZddZdS)_function_wrappercCs ||_dS)N)function)r rr r r __init__sz_function_wrapper.__init__N)r rrrr r r r rsrcCs tddS)Nz=Pool was sent tasks before being told what function to apply.) RuntimeError)taskr r r _error_functionsrc@sReZdZdZdddZddZdd Zd d Zd d ZddZ ddZ ddZ dS)rav A pool that distributes tasks over a set of MPI processes. MPI is an API for distributed memory parallelism. This pool will let you run emcee without shared memory, letting you use much larger machines with emcee. The pool only support the :func:`map` method at the moment because this is the only functionality that emcee needs. That being said, this pool is fairly general and it could be used for other purposes. Contributed by `Joe Zuntz `_. :param comm: (optional) The ``mpi4py`` communicator. :param debug: (optional) If ``True``, print out a lot of status updates at each step. :param loadbalance: (optional) if ``True`` and ntask > Ncpus, tries to loadbalance by sending out one task to each cpu first and then sending out the rest as the cpus get done. NFcCshtdkrtd|dkrtjn||_|j|_|jd|_||_t |_ ||_ |jdkrdt ddS)NzPlease install mpi4pyrz]Tried to create an MPI pool, but there was only one MPI process available. Need at least two.) r ImportErrorZ COMM_WORLDcommZGet_rankrankZGet_sizesizedebugrr loadbalance ValueError)r rrrr r r r7s  zMPIPool.__init__cCs |jdkS)z5 Is the current process the master? r)r)r r r r is_masterFszMPIPool.is_mastercCs|rtdt}x|jr0td|j|jj dtj |d}|jr`td|j||j t |t r|jrtd|jPt |tr|j|_|jrtd|j|jq||}|jrtd|j||j |jj|d|j d qWd S) zK If this isn't the master process, wait for instructions. zMaster node told to await jobs.zWorker {0} waiting for task.r)sourcetagstatusz%Worker {0} got task {1} with tag {2}.zWorker {0} told to quit.z+Worker {0} replaced its task function: {1}.z+Worker {0} sending answer {1} with tag {2}.)destr N)rrrStatusrprintformatrrrecvANY_TAGr isinstancerrrisend)r r!rresultr r r waitMs2   z MPIPool.waitcCsvt|}|s|dS||jk r|jr:td|||_t|}g}x0t|j D]"}|j j ||dd}| |qXWt j||jr||j kr`g}xVt|D]J\}}||j d} |jrtd|| ||j j || |d}| |qWt j|g} xRt|D]F}||j d} |jr\}}|d} |jrtd|| ||j j || |dqtW|j } dg|} xt|D]} t }|j jt jt j|d } |j} |j}| | |<|jrtd | || |kr|| }| }|jrNtd|| ||j j || |d| d7} qW| SdS) a Like the built-in :func:`map` function, apply a function to all of the values in a list and return the list of results. :param function: The function to apply to the list. :param tasks: The list of elements. Nz(Master replacing pool function with {0}.r)r"z)Sent task {0} to worker {1} with tag {2}.)r"r z*Master waiting for worker {0} with tag {1})rr r)rr r!z,Master received from worker {0} with tag {1})lenrr+rrr$r%rrangerrr)appendrRequestZwaitallr enumerater&r#Z ANY_SOURCEr'rr )r rtasksZntaskFrequestsirrZworkerresultsr*Zntasks_dispatchedZitaskr!r r r map{sz             z MPIPool.mapcOs|jj||S)zJ Equivalent to mpi4py :func:`bcast` collective operation. )rbcast)r argskwargsr r r r8sz MPIPool.bcastcCs6|r2x(t|jD]}|jjt|ddqWdS)z Just send a message off to all the pool members which contains the special :class:`_close_pool_message` sentinel. r)r"N)rr-rrr)r)r r4r r r closesz MPIPool.closecCs|S)Nr )r r r r __enter__szMPIPool.__enter__cGs |dS)N)r;)r r9r r r __exit__szMPIPool.__exit__)NFF) r rr__doc__rrr+r7r8r;r<r=r r r r rs .i ) __future__rrrr__all__Zmpi4pyrrobjectrrrrr r r r s