3 QfXV]@sdZddlmZddlZddlZddlZddlZddlZddlZddlZddl Z yddl m Z ddl m Z Wn4e k rddlm Z ddlm Z ddlZYnXddlmZddlmZddlmZd d lmZmZdd lmZyddlZdd lmZWne k rdZYnXd d lmZd dlmZd dl m!Z!dZ"ej#ej$Bej%BZ&ej#ej$BZ'ddZ(ddZ)ddZ*ddZ+ddZ,Gddde-Z.GdddeZ/Gdd d e-Z0Gd!d"d"eZ1d#d$Z2Gd%d&d&e1Z3dS)'aCustom implementation of multiprocessing.Pool with custom pickler This module provides efficient ways of working with data stored in shared memory with numpy.memmap arrays without inducing any memory copy between the parent and child processes. This module should not be imported if multiprocessing is not available as it implements subclasses of multiprocessing Pool that uses a custom alternative to SimpleQueue. )mmapN)loads)dumps)Pickler)HIGHEST_PROTOCOL)BytesIO)mpassert_spawning)Pool) as_strided)load)dump)hashz/dev/shmcCs2t|dd}|dkrdSt|tr&|St|SdS)z?Recursively look up the original np.memmap instance base if anybaseN)getattr isinstancer_get_backing_memmap)abrR/home/psgendb/BIRCHDEV/pkg/SPAdes-3.15.4/linux-x86_64/share/spades/joblib3/pool.pyrGs   rcCs t|dk S)z>Return True if a is backed by some mmap buffer directly or notN)r)rrrrhas_shareable_memoryXsrc CsR|dkr d}|dkr*tj||||||dStj||||||d}t|||dSdS)z2Reconstruct an array view on a memmory mapped filezw+zr+N)dtypeshapemodeoffsetorder)rstrides)npmemmapr ) filenamerrrrrrtotal_buffer_lenrrrr_strided_from_memmap]s    r#c Cstj|\}}tj|d}||}||j7}|jdr>d}nd}|jdsV|jdr`d}d}n|j}|||j}t|j|j|j |||j ||ffS)a Pickling reduction for memmap backed arrays a is expected to be an instance of np.ndarray (or np.memmap) m is expected to be an instance of np.memmap on the top of the ``base`` attribute ancestry of a. ``m.base`` should be the real python mmap object. rZ F_CONTIGUOUSFCZ C_CONTIGUOUSN) rZ byte_boundsrflagsritemsizer#r!rrr) rmZa_startZa_endZm_startrrrr"rrr_reduce_memmap_backedps  r)cCs6t|}|dk rt||Stttj|tdffSdS)zBPickle the descriptors of a memmap instance to reopen on same fileN)protocol)rr)rrrasarrayr)rr(rrr reduce_memmaps r,c@s"eZdZdZd ddZddZdS) ArrayMemmapReduceraReducer callable to dump large arrays to memmap files. Parameters ---------- max_nbytes: int Threshold to trigger memmaping of large arrays to files created a folder. temp_folder: str Path of a folder where files for backing memmaped arrays are created. mmap_mode: 'r', 'r+' or 'c' Mode for the created memmap datastructure. See the documentation of numpy.memmap for more details. Note: 'w+' is coerced to 'r+' automatically to avoid zeroing the data on unpickling. verbose: int, optional, 0 by default If verbose > 0, memmap creations are logged. If verbose > 1, both memmap creations, reuse and array pickling are logged. context_id: int, optional, None by default Set to a value identifying a call context to spare costly hashing of the content of the input arrays when it is safe to assume that each array will not be mutated by the parent process for the duration of the dispatch process. This is the case when using the high level Parallel API. It might not be the case when using the MemmapingPool API directly. prewarm: bool, optional, False by default. Force a read on newly memmaped array to make sure that OS pre-cache it memory. This can be useful to avoid concurrent disk access when the same data array is passed to different worker processes. rNTcCs,||_||_||_t||_||_||_dS)N) _max_nbytes _temp_folder _mmap_modeintverbose _context_id_prewarm)self max_nbytes temp_folder mmap_moder2 context_idprewarmrrr__init__s  zArrayMemmapReducer.__init__cCst|}|dk rt||S|jj o8|jdk o8|j|jkrzytj|jtj |jt Wn2t k r}z|j t j kr||WYdd}~XnX|jdk r|j}nt|}dtjttjt||f}tjj|j|}tjj|sF|jdkrtd|j|j|fx t||D]}tj |tqW|jrht||jdjn"|jdkrhtd|j|j|ft t||jdS|jdkrtd|j|jft!t"|t#dffSdS) Nz%d-%d-%d-%s.pklrz-Memmaping (shape=%r, dtype=%s) to new file %s)r8rz-Memmaping (shape=%s, dtype=%s) to old file %sz$Pickling array (shape=%r, dtype=%s).)r*)$rr)r hasobjectr.nbytesosmakedirsr/chmodFOLDER_PERMISSIONSOSErrorerrnoEEXISTr3rgetpidid threadingcurrent_threadpathjoinexistsr2printrrFILE_PERMISSIONSr4r r0maxr,rrr)r5rr(eZmarkerbasenamer!Zdumped_filenamerrr__call__sD         zArrayMemmapReducer.__call__)rNT)__name__ __module__ __qualname____doc__r;rQrrrrr-s r-c@s&eZdZdZdefddZddZdS)CustomizablePickleraPickler that accepts custom reducers. HIGHEST_PROTOCOL is selected by default as this pickler is used to pickle ephemeral datastructures for interprocess communication hence no backward compatibility is required. `reducers` is expected expected to be a dictionary with key/values being `(type, callable)` pairs where `callable` is a function that give an instance of `type` will return a tuple `(constructor, tuple_of_objects)` to rebuild an instance out of the pickled `tuple_of_objects` as would return a `__reduce__` method. See the standard library documentation on pickling for more details. NcCsftj|||d|dkri}ttdr4tjj|_n tjj|_x |jD]\}}|j||qJWdS)N)r*dispatch) rr;hasattrrWcopycopyregdispatch_tableitemsregister)r5writerreducersr*type reduce_funcrrrr;s  zCustomizablePickler.__init__cs0ttdr"fdd}||j|<n |j|<dS)NrWcs|}|j|d|idS)Nobj) save_reduce)r5rbZreduced)rarr dispatcher2sz0CustomizablePickler.register..dispatcher)rXrrWr[)r5r`rardr)rarr].s   zCustomizablePickler.register)rRrSrTrUrr;r]rrrrrV srVc@s:eZdZdZd ddZddZddZd d Zd d ZdS)CustomizablePicklingQueueaLocked Pipe implementation that uses a customizable pickler. This class is an alternative to the multiprocessing implementation of SimpleQueue in order to make it possible to pass custom pickling reducers, for instance to avoid memory copy when passing memmory mapped datastructures. `reducers` is expected expected to be a dictionary with key/values being `(type, callable)` pairs where `callable` is a function that give an instance of `type` will return a tuple `(constructor, tuple_of_objects)` to rebuild an instance out of the pickled `tuple_of_objects` as would return a `__reduce__` method. See the standard library documentation on pickling for more details. NcCsL||_|jdd\|_|_|j|_tjdkr6d|_n |j|_|j dS)NF)duplexwin32) _reducersPipe_reader_writerLock_rlocksysplatform_wlock _make_methods)r5contextr_rrrr;Js   z"CustomizablePicklingQueue.__init__cCs t||j|j|j|j|jfS)N)r rjrkrmrprh)r5rrr __getstate__Tsz&CustomizablePicklingQueue.__getstate__cCs$|\|_|_|_|_|_|jdS)N)rjrkrmrprhrq)r5staterrr __setstate__Ysz&CustomizablePicklingQueue.__setstate__cCs |jj S)N)rjpoll)r5rrrempty^szCustomizablePicklingQueue.emptycsjj_jjjjfdd}|_jrPfdd_nj j _j dkrp_ n(j jj jfdd}|_ dS)Nc szSXdS)Nrr)racquirerecvrreleaserrgetesz4CustomizablePicklingQueue._make_methods..getcs,t}t|jj|jj|jdS)N)rrVrhrrkZ send_bytesgetvalue)rbbuffer)r5rrsendosz5CustomizablePicklingQueue._make_methods..sendc sz|SXdS)Nr)rb)r~ wlock_acquire wlock_releaserrput}sz4CustomizablePicklingQueue._make_methods..put) rjry_recvrmacquirereleaser{rh_sendrkr~rpr)r5r{rr)rxryrzr5r~rrrrqas  z'CustomizablePicklingQueue._make_methods)N) rRrSrTrUr;rsrurwrqrrrrre:s  recs*eZdZdZdfdd ZddZZS) PicklingPoolaPool implementation with customizable pickling reducers. This is useful to control how data is shipped between processes and makes it possible to use shared memory without useless copies induces by the default pickling methods of the original objects passed as arguments to dispatch. `forward_reducers` and `backward_reducers` are expected to be dictionaries with key/values being `(type, callable)` pairs where `callable` is a function that give an instance of `type` will return a tuple `(constructor, tuple_of_objects)` to rebuild an instance out of the pickled `tuple_of_objects` as would return a `__reduce__` method. See the standard library documentation on pickling for more details. Nc sR|dkrt}|dkrt}||_||_t|d}|j|tt|jf|dS)N) processes)dict_forward_reducers_backward_reducersupdatesuperrr;)r5rforward_reducersbackward_reducerskwargspoolargs) __class__rrr;s  zPicklingPool.__init__cCs@t|dt}t||j|_t||j|_|jj|_|jj |_ dS)N_ctx) rr rer_inqueuer _outqueuer _quick_putr _quick_get)r5rrrrr _setup_queuess    zPicklingPool._setup_queues)NNN)rRrSrTrUr;r __classcell__rr)rrrs rcCstjj|rtj|dS)z@Utility function to cleanup a temporary folder if still existingN)r>rIrKshutilrmtree)Z folder_pathrrr delete_folders rc s.eZdZdZd fdd Zfd d ZZS) MemmapingPoolaDProcess pool that shares large arrays to avoid memory copy. This drop-in replacement for `multiprocessing.pool.Pool` makes it possible to work efficiently with shared memory in a numpy context. Existing instances of numpy.memmap are preserved: the child suprocesses will have access to the same shared memory in the original mode except for the 'w+' mode that is automatically transformed as 'r+' to avoid zeroing the original data upon instantiation. Furthermore large arrays from the parent process are automatically dumped to a temporary folder on the filesystem such as child processes to access their content via memmaping (file system backed shared memory). Note: it is important to call the terminate method to collect the temporary folder used by the pool. Parameters ---------- processes: int, optional Number of worker processes running concurrently in the pool. initializer: callable, optional Callable executed on worker process creation. initargs: tuple, optional Arguments passed to the initializer callable. temp_folder: str, optional Folder to be used by the pool for memmaping large arrays for sharing memory with worker processes. If None, this will try in order: - a folder pointed by the JOBLIB_TEMP_FOLDER environment variable, - /dev/shm if the folder exists and is writable: this is a RAMdisk filesystem available by default on modern Linux distributions, - the default system temporary folder that can be overridden with TMP, TMPDIR or TEMP environment variables, typically /tmp under Unix operating systems. max_nbytes int or None, optional, 1e6 by default Threshold on the size of arrays passed to the workers that triggers automated memmory mapping in temp_folder. Use None to disable memmaping of large arrays. forward_reducers: dictionary, optional Reducers used to pickle objects passed from master to worker processes: see below. backward_reducers: dictionary, optional Reducers used to pickle return values from workers back to the master process. verbose: int, optional Make it possible to monitor how the communication of numpy arrays with the subprocess is handled (pickling or memmaping) context_id: int, optional, None by default Set to a value identifying a call context to spare costly hashing of the content of the input arrays when it is safe to assume that each array will not be mutated by the parent process for the duration of the dispatch process. This is the case when using the high level Parallel API. prewarm: bool or str, optional, "auto" by default. If True, force a read on newly memmaped array to make sure that OS pre- cache it in memory. This can be useful to avoid concurrent disk access when the same data array is passed to different worker processes. If "auto" (by default), prewarm is set to True, unless the Linux shared memory partition /dev/shm is available and used as temp_folder. `forward_reducers` and `backward_reducers` are expected to be dictionaries with key/values being `(type, callable)` pairs where `callable` is a function that give an instance of `type` will return a tuple `(constructor, tuple_of_objects)` to rebuild an instance out of the pickled `tuple_of_objects` as would return a `__reduce__` method. See the standard library documentation on pickling for more details. N.ArrFc  s|dkrt}|dkrt}d} dtjt|f} |dkrJtjjdd}|dkrtjjtry0t}tjj || tjjstj d} Wnt k rd}YnX|dkrt j }tjjtjj|}tjj || |_tjfddtdk rV| dkr | } t||||| d} | |tj<t|tj<td||}||tj<t|tj<t|||d }|j| tt|jf|dS) NFzjoblib_memmaping_pool_%d_%dZJOBLIB_TEMP_FOLDERTcstS)N)rr) pool_folderrr(sz(MemmapingPool.__init__..auto)r9r:)rrr)rr>rErFenvironr{rIrKSYSTEM_SHARED_MEM_FSrJr?IOErrortempfile gettempdirabspath expanduserr/atexitr]rr-ndarrayr,r rrrr;)r5rr7r6r8rrr2r9r:rZuse_shared_memZpool_folder_nameZforward_reduce_ndarrayZbackward_reduce_ndarrayr)r)rrr;sV             zMemmapingPool.__init__cstt|jt|jdS)N)rr terminaterr/)r5)rrrrFszMemmapingPool.terminate) NNrrNNrNF)rRrSrTrUr;rrrr)rrrs IDr)4rUrrCr>statrnrGrrrcPicklerr ImportErrorpicklerZrrior_multiprocessing_helpersr r multiprocessing.poolr numpyrZnumpy.lib.stride_tricksr numpy_pickler rhashingrrS_IRUSRS_IWUSRS_IXUSRrArMrrr#r)r,objectr-rVrerrrrrrr sV             $g1M(