3 d:K@sdZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZmZe jdZeeejd d Zd d ZGd ddeZGdddeZddZGdddeZdS)z$Parallel workflow execution via SGE N)logging) CommandLine)SGELikeBatchManagerBaseloggerznipype.interfacedicCstjtddd|dS)zPNeeded for debugging on big jobs. Once this is fully vetted, it can be removed. z=!rz Nz=!=!=!)rdebugDEBUGGING_PREFIX)messager =/tmp/pip-build-7vycvbft/nipype/nipype/pipeline/plugins/sge.pysge_debug_printsrc@sXeZdZdZddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ dS)QJobInfozInformation about a single job created by OGE/SGE or similar Each job is responsible for knowing it's own refresh state :author Hans J. Johnson cCs>t||_t||_||_tj|_||_t||_||_ dS)N) int_job_numstr_job_queue_state _job_timetime_job_info_creation_time_job_queue_name _job_slots_qsub_command_line)selfjob_numjob_queue_statejob_timejob_queue_name job_slotsqsub_command_liner r r__init__"s    zQJobInfo.__init__cCs.dj|j|j|jtjdtj|j|j|j S)Nz{:<8d}{:12}{:<3d}{:20}{:8}{}z%Y-%m-%dT%H:%M:%S) formatrrrrstrftimegmtimerrr)rr r r__repr__:szQJobInfo.__repr__cCs |jdkS)N initializing)r)rr r ris_initializingDszQJobInfo.is_initializingcCs|jdkp|jdkS)Nzombiefinished)r)rr r r is_zombieGszQJobInfo.is_zombiecCs |jdkS)NZrunning)r)rr r r is_runningJszQJobInfo.is_runningcCs |jdkS)Npending)r)rr r r is_pendingMszQJobInfo.is_pendingcCsXtj|j}|jr*tdj|d}n*|jrP|dkrPtdj||d}nd}|S)z1Return True, unless job is in the "zombie" statuszVDONE! QJobInfo.IsPending found in 'zombie' list, returning False so claiming done! {0}FiXzrFAILURE! QJobInfo.IsPending found long running at {1} seconds'initializing' returning False for to break loop! {0}T)rrr+rr#r()rZ time_diffZis_pending_statusr r ris_job_state_pendingPs zQJobInfo.is_job_state_pendingcCs ||_||_||_t||_dS)N)rrrrr)rrrrr r r r update_infohszQJobInfo.update_infocCs ||_dS)N)r)rZ new_stater r r set_statenszQJobInfo.set_stateN) __name__ __module__ __qualname____doc__r"r&r(r+r,r.r/r0r1r r r rrs rc@sXeZdZdZdddZddZddZed d Zd d Z dddZ ddZ ddZ dS)QstatSubstitutezdA wrapper for Qstat to avoid overloading the SGE/OGS server with rapid continuous qstat requestsqstatcCs(||_||_t|_t|_|jdS)zZ :param qstat_instant_executable: :param qstat_cached_executable: N)_qstat_instant_executable_qstat_cached_executablelist_out_of_scope_jobsdict_task_dictionary_remove_old_jobs)rZqstat_instant_executableZqstat_cached_executabler r rr"vs zQstatSubstitute.__init__cCs|jdddS)zThis is only called during initialization of the function for the purpose of identifying jobs that are not part of this run of nipype. They are jobs that existed prior to starting a new jobs, so they are irrelevant. ZQstatInitializationTN) _run_qstat)rr r rr>sz QstatSubstitute._remove_old_jobscCs(t|}t|dtjdd||j|<dS)z :param taskid: The job id :param qsub_command_line: When initializing, re-use the job_queue_name :return: NONE r'ZnoQueuerN)rrrr=)rtaskidr!r r radd_startup_jobszQstatSubstitute.add_startup_jobc Cstdjtjdd}d}d}x|dkr|d8}ybtj|dtjtjdd t |gtj tj d }|j \}}|j t |rd }td j|PWq"td tj dYq"Xq"W|S)zhrequest definitive job completion information for the current job from the qacct report z6WARNING: CONTACTING qacct for finished jobs, {0}: {1}zVerifying CompletionZqacct Frrz-oz-j)stdoutstderrTzNOTE: qacct for jobs {0}zNOTE: qacct call failed)rr#r subprocessPopenpwdgetpwuidosgetuidrPIPE communicatefindsleep)r@ this_commandZ qacct_retriesZ is_completeprocZ qacct_result_r r r_qacct_verified_completes6     z(QstatSubstitute._qacct_verified_completec Cst}x*|D] }y|jddjdj}Wnd}YnXyt|jddjdj}Wnd}YnX|jd}t|jddjdj}y0|jddjdj}ttjtj |d } Wntd } YnXt|} | |j kr$|j | j || ||t d j |j | |j| q|jj| qWxt|j jD]} | |kr|j| } | rt|j | jd nt d j | |j | |j | jrD|j| } | r|j | jd nt dj | |j | qDWdS)NZ queue_namerunknownslotsrstateZ JB_job_numberZJAT_start_timez%Y-%m-%dT%H:%M:%SgzUpdating job: {0}r)zCERROR: Job not in current parselist, and not in done list {0}: {1}zKERROR: Job not in still in intializing mode, and not in done list {0}: {1})r:getElementsByTagName childNodesdatar getAttributefloatrmktimestrptimer=r0rr#appendr;keysrSr1r() rZ xml_job_listZcurrent_jobs_parsedZcurrent_job_elementrr rrZ job_time_textrtask_idZdictionary_jobZ is_completedr r r_parse_qstat_job_lists`            z%QstatSubstitute._parse_qstat_job_listTcCstdjtj||r |j}n|j}d}x|dkr|d8}yttj|dtjt j ddddgtj tj d }|j \}}t jjj|}|jd } | d} | jd } |j| PWq,tk r} z(d jt| | } t| tjd WYdd} ~ Xq,Xq,WdS)airequest all job information for the current user in xmlformat. See documentation from java documentation: http://arc.liv.ac.uk/SGE/javadocs/jgdi/com/sun/grid/jgdi/monitoring/filter/JobStateFilter.html -s r gives running jobs -s z gives recently completed jobs (**recently** is very ambiguous) -s s suspended jobs z/WARNING: CONTACTING qmaster for jobs, {0}: {1}rBrrz-uz-xmlz-sZpsrz)rCrDZjob_infoZjob_listzQstatParsingError: {0} {1} rEN)rr#rr8r9rFrGrHrIrJrKrLrMxmldomminidom parseStringrXrb ExceptiontyperO)rZreason_for_qstatZ force_instantrPZ qstat_retriesrQZqstat_xml_resultrRrdjobsrunZrunjobsinstZexception_messager r rr?sB       zQstatSubstitute._run_qstatcCs(x"t|jjD]}tt|qWdS)z For debuggingN)r:r=valuesrr)rvvr r rprint_dictionaryIsz QstatSubstitute.print_dictionarycCst|}||jkrF|j|j}|r|jdj|d|j|j}n>|jdj|d||jkrr|j|j}ntdj|d}|stdj|||jkrtdj||jjt||jj|ntdj||S)Nzchecking job pending status {0}FTz9ERROR: Job {0} not in task list, even after forced qstat!z&DONE! Returning for {0} claiming done!z(NOTE: Adding {0} to OutOfScopeJobs list!z=ERROR: Job {0} not in task list, but attempted to be removed!) rr=r/r?r#rr;r_pop)rraZjob_is_pendingr r ris_job_pendingNs2    zQstatSubstitute.is_job_pendingN)r7r7)T) r2r3r4r5r"r>rA staticmethodrSrbr?rnrpr r r rr6rs   'X 1r6cCs|djr|Sd|SdS)zEnsure that qsub job names must begin with a letter. Numbers and punctuation are not allowed. >>> qsub_sanitize_job_name('01') 'J01' >>> qsub_sanitize_job_name('a01') 'a01' rJN)isalpha)Z testjobnamer r rqsub_sanitize_job_nameus rtcs0eZdZdZfddZddZddZZS) SGEPlugina?Execute using SGE (OGE not tested) The plugin_args input to run can be used to control the SGE execution. Currently supported options are: - template : template to use for batch job submission - qsub_args : arguments to be prepended to the job execution script in the qsub call c sd}d|_d|_d}d}d|kr|drd|dkrB|dd|_d|dkr\|dd|_d|dkrt|dd}d|dkr|dd}t|||_tt|j|f|dS) Nz #$ -V #$ -S /bin/sh r7 plugin_argsZ retry_timeoutZ max_triesZqstatProgramPathZqstatCachedProgramPath)_retry_timeout _max_triesr6_refQstatSubstitutesuperrur")rkwargstemplateZ instant_qstatZ cached_qstat) __class__r rr"s        zSGEPlugin.__init__cCs|jjt|S)N)rzrpr)rr@r r r _is_pendingszSGEPlugin._is_pendingcCs(tdttjddd}tjj|}d}|jr2|j}d|jkrnd|jkr\|jdr\|jd}n|d|jd7}d |krd ||f}d |krd ||f}|jrd j ttjd|j|j f}nd j ttjd|j f}|j d }|j d j |}t |}d|||f|j_tj}tjtjdd} t} xy |j} Wnltk r} zN| |jkrp| d7} tj|jn&tj|tdj d|j t| fWYdd} ~ XnXPq.Wtj|dd| jjj dD} t t!j"d| dj#d} |j$|j%| <|j&j'| |j(t)j*d| |j |j(f| S)NZqsubFZ allatonce)environZresource_monitorZterminal_outputZ qsub_args overwriter z-oz%s -o %sz-ez%s -e %s.LOGNAMEz %s -N %s %sCRITICALrr z%Could not submit sge task for node %scSsg|] }|r|qSr r ).0liner r r sz/SGEPlugin._submit_batchtask..z'Your job ([0-9]*) .* has been submittedz*submitted sge task: %d for node %s with %srW)+rr<rJrpathdirnameZ _qsub_argsrwZ _hierarchyjoin_idsplitreversertinputsargsifloggerlevelsetLevelr getLevelNamer:rjrgryrrOrx RuntimeErrorrZruntimerCrrematchgroups output_dirZ_pendingrzrAZcmdlinerr )rZ scriptfilenodecmdrZqsubargsZjobnameZ jobnameitemsZoldleveltriesresultelinesr@r r r_submit_batchtasksh             zSGEPlugin._submit_batchtask)r2r3r4r5r"rr __classcell__r r )r~rrus  ru)r5rJrHrrFrZxml.dom.minidomrcrandomrrZinterfaces.baserbaserr getLoggerrrruniformr robjectrr6rtrur r r rs&   V