a ƹDf7#@sPdZddgZiZZddlmZddlmZddl m Z m Z GdddeZ dS) a This module contains map and pipe interfaces to python's threading module. Pipe methods provided: pipe - blocking communication pipe [returns: value] apipe - asynchronous communication pipe [returns: object] Map methods provided: map - blocking and ordered worker pool [returns: list] imap - non-blocking and ordered worker pool [returns: iterator] uimap - non-blocking and unordered worker pool [returns: iterator] amap - asynchronous worker pool [returns: object] Usage ===== A typical call to a pathos threading map will roughly follow this example: >>> # instantiate and configure the worker pool >>> from pathos.threading import ThreadPool >>> pool = ThreadPool(nodes=4) >>> >>> # do a blocking map on the chosen function >>> print(pool.map(pow, [1,2,3,4], [5,6,7,8])) >>> >>> # do a non-blocking map, then extract the results from the iterator >>> results = pool.imap(pow, [1,2,3,4], [5,6,7,8]) >>> print("...") >>> print(list(results)) >>> >>> # do an asynchronous map, then get the results >>> results = pool.amap(pow, [1,2,3,4], [5,6,7,8]) >>> while not results.ready(): ... time.sleep(5); print(".", end=' ') ... >>> print(results.get()) >>> >>> # do one item at a time, using a pipe >>> print(pool.pipe(pow, 1, 5)) >>> print(pool.pipe(pow, 2, 6)) >>> >>> # do one item at a time, using an asynchronous pipe >>> result1 = pool.apipe(pow, 1, 5) >>> result2 = pool.apipe(pow, 2, 6) >>> print(result1.get()) >>> print(result2.get()) Notes ===== This worker pool leverages the python's multiprocessing.dummy module, and thus has many of the limitations associated with that module. The function f and the sequences in args must be serializable. The maps in this worker pool have full functionality whether run from a script or in the python interpreter, and work reliably for both imported and interactively-defined functions. Unlike python's multiprocessing.dummy module, pathos.threading maps can directly utilize functions that require multiple arguments. ThreadPool _ThreadPool)AbstractWorkerPool)starargs) cpu_countrc@seZdZdZddZejjr,ejjeje_d$ddZddZeZd d Z ej je _d d Z ej je _d dZ ej je _ddZ ej je _ddZ ddZddZddZddZd%ddZddZd d!Zd"d#ZeeeZeeeZeZdS)&rz/ Mapper that leverages python's threading. cOsd|v}t|}d|vr.|s |r.d}t|n4|rR|rBd}t||d|d<n|rb|d|d<d|vrd|vr~d}t||d|d<|dt|_|dd |_|jd ur|j|_||_|d S) a5 NOTE: if number of nodes is not given, will autodetect processors. NOTE: additional keyword input is optional, with: id - identifier for the pool initializer - function that takes no input, called when node is spawned initargs - tuple of args for initializers that have args nodesnthreadsz3got multiple values for keyword argument 'nthreads'z0got multiple values for keyword argument 'nodes'rZ processesz4got multiple values for keyword argument 'processes'idN)len TypeErrorpopr_ThreadPool__nodes_id_kwds_serve)selfargskwdsZhasnodesZarglenmsgr]/nfs/NAS7/SABIOD/METHODE/ermites/ermites_venv/lib/python3.9/site-packages/pathos/threading.py__init__Rs.   zThreadPool.__init__NcCsl|dur|j}t|jd}|r6||jks6|j|jkrh|t|fi|j}||_|j|_|t|j<|S)z4Create a new server if one isn't already initializedN)r _ThreadPool__STATEgetrr_clearr)rr_poolrrrrys zThreadPool._servecCsLt|jd}|rH|j|jkrH|j|jkrH||t|jddS)z!Remove server with matching stateN)rrrr rclosejoinr rrrrrrs zThreadPool._clearcOs>tj||g|Ri||}|jt|t|fi|SN)r_AbstractWorkerPool__maprmapstarziprfrrrrrrr!szThreadPool.mapcOs>tj||g|Ri||}|jt|t|fi|Sr)r_AbstractWorkerPool__imaprimapr"r#r$rrrr'szThreadPool.imapcOs>tj||g|Ri||}|jt|t|fi|Sr)rr&rZimap_unorderedr"r#r$rrruimapszThreadPool.uimapcOs>tj||g|Ri||}|jt|t|fi|Sr)rr rZ map_asyncr"r#r$rrramapszThreadPool.amapcOs|}||||Sr)rapplyr$rrrpipeszThreadPool.pipecOs|}||||Sr)rZ apply_asyncr$rrrapipeszThreadPool.apipecCs|jj|jf}d|S)Nz) __class____name__r)rZmapargsrrr__repr__szThreadPool.__repr__cCs|jS)z'get the number of nodes used in the map)r )rrrrZ __get_nodesszThreadPool.__get_nodescCs||||_dS)z'set the number of nodes used in the mapN)rr )rrrrrZ __set_nodess zThreadPool.__set_nodesFcCszt|jd}|rv|j|jkrv|j|jkrvd}|s@|j|ks@J|t|jfi|j}|j|_|j|_|t|j<|S)zrestart a closed poolNr)rrrr r_staterr)rforcerZRUNrrrrestarts zThreadPool.restartcCs*t|jd}|r&|j|jkr&|dS)zclose the pool to any new jobsN)rrrr rrrrrrszThreadPool.closecCs*t|jd}|r&|j|jkr&|dS)za more abrupt closeN)rrrr terminaterrrrr3szThreadPool.terminatecCs*t|jd}|r&|j|jkr&|dS)z#cleanup the closed worker processesN)rrrr rrrrrrszThreadPool.join)N)F)r. __module__ __qualname____doc__rrrrclearr!r'r(r)r+r,r/Z_ThreadPool__get_nodesZ_ThreadPool__set_nodesr2rr3rpropertyrrrZ __state__rrrrrNs6#        N) r6__all__Z__STATErZpathos.abstract_launcherrZpathos.helpers.mp_helperrr"Zpathos.helpersrrrrrrrs =