ó ë™xTc@ s‘dZddlmZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z ddlZdefd„ƒYZdefd„ƒYZd efd „ƒYZd efd „ƒYZ d efd„ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZdefd„ƒYZddded„ZdS(sá Generic module to provide parallel job execution on queuing systems Provides drop-in replacement classes to those defined in the multiprocessing module (Queue and Job), with certain restrictions placed by the pickle module iÿÿÿÿ(tdivisionN(tEmptytInstantTimeoutcB seZdZd„ZRS(s Timeout immediately cC s td‚dS(NsNo data found in queue(tQueueEmptyException(tselftwaittime((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pytdelays(t__name__t __module__t__doc__R(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRst TimedTimeoutcB s eZdZd„Zd„ZRS(s Timeout after given time cC s ||_dS(N(t max_delay(RR ((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyt__init__$scC s;||jkr.|j|8_tj|ƒn td‚dS(Ns%No data found in queue within timeout(R ttimetsleepR(RR((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR)s(RRR R R(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR s t NoTimeoutcB seZdZd„ZRS(s No timeout cC stj|ƒdS(N(R R(RR((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR8s(RRR R(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR3stQueuecB sGeZdZdZd„Zd„Zedd„Zd„Z d„Z RS(sÖ Queue object to receive data from jobs running on remote machines Data transfer is achieved via files. It is safe to use any number of Queue objects in the same directory, even with a matching identifier gš™™™™™¹?cC s>d|tjƒt|ƒf|_tjƒ|_g|_dS(Ns%s_%d_%d(tostgetpidtidtroott itertoolstcounttwaiting(Rt identifier((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR Gs"cC s¤|jjƒ}d|j|f}tjj|ƒ s;t‚t|dƒ}tj ||ƒ|j ƒd|j|f}tjj|ƒ st‚tj ||ƒdS(Ns %s.%d.tmptwbs%s.%d( RtnextRRtpathtexiststAssertionErrortopentpickletdumptclosetrename(Rtobjtindexttmp_nametofilet target_name((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pytputNs cC s°|stƒ}n'|dk r0td|ƒ}n tƒ}x9trt|jƒ}|dk r^Pn|jd|jƒq<W|dk s‡t‚t j t |dƒƒ}t j |ƒ|S(NR Rtrb(RtNoneR RtTrueRRtWAITTIMERRtloadRRtremove(Rtblockttimeoutt predicatetfnametdata((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pytget\s       cC sC|js|jƒny|jjdƒSWntk r>dSXdS(Ni(Rt read_waitingtpopt IndexErrorR*(R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRvs    cC sÊg}xŠtjd|jƒD]r}|ddkr9qntjj|ƒ\}}yt|dƒ}Wntk r{qnX|j||fƒqW|jdd„ƒg|D]}|d^q­|_ dS(Ns%s.*iüÿÿÿs.tmpitkeycS s|dS(Ni((tp((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyt”s( tglobRRRtsplitexttintt ValueErrortappendtsortR(RRR2RtexttnumberR9((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR5‚s N( RRR R,R R(R+R*R4RR5(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR=s   tJobcB s}eZdZdZejjdƒZd id„Zd„Z d„Z d„Z d„Z d„Z d „Zd „Zd „ZRS( s Job object to execute function calls on remote machines accessible via queuing systems Data transfer is achieved via files. It is safe to use any number of Job objects in the same directory, even with a matching identifier Restrictions: target has to be pickleable s‰source %s libtbx.python << EOF import pickle ( target, args, kwargs ) = pickle.load( open( "%s.target" ) ) target( *args, **kwargs ) EOF s setpaths.shcC s\d|tjƒt|ƒf|_||_||_||_||_d|_ d|_ dS(Ns%s_%d_%d( RRRtnamettargettargstkwargst qinterfaceR*tprocesstjobid(RRDRERHRFRG((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR ®s"     cC sù|jdk rtd‚n|jƒ|jd|jd|jƒd|jƒƒ}y1tj |dtj dtj dtj ƒ|_Wn/t k r´}tdd j |ƒ|f‚nX|jjj|j|j|jfƒ|jjjƒ|jƒdS( Nsstart called second timeRDtoutterrtstdintstdouttstderrsError while executing: '%s': %st (RIR*t RuntimeErrortwrite_input_dataRHRDtout_fileterr_filet subprocesstPopentPIPEtSTDOUTtOSErrortjoinRMtwritetSCRIPTtSETPATHSR!tparse_process_stdout(Rtcmdte((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pytstart¸s(        &cC sdS(N((R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR^ÙscC s.|jdkrtd‚n|jjƒdkS(Nsjob has not been submitted yet(RIR*RQtpoll(R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pytis_aliveÜs cC s·x|jƒrtjdƒqWtjj|jƒƒret|jƒƒjƒ}|ret |‚qenxK|j ƒ|j ƒ|jƒgD](}tjj|ƒr‡tj |ƒq‡q‡WdS(Ngš™™™™™¹?( RcR RRRRRTRtreadRQt target_fileRSR.(RterrorR2((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRZäs(cC sEt|jƒdƒ}tj|j|j|jf|ƒ|jƒdS(NR(RReRR RERFRGR!(Rtifile((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRRôs"cC s d|jS(Ns %s.target(RD(R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyReûscC s d|jS(Ns%s.out(RD(R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRSscC s d|jS(Ns%s.err(RD(R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRTs((RRR R\tlibtbxtenvt under_buildR]R RaR^RcRZRRReRSRT(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRC˜s   !      tPBSJobcB sGeZdZejdƒZdid„Zd„Zd„Zd„Z RS(s„ Job object to execute function calls on remote machines accessible via Portable Batch System (PBS) Data transfer is achieved via files. It is safe to use any number of Job objects in the same directory, even with a matching identifier If PBS support for 'wait for job to finish' is available, this class can be merged into Job Restrictions: target has to be picklable sjob_state\s*=\s*(\w+)cC sSd|tjƒt|ƒf|_||_||_||_||_d|_ dS(Ns%s_%d_%d( RRRRDRERFRGRHR*RJ(RRDRERHRFRG((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR s "    cC s|jdk rtd‚n|jƒ|jd|jd|jƒd|jƒƒ}y.tj |dtj dtj dtj ƒ}Wn/t k r±}tdd j |ƒ|f‚nX|j d |j|j|jfƒ\}}|rït|‚n|dk st‚|jƒ|_dS( Nsstart called second timeRDRKRLRMRNROsError while executing: '%s': %sRPtinput(RJR*RQRRRHRDRSRTRURVRWRYRZt communicateR\R]Rtstrip(RR_RIR`RKRL((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRa$s.         " cC s|jƒdkS(NtC(t job_status(R((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRcJscC s®|jdkrtd‚ntjdd|jfdtjdtjƒ}|jƒ\}}|rytd|j|f‚n|jj|ƒ}|s¡td|‚n|j dƒS( Nsjob has not been submitted yettqstats-fRNROsJobid: %s PBS error: %ssIncorrect qstat output: %si( RJR*RQRURVRWRmtREGEXtsearchtgroup(RRIRKRLtm((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRpOs   (( RRR tretcompileRrR RaRcRp(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRk s   & tSGEJobcB seZd„ZRS(cC sS|jjjƒ}x:|D]2}|jdƒrt|jƒdƒ|_PqqWdS(NsYour jobi(RIRNt readlinest startswithR=tsplitRJ(Rtqsub_outtline((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR^gs  (RRR^(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRxfstqueue_interfacecB seZdZed„ZRS(cC sg||_|dkr<|jdk s*t‚|jg|_n't|tƒrZ|g|_n ||_dS(N(tasyncR*tCOMMANDRtcommandt isinstancetstr(RRt asynchronous((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR ps  N(RRR*R€tFalseR (((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR~nst sge_interfacecB seZdZdZd„ZRS(s( Interface to Sun Grid Engine (SGE) tqsubc C sK|jdddd|d|d|g }|jsG|jddgƒn|S( Ns-Ss/bin/shs-cwds-Ns-os-es-syncty(RRtextend(RRDRKRLR_((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyt__call__‚s   (RRR R€RŠ(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR†{st lsf_interfacecB seZdZdZd„ZRS(s. Interface to Load Sharing Facility (LSF) tbsubcC s<|jd|d|d|g}|js8|jdƒn|S(Ns-Js-os-es-K(RRR?(RRDRKRLR_((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRŠ’s (RRR R€RŠ(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR‹‹st pbs_interfacecB seZdZdZd„ZRS(s. Interface to Portable Batch System (PBS) R‡c C s#|jddd|d|d|gS(Ns-dt.s-Ns-os-e(R(RRDRKRL((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyRŠ s(RRR R€RŠ(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR™st libtbx_pythontsgec C s£t|dƒst‚|dkrCtd|d|dt||ƒƒS|dkrqtd|d|dt||ƒƒS|dkrŸtd|d|dt||ƒƒSdS(NRŠRRERDRHtpbstlsf(thasattrRRxR†RkRRCR‹(RERDtplatformRR„((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyR‡¤s    (R t __future__RtcPickleRRURR RR;RvRRRtlibtbx.load_envRhtobjectRR RRCRkRxR~R†R‹RR*R+R‡(((sQ/public2/ftp/pub/ccp4/6.5/ccp4-6.5/lib/py2/libtbx/queuing_system_utils/generic.pyts2          [r\