ó gbc@sŽddlZddlmZmZddlmZmZddlmZdefd„ƒYZ de fd„ƒYZ d e fd „ƒYZ dS( i˙˙˙˙N(twarningtinfo(tsleepttime(tTaskPooltJobPoolcBsVeZdZdd„Zd„Zd„Zd„Zdd d„Zd„Z d „Z RS( sm JobPool is dedicated to manage a job queue. These jobs will run in a limited number of thread. g{ŽGáz„?cCsqt|ƒ|_t|j|j|ƒ|_||_tƒ|_tj d|j ƒ|_ |j j ƒt |_dS(sŐ @param count: number of thread dedicated to this JobPool @type count: int @param precision: delay between two check for new job (in second) @type precision: float ttargetN(t JobIteratort _iteratorRt_runJobt _taskPoolt _precisiontsett_toRunt threadingtThreadt_runnert _runnerThreadtstarttFalset _finalyzed(tselftcountt precision((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyt__init__s     cCs@x%|jD]}tdt|ƒƒq Wtdt|ƒƒdS(NsJob %d finnishedsAll jobs in %d JobPool finished(R Rtid(Rtrep((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR scCs|jS(N(R(R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyt _jobIterator%scCsEtƒ|_tdt|ƒƒ|ƒ|_tƒ|_t|_|S(NsJob %d started(RtstartedRRtresulttendedtTruetfinished(Rtjob((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR (s     gđ?cCs˘|j stdƒ‚|jdk rCtdt|ƒƒt|ƒStƒ|_||_||_t|ƒ}t|_|||<|j j |ƒt d|ƒ|S(sb Submit a new job to the JobPool. @param job: the new submited job @type job: Job instance @param priority: priority level of this job (higher is better) @type priority: float @param userid: a user identifier (Default is None) @return: job identifier @rtype: int s$This jobPool does not accept new jobsJob %d was already submittedsJob %d submittedN( RtAssertionErrort submittedtNoneRRRtprioritytuseridR taddR(RR!R%R&ti((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pytsubmit0s         cCs|jjƒt|_dS(sW Indicate to the JobPool, that no new jobs will be submitted. N(RtfinalyzeRR(R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR*Ps cCs|jƒdS(N(R*(R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyt__del__XsN( t__name__t __module__t__doc__RRRR R$R)R*R+(((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyRs     RcBs,eZd„Zd„Zd„Zd„ZRS(cCs%||_t|_tjƒ|_dS(N(t_poolRt _finalyzeRtLockt _nextLock(Rtpool((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR]s  cCs|S(N((R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyt__iter__cscCs t|_dS(s[ Indicate to the JobIterator, that no new jobs will be submitted. N(RR0(R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR*fscCsĺ|jjƒx§|jjs&|j rśd}d}x;|jjD]-}|jƒ}||kr?|}|}q?q?W|dk rŁ|jjj|ƒ|jjƒ|fSt |jj ƒqW|jjƒt dt |jƒƒt ‚dS(s[ @return: the next job to run @rtype: Job instance isNo more jobs in %d JobPoolN(R2tacquireR/R R0R$trunScoretremovetreleaseRR RRt StopIteration(RRtmaxScoretkts((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pytnextns"       (R,R-RR4R*R=(((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR\s   tJobcBs)eZddd„Zd„Zd„ZRS(cOsz|dk st‚||_||_||_t|_t|_d|_d|_ d|_ |dk rv|j |ƒndS(sŮ Create a new job @param pool: the jobpool used to run job. Can be None to not execute the job immediately. @type pool: JobPool instance @param function: the function to run for the job @type function: callable object @param args: parametters for function call @param kwargs: named parametters for function call @precondition: function cannot be None N( R$R"t_argst_kwargst _functionRtrunningR R#R%R&R)(RR3tfunctiontargstkwargs((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyRŠs         cCstƒ|j|jS(sa @return: the score used to ordonnance job in the queue @rtype: C{float} (RR#R%(R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR6§scCs|j|j|jŽS(N(RAR?R@(R((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyt__call__ŻsN(R,R-R$RR6RF(((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyR>ˆs ( RtloggingRRRRtobitools.parallelRtdictRtobjectRR>(((s9lib/python2.7/site-packages/obitools/parallel/jobqueue.pyts T,