3 QfX @sdZddlmZddlZddlZddlZddlZddlmZddl Z ddl Z ddl Z ddl Z ddl mZy ddlZWnddlZYnXddlmZedk rddlmZdd lmZdd lmZmZdd lmZmZdd lmZmZdd l m!Z!ddl"m#Z#ddgZ$dZ%dZ&dZ'e(edrDej)j*ddj+p4dZ,ej-e,dZ.ndZ.Gddde/Z0ddZ1ddZ2Gddde3Z4Gd d!d!e/Z5d+d#d$Z6Gd%d&d&e/Z7Gd'd(d(e/Z8Gd)d*d*eZ9dS),z+ Helpers for embarrassingly parallel code. )divisionN)sqrt)Integral)mp) MemmapingPool) ThreadPool) format_excformat_outer_frames)Loggershort_format_time)TransportableException _mk_exception)memstr_to_kbytes) _basestringmultiprocessing threadingZ__JOBLIB_SPAWNED_PARALLEL__g? get_contextZJOBLIB_START_METHOD)methodc@s(eZdZdZddZddZddZdS) BatchedCallszCWrap a sequence of (func, args, kwargs) tuples as a single callablecCst||_t|j|_dS)N)listitemslen_size)selfZiterator_slicerV/home/psgendb/BIRCHDEV/pkg/SPAdes-3.15.4/linux-x86_64/share/spades/joblib3/parallel.py__init__Cs zBatchedCalls.__init__cCsdd|jDS)NcSsg|]\}}}|||qSrr).0funcargskwargsrrr Hsz)BatchedCalls.__call__..)r)rrrr__call__GszBatchedCalls.__call__cCs|jS)N)r)rrrr__len__JszBatchedCalls.__len__N)__name__ __module__ __qualname____doc__rr%r&rrrrr@srcCstdkr dStjS)z Return the number of CPUs. Nr)r cpu_countrrrrr+Qsr+cCs\|sdS|dkrdS|dkr dSdd|d}t||}t|d|}t|t|kS) z Returns False for indices increasingly apart, the distance depending on the value of verbose. We use a lag increasing as the square of index T Frg? rr)rint)indexverboseZscaleZ next_scalerrr_verbosity_filter\s r1c@seZdZdZdS)WorkerInterruptza An exception that is not KeyboardInterrupt to allow subprocesses to be interrupted. N)r'r(r)r*rrrrr2osr2c@s eZdZdZddZddZdS) SafeFunctionz Wraps a function to make it exception with full traceback in their representation. Useful for parallel computing with multiprocessing, for which exceptions cannot be captured. cCs ||_dS)N)r!)rr!rrrr}szSafeFunction.__init__c Osny |j||Stk r&tYnDtj\}}}t|||ddd}t|trZn t||YnXdS)Nr,r)contextZ tb_offset)r!KeyboardInterruptr2sysexc_infor issubclassr )rr"r#Ze_typeZe_valueZe_tbtextrrrr%s    zSafeFunction.__call__N)r'r(r)r*rr%rrrrr3wsr3Tc sF|rtjfdd}ytj|}Wntk r@YnX|S)a!Decorator used to capture the arguments of a function. Pass `check_pickle=False` when: - performing a possibly repeated check is too costly and has been done already once outside of the call to delayed. - when used in conjunction `Parallel(backend='threading')`. cs ||fS)Nr)r"r#)functionrrdelayed_functionsz!delayed..delayed_function)pickledumps functoolswrapsAttributeError)r:Z check_pickler;r)r:rdelayeds   rAc@s eZdZdZddZddZdS)ImmediateComputeBatchzSequential computation of a batch of tasks. This replicates the async computation API but actually does not delay the computations when joblib.Parallel runs in sequential mode. cCs ||_dS)N)results)rbatchrrrrszImmediateComputeBatch.__init__cCs|jS)N)rC)rrrrgetszImmediateComputeBatch.getN)r'r(r)r*rrErrrrrBsrBc@s eZdZdZddZddZdS)BatchCompletionCallBacka_Callback used by joblib.Parallel's multiprocessing backend. This callable is executed by the parent process whenever a worker process has returned the results of a batch of tasks. It is used for progress reporting, to update estimate of the batch processing duration and to schedule the next batch of tasks to be processed. cCs||_||_||_dS)N)dispatch_timestamp batch_sizeparallel)rrGrHrIrrrrsz BatchCompletionCallBack.__init__cCs|jj|j7_tj|j}|jjdkrh|j|jjkrh|jj}|dkrP|}nd|d|}||j_|jj|jjdk r|jj dS)Nautorg?g?) rIn_completed_tasksrHtimerG_effective_batch_size_smoothed_batch_durationprint_progress_original_iterator dispatch_next)routZthis_batch_durationZ old_durationZ new_durationrrrr%s   z BatchCompletionCallBack.__call__N)r'r(r)r*rr%rrrrrFs rFc@seZdZdZd&d d Zd d ZddZddZddZddZ ddZ ddZ ddZ ddZ ddZd d!Zd"d#Zd$d%ZdS)'ParallelaG% Helper class for readable parallel mapping. Parameters ----------- n_jobs: int, default: 1 The maximum number of concurrently running jobs, such as the number of Python worker processes when backend="multiprocessing" or the size of the thread-pool when backend="threading". If -1 all CPUs are used. If 1 is given, no parallel computing code is used at all, which is useful for debugging. For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. Thus for n_jobs = -2, all CPUs but one are used. backend: str or None, default: 'multiprocessing' Specify the parallelization backend implementation. Supported backends are: - "multiprocessing" used by default, can induce some communication and memory overhead when exchanging input and output data with the with the worker Python processes. - "threading" is a very low-overhead backend but it suffers from the Python Global Interpreter Lock if the called function relies a lot on Python objects. "threading" is mostly useful when the execution bottleneck is a compiled extension that explicitly releases the GIL (for instance a Cython loop wrapped in a "with nogil" block or an expensive call to a library such as NumPy). verbose: int, optional The verbosity level: if non zero, progress messages are printed. Above 50, the output is sent to stdout. The frequency of the messages increases with the verbosity level. If it more than 10, all iterations are reported. pre_dispatch: {'all', integer, or expression, as in '3*n_jobs'} The number of batches (of tasks) to be pre-dispatched. Default is '2*n_jobs'. When batch_size="auto" this is reasonable default and the multiprocessing workers shoud never starve. batch_size: int or 'auto', default: 'auto' The number of atomic tasks to dispatch at once to each worker. When individual evaluations are very fast, multiprocessing can be slower than sequential computation because of the overhead. Batching fast computations together can mitigate this. The ``'auto'`` strategy keeps track of the time it takes for a batch to complete, and dynamically adjusts the batch size to keep the time on the order of half a second, using a heuristic. The initial batch size is 1. ``batch_size="auto"`` with ``backend="threading"`` will dispatch batches of a single task at a time as the threading backend has very little overhead and using larger batch size has not proved to bring any gain in that case. 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. Only active when backend="multiprocessing". max_nbytes int, str, or None, optional, 1M by default Threshold on the size of arrays passed to the workers that triggers automated memory mapping in temp_folder. Can be an int in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte. Use None to disable memmaping of large arrays. Only active when backend="multiprocessing". Notes ----- This object uses the multiprocessing module to compute in parallel the application of a function to many different arguments. The main functionality it brings in addition to using the raw multiprocessing API are (see examples for details): * More readable code, in particular since it avoids constructing list of arguments. * Easier debugging: - informative tracebacks even when the error happens on the client side - using 'n_jobs=1' enables to turn off parallel computing for debugging without changing the codepath - early capture of pickling errors * An optional progress meter. * Interruption of multiprocesses jobs with 'Ctrl-C' * Flexible pickling control for the communication to and from the worker processes. * Ability to use shared memory efficiently with worker processes for large numpy-based datastructures. Examples -------- A simple example: >>> from math import sqrt >>> from joblib import Parallel, delayed >>> Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10)) [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0] Reshaping the output when the function has several return values: >>> from math import modf >>> from joblib import Parallel, delayed >>> r = Parallel(n_jobs=1)(delayed(modf)(i/2.) for i in range(10)) >>> res, i = zip(*r) >>> res (0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5) >>> i (0.0, 0.0, 1.0, 1.0, 2.0, 2.0, 3.0, 3.0, 4.0, 4.0) The progress meter: the higher the value of `verbose`, the more messages:: >>> from time import sleep >>> from joblib import Parallel, delayed >>> r = Parallel(n_jobs=2, verbose=5)(delayed(sleep)(.1) for _ in range(10)) #doctest: +SKIP [Parallel(n_jobs=2)]: Done 1 out of 10 | elapsed: 0.1s remaining: 0.9s [Parallel(n_jobs=2)]: Done 3 out of 10 | elapsed: 0.2s remaining: 0.5s [Parallel(n_jobs=2)]: Done 6 out of 10 | elapsed: 0.3s remaining: 0.2s [Parallel(n_jobs=2)]: Done 9 out of 10 | elapsed: 0.5s remaining: 0.1s [Parallel(n_jobs=2)]: Done 10 out of 10 | elapsed: 0.5s finished Traceback example, note how the line of the error is indicated as well as the values of the parameter passed to the function that triggered the exception, even though the traceback happens in the child process:: >>> from heapq import nlargest >>> from joblib import Parallel, delayed >>> Parallel(n_jobs=2)(delayed(nlargest)(2, n) for n in (range(4), 'abcde', 3)) #doctest: +SKIP #... --------------------------------------------------------------------------- Sub-process traceback: --------------------------------------------------------------------------- TypeError Mon Nov 12 11:37:46 2012 PID: 12934 Python 2.7.3: /usr/bin/python ........................................................................... /usr/lib/python2.7/heapq.pyc in nlargest(n=2, iterable=3, key=None) 419 if n >= size: 420 return sorted(iterable, key=key, reverse=True)[:n] 421 422 # When key is none, use simpler decoration 423 if key is None: --> 424 it = izip(iterable, count(0,-1)) # decorate 425 result = _nlargest(n, it) 426 return map(itemgetter(0), result) # undecorate 427 428 # General case, slowest method TypeError: izip argument #1 must support iteration ___________________________________________________________________________ Using pre_dispatch in a producer/consumer situation, where the data is generated on the fly. Note how the producer is first called a 3 times before the parallel loop is initiated, and then called to generate new data on the fly. In this case the total number of iterations cannot be reported in the progress messages:: >>> from math import sqrt >>> from joblib import Parallel, delayed >>> def producer(): ... for i in range(6): ... print('Produced %s' % i) ... yield i >>> out = Parallel(n_jobs=2, verbose=100, pre_dispatch='1.5*n_jobs')( ... delayed(sqrt)(i) for i in producer()) #doctest: +SKIP Produced 0 Produced 1 Produced 2 [Parallel(n_jobs=2)]: Done 1 jobs | elapsed: 0.0s Produced 3 [Parallel(n_jobs=2)]: Done 2 jobs | elapsed: 0.0s Produced 4 [Parallel(n_jobs=2)]: Done 3 jobs | elapsed: 0.0s Produced 5 [Parallel(n_jobs=2)]: Done 4 jobs | elapsed: 0.0s [Parallel(n_jobs=2)]: Done 5 out of 6 | elapsed: 0.0s remaining: 0.0s [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed: 0.0s finished rrr 2 * n_jobsrJN1Mrc Cs||_t|_|dkrd}nt|dr8t|dr8||_d}|tkrPtd|tf||_||_|dksvt|t r~|dkr~||_ n td|||_ ||_ t|t rdt||_n||_||_d|_d|_t|_d |_tj|_dS) NrZPoolLockz'Invalid backend: %s, expected one of %rrJrz8batch_size must be 'auto' or a positive integer, got: %riF)r0DEFAULT_MP_CONTEXT _mp_contexthasattrVALID_BACKENDS ValueErrorbackendn_jobs isinstancerrH pre_dispatch _temp_folderrr _max_nbytes _mmap_mode_pool_outputr_jobs _managed_poolrrW_lock) rr^r]r0r`rH temp_folder max_nbytes mmap_moderrrrs:  zParallel.__init__cCsd|_|j|S)NT)rg_initialize_pool)rrrr __enter__szParallel.__enter__cCs|jd|_dS)NF)_terminate_poolrg)rexc_type exc_value tracebackrrr__exit__szParallel.__exit__cCsN|j}|dkrtdn2tdks(|dkr,dS|dkrJttjd|d}|S)Nrz&n_jobs == 0 in Parallel has no meaningr)r^r\rmaxr+)rr^rrr_effective_n_jobss zParallel._effective_n_jobscCs&|j}tg|_|dkr"d|_n|jdkr8t||_n|jdkrtjjrfd|_t j ddddSt j j dkrd|_t j d dddSttjjtd }|rtd d tjt<tjt|j|j|jtd |jd d d}|jdk r|j|d<t|f||_|jjtt gnt!d|j|S)z?Build a process or thread pool and return the number of workersrNrrzHMultiprocessing-backed parallel loops cannot be nested, setting n_jobs=1) stacklevel MainThreadzVMultiprocessing backed parallel loops cannot be nested below threads, setting n_jobs=1ra%[joblib] Attempting to do parallel computing without protecting your import on a system that does not support forking. To use parallel-computing in a script, you must protect your main loop using "if __name__ == '__main__'". Please see the joblib documentation on Parallel for more information12)rjrkrir0Z context_idr4zUnsupported backend: %s)"rtr exceptionsrdr]rrcurrent_processdaemonwarningswarnrcurrent_threadnamer.osenvironrEJOBLIB_SPAWNED_PROCESS ImportErrorgcZcollectdictrbrcrarsr0rYrextendr5r2r\)rr^Zalready_forkedZpoolargsrrrrlsH        zParallel._initialize_poolcCs@|jdk r<|jj|jjd|_|jdkr|t kr|dkr|d|_}|jdkr|jd||fn|}||krd|_n|j}|j ,t t j ||}|sd S|j|d SWd QRXd S) aTPrefetch the tasks for the next batch and dispatch them. The effective size of the batch is computed here. If there are no more jobs to dispatch, return False, else return True. The iterator consumption and dispatching is protected by the same lock so calling this function should be thread safe. rJrrrrr,z:Batch computation too fast (%.4fs.) Setting batch_size=%d.z:Batch computation too slow (%.2fs.) Setting batch_size=%d.FTN)rHr]rMrNMIN_IDEAL_BATCH_DURATIONr.rsr0rMAX_IDEAL_BATCH_DURATIONrhr itertoolsislicer)riteratorrHZold_batch_sizeZbatch_durationZideal_batch_sizeZtasksrrrrXs<       zParallel.dispatch_one_batchcCsB|js dS|jdkrtjj}ntjj}||}|d||fdS)z=Display the message on stout or stderr depending on verbosityNryz [%s]: %s )r0r6stderrwritestdout)rmsgZmsg_argswriterrrrrs  zParallel._printcCs|js dStj|j}|jrHt|j|jr0dS|jd|jt|fn|j}|j }|dks||d|j }||jd}|d|k}|s||rdS||d|j |d}|jd|d|t|t|fdS)zvDisplay the process of the parallel execution only a fraction of time, controlled by self.verbose. Nz!Done %3i tasks | elapsed: %srrg?z/Done %3i out of %3i | elapsed: %s remaining: %s) r0rLrrPr1rrrKr r_pre_dispatch_amount)r elapsed_timer/Z total_tasksZcursorZ frequencyZ is_last_itemZremaining_timerrrrOs0   zParallel.print_progresscCs t|_x|js t|jdkrt|jdkr:tjdq |j|jjd}WdQRXy|jj |j Wq t |j k r}zrd|_ t|trtddd}d||jf}t|jd}||}|j|j|jr|jWdQRX|WYdd}~Xq Xq WdS)Nrg{Gz?Tr,r)r4Z stack_startzMultiprocessing exception: %s --------------------------------------------------------------------------- Sub-process traceback: --------------------------------------------------------------------------- %s)rrerrrfrLsleeprhrrrEtuplerzrr_r r messageretypernrgrl)rr exceptionZ this_reportZreportZexception_typerrrretrieves,  zParallel.retrievec CsP|jrtdd|_|js$|j}n|j}|jdkrrS)T):r* __future__rrr6rr}mathrr>rLrrnumbersrcPickler<Z_multiprocessing_helpersrpoolrZmultiprocessing.poolr format_stackr r loggerr r Z my_exceptionsr rdiskr_compatrr[rrrrZrrEstriprrrXobjectrr+r1 Exceptionr2r3rArBrFrSrrrrsR            )