3 d4 @sdZddlZddlZddlmZmZddlmZddl Z ddl m Z ddl Z ddl mZddlZddlm Z dd lmZd d lmZd d lmZyddlmZWnek rddZYnXe jdZddZddZGdddeZdS)zParallel workflow execution via multiprocessing Support for child processes running as non-daemons based on http://stackoverflow.com/a/8963618/1183453 N)ProcessPoolExecutorwait)format_exception)INFO)deepcopy)logging)get_system_total_memory_gb)MapNode)DistributedPluginBase)indentcCs |s|S|jd}||j|S)z.A textwrap.indent replacement for Python < 3.3T) splitlinesjoin)textprefixZ splittextrC/tmp/pip-build-7vycvbft/nipype/nipype/pipeline/plugins/multiproc.pyrs rznipype.workflowc CsNtdd|d}y|j|d|d<Wn&ttj|d<|j|d<YnX|S)aFunction to execute node.run(), catch and log any errors and return the result dictionary Parameters ---------- node : nipype Node instance the node to run updatehash : boolean flag for updating hash taskid : int an identifier for this task Returns ------- result : dictionary dictionary containing the node runtime results and stats N)result tracebacktaskid) updatehashrr)dictrunrsysexc_infor)noderrrrrrrun_node+srcCstj|dtjd<dS)z0Initializes the environment of the child process1Z NIPYPE_NO_ETN)oschdirenviron)cwdrrrprocess_initializerLs r$cspeZdZdZdfdd ZddZddZd d Zdd d ZddZ ddZ ddZ dddZ dddZ ZS)MultiProcPlugina Execute workflow with multiprocessing, not sending more jobs at once than the system can support. The plugin_args input to run can be used to control the multiprocessing execution and defining the maximum amount of memory and threads that should be used. When those parameters are not specified, the number of threads and memory of the system is used. System consuming nodes should be tagged:: memory_consuming_node.mem_gb = 8 thread_consuming_node.n_procs = 16 The default number of threads and memory are set at node creation, and are 1 and 0.25GB respectively. Currently supported options are: - non_daemon: boolean flag to execute as non-daemon processes - n_procs: maximum number of threads to be executed in parallel - memory_gb: maximum memory (in GB) that can be used at once. - raise_insufficient: raise error if the requested resources for a node over the maximum `n_procs` and/or `memory_gb` (default is ``True``). - scheduler: sort jobs topologically (``'tsort'``, default value) or prioritize jobs by, first, memory consumption and, second, number of threads (``'mem_thread'`` option). - mp_context: name of multiprocessing context to use Nc stt|j|di|_i|_d|_tj|_|j j dt j |_ |j j dtd|_|j j dd|_tjd|j |j|jy.t j|j j d }t|j t|jf|d |_WnDttfk rt|j d |_|jjt|j}t|gd d YnXd|_dS)N) plugin_argsrn_procs memory_gbg?raise_insufficientTz7[MultiProc] Starting (n_procs=%d, mem_gb=%0.2f, cwd=%s) mp_context) max_workersZ initializerZinitargsr*)r+)timeout)superr%__init__ _taskresult _task_obj_taskidr getcwdZ_cwdr&getmp cpu_count processorsr r(r)loggerdebugZ get_contextrr$poolAttributeError TypeErrorsubmitr_stats)selfr&r* result_future) __class__rrr/ss6 zMultiProcPlugin.__init__cCs|j}||j|d<dS)Nr)rr0)r?argsrrrr_async_callbackszMultiProcPlugin._async_callbackcCs |jj|S)N)r0r4)r?rrrr _get_resultszMultiProcPlugin._get_resultcCs |j|=dS)N)r1)r?rrrr _clear_taskszMultiProcPlugin._clear_taskFcCsl|jd7_t|jdddkr(d|j_|jjt|||j}|j|j||j |j<t j d|j |j|jS)Nr terminal_outputstreamZ allatoncez*[MultiProc] Submitted task %s (taskid=%d).) r2getattrZ interfacerFr:r=rZadd_done_callbackrCr1r8r9fullname)r?rrr@rrr _submit_jobs  zMultiProcPlugin._submit_jobcCsg}g}x(|jD]}|j|j|j|jqWtjtj||jkrdtj d|j|j rdt dtjtj||j krtj d|j |j rt ddS)z0Check if any node exeeds the available resourceszASome nodes exceed the total amount of memory available (%0.2fGB).z(Insufficient resources available for jobz7Some nodes demand for more threads than available (%d).N) Znodesappendmem_gbr'npanyarrayr(r8warningr) RuntimeErrorr7)r?graphZ tasks_mem_gbZ tasks_num_thrrrr _prerun_checks" zMultiProcPlugin._prerun_checkcCs|jjdS)N)r:shutdown)r?rrr_postrun_checkszMultiProcPlugin._postrun_checkcCsR|j}|j}x<|D]4\}}|t|j|j|8}|t|j|j|8}qW||fS)z9 Make sure there are resources available )r(r7minprocsrMr')r? running_tasksfree_memory_gbfree_processors_jobidrrr_check_resourcess z MultiProcPlugin._check_resourcescstjjjjdddkj@}jj\}}tjt||j |j f}j |krd}t j tkrfddjD}|rd}|dj|7}t|dd }t jd tjt||j |j ||_ |d ks|dkrt jd d St|tjdkrt jdd Sj|jjdd}tjxR|D]H} tj| tryj| j} WnFtk rttj} j | |d | dddj!| <w8YnX| dkrȈj"| } | sȐq8t#j| j$j } t#j| j%j }||ks| |krt jd| | |q8|| 8}||8}t jdj| j&| | |||dj| <dj!| <j'| |rlq8|sj| j(rt jdj| yj| j)|dWn8tk rttj} j | |d | ddYnXj*| j+|| 7}||7}d _ tjq8j,r0j,j| dj-t.j| |d}|d krhdj| <dj!| <njj/d|| fd _ q8Wd S)zL Sends jobs to workers when system resources are available. r)ZaxisrGcs g|]\}}dj|jqS)z * %s)rXrJ).0r\r])r?rr sz:MultiProcPlugin._send_procs_to_workers..z Currently running:   zi[MultiProc] Running %d tasks, and %d jobs ready. Free memory (GB): %0.2f/%0.2f, Free processors: %d/%d.%sg{Gz?zNo resources availableNzUNo tasks are being run, and no jobs can be submitted to the queue. Potential deadlock scheduler)rd)rr)rFr z-Cannot allocate job %d (%0.2fGB, %d threads).zEAllocating %s ID=%d (%0.2fGB, %d threads). Free: %0.2fGB, %d threads.Tz Running node %s on master thread)rstart)0rNZ flatnonzeroZ proc_doneZdepidxsumZ __array__r^Z pending_taskslenr(r7r>r8levelrrrinfor9 _sort_jobsr&r4gcZcollect isinstancerXr num_subnodes ExceptionrrrZ _clean_queueZ proc_pendingZ_submit_mapnoderWrMr'rJZ_local_hash_checkZrun_without_submittingrZ_task_finished_cbZ_remove_node_dirsZ_status_callbackrKrinsert)r?rrSjobidsrZr[statsZtasks_list_msgrYr]rmrr=Z next_job_gbZ next_job_thtidr)r?r_send_procs_to_workerss                   z&MultiProcPlugin._send_procs_to_workerstsortcs |dkrt|fdddS|S)NZ mem_threadcsj|jj|jfS)N)rXrMr')item)r?rr|sz,MultiProcPlugin._sort_jobs..)key)sorted)r?rprdr)r?rrjxs zMultiProcPlugin._sort_jobs)N)F)FN)rt)__name__ __module__ __qualname____doc__r/rCrDrErKrTrVr^rsrj __classcell__rr)rArr%Rs+  r%) r|r multiprocessingr5concurrent.futuresrrrrrrrrkcopyrnumpyrNrGZutils.profilerr Zenginer baser textwrapr ImportError getLoggerr8rr$r%rrrrs*        !