B dV@sdZddlZddlZddlmZmZddlZddlZddlZddl Z ddZ ddZ da da dd Zd d ZGd d d ejjZddZGdddeZdddZdddZdS)z. Tools for creating pools of worker processes N) TimeoutError cpu_countc CsBy"ddlm}|j}|}|dkStttfk r<dSXdS)zM Check if this is the main control process and may handle one time tasks r)MPITN)mpi4pyr COMM_WORLDGet_rank ImportError ValueError RuntimeError)rcommrankr W/work/yifan.wang/ringdown/master-ringdown-env/lib/python3.7/site-packages/pycbc/pool.pyis_main_process s rcGs$ttjtj|dk r ||SdS)N)signalSIGINTSIG_IGN)initargsr r r_nointsrc CsD|\}}}ttjd7_WdQRXxtj|kr*||Sq*WdS)z7 Wrapper to ensure that all processes execute together N) _process_lock_numdonevalue)valuesZ numrequiredfcnrr r r _lockstep_fcns   rcCs||dS)N) terminatejoin)pr r r_shutdown_pool+sr csDeZdZdZdfdd ZddZdd Zd d Zdd d ZZ S) BroadcastPoolz2 Multiprocessing pool with a broadcast method Nr c sHtatddatt|}tt |j |||f|t t |dS)Nir)multiprocessingLockrValuer functoolspartialrsuperr!__init__atexitregisterr )self processesZ initializerZinitargskwdsZnoint) __class__r rr)2s   zBroadcastPool.__init__cCs t|jS)N)len_pool)r,r r r__len__;szBroadcastPool.__len__cCs*|tt|||fgt|}dt_|S)z Do a function call on every worker. Parameters ---------- fcn: funtion Function to call. args: tuple The arguments for Pool.map r)maprr0rr)r,rrresultsr r r broadcast>s zBroadcastPool.broadcastcs&tfdd|D}dt_|S)z Do a function call on every worker with different arguments Parameters ---------- fcn: funtion Function to call. args: tuple The arguments for Pool.map csg|]}t|fqSr )r0).0arg)rr,r r Wsz(BroadcastPool.allmap..r)r3rrr)r,rrr4r )rr,rallmapLs zBroadcastPool.allmapcCs^||||}xJy |dStk r.Yqtk rT||tYqXqWdS)a= Catch keyboard interuppts to allow the pool to exit cleanly. Parameters ---------- func: function Function to call items: list of tuples Arguments to pass chunksize: int, Optional Number of calls for each process to handle at once iN)Z map_asyncgetrKeyboardInterruptrr)r,funcitems chunksizer4r r rr3[s  zBroadcastPool.map)NNr )N) __name__ __module__ __qualname____doc__r)r2r5r9r3 __classcell__r r )r/rr!/s  r!cCs|||g|jdS)N)r3size)r,frr r r_dummy_broadcastrsrFc@seZdZddZddZdS) SinglePoolcCs|||gS)N)r3)r,rrr r rr5vszSinglePool.broadcastcsfdd|DS)Ncsg|] }|qSr r )r6a)rEr rr8zsz"SinglePool.map..r )r,rEr=r )rErr3yszSinglePool.mapN)r?r@rAr5r3r r r rrGusrGFTc Csd}yDddlm}|j}|}|}|dkrFd}|rFtd||Wn6tk r~}z|rnt|t dWdd}~XYnX|sd}}|||fS) zD Get whether MPI is enabled and if so the current size and rank Fr)rrTz)Running under mpi with size: %s, rank: %sz.Failed to load mpi, ensure mpi4py is installedN) rrrZGet_sizerlogginginforprintr ) require_mpiloguse_mpirr rDr er r rrN|s$  rNcCst|d\}}}|ryXddl}|j||dd}tt||_t|j |rXt d|rj|sjt dWqt k rt dYqXn&|dkrt}n|d krt}t|}||_|r||_|S) z Get processing pool )rLrNr)mpir-zYNOTE: that for MPI process size determined by MPI launch size, not the processes argumentz5NOTE: using MPI as this process was launchedunder MPIz;Failed to start up an MPI pool, install mpi4py / schwimmbad)rN schwimmbad choose_pooltypes MethodTyperFr5r*r+closerIrJrr rGrr!rD)r-rPZdo_mpirDr rRpoolr r rrSs.   rS)FT)F)rBZmultiprocessing.poolr#r&rrrTrr*rIrrrrrr rWZPoolr!rFobjectrGrNrSr r r rs$  C