ph+SrSSKrSSKrSSKrSSKrSSKJrJr SSKJ r SSK J r SSK J r \R"\5rSr"S S 5rSr\R&R(r"S S 5r"S S\R.5rSrg)z% Utilities for parallel computation. N)chainislice) Traceback)tqdm)configc[R"5n[RS:Xa [ S5e[RU:a'[ R S[RU5 U$[RS:a8U[R-S-nUS::a[ SRX55eU$[R$)z2Return the number of processes to use in parallel.rz,Invalid NUMBER_OF_CORES; value may not be 0.z&Requesting %s cores; only %s availablez[Invalid NUMBER_OF_CORES; negative value is too negative: requesting {} cores, {} available.)multiprocessing cpu_countrNUMBER_OF_CORES ValueErrorloginfoformat)r nums P/home/james-whalen/.local/lib/python3.13/site-packages/pyphi/compute/parallel.pyget_num_processesrs))+I " :< < ) 9'' 4 !&00014 !855;VC5KM M  ! !!c$\rSrSrSrSrSrSrg)ExceptionWrapper1zA picklable wrapper suitable for passing exception tracebacks through instances of ``multiprocessing.Queue``. Args: exception (Exception): The exception to wrap. c`Xl[R"5u p#[U5Ulg)N) exceptionsysexc_infortb)selfr_rs r__init__ExceptionWrapper.__init__9s!"<<>1B-rchURRURR55e)zRe-raise the exception.)rwith_tracebackr as_traceback)rs rreraiseExceptionWrapper.reraise>s$nn++DGG,@,@,BCCr)rrN)__name__ __module__ __qualname____firstlineno____doc__r r%__static_attributes__rrrr1s Drrc\rSrSrSrSrSrSr\S5r Sr Sr S r \S 5r S rS rS rSrSrSrSSjrSrg) MapReduceGaAn engine for doing heavy computations over an iterable. This is similar to ``multiprocessing.Pool``, but allows computations to shortcircuit, and supports both parallel and sequential computations. Args: iterable (Iterable): A collection of objects to perform a computation over. *context: Any additional data necessary to complete the computation. Any subclass of ``MapReduce`` must implement three methods:: - ``empty_result``, - ``compute``, (map), and - ``process_result`` (reduce). The engine includes a builtin ``tqdm`` progress bar; this can be disabled by setting ``pyphi.config.PROGRESS_BARS`` to ``False``. Parallel operations start a daemon thread which handles log messages sent from worker processes. Subprocesses spawned by ``MapReduce`` cannot spawn more subprocesses; be aware of this when composing nested computations. This is not an issue in practice because it is typically most efficient to only parallelize the top level computation. cXlX lSUlUR5UlSUlSUlSUlSUlSUl SUl SUl SUl g)NF) iterablecontextdoneinit_progress_barprogress task_queue result_queue log_queue log_thread processes num_processestaskscomplete)rr3r4s rr MapReduce.__init__gs_   ..0  !  rc[e)z>Return the default result with which to begin the computation.NotImplementedError)rr4s r empty_resultMapReduce.empty_resultws!!rc[e)z0Map over a single object from ``self.iterable``.rB)objr4s rcomputeMapReduce.compute{s "!rc[e)a\Reduce handler. Every time a new result is generated by ``compute``, this method is called with the result and the previous (accumulated) result. This method compares or collates these two values, returning the new result. Setting ``self.done`` to ``True`` in this method will abort the remainder of the computation, returning this final result. rB)r new_result old_results rprocess_resultMapReduce.process_results "!rFc[R=(d [R(+nU(aSnO/[ UR 5Ul[ UR 5n[X!SURS9$)z%Initialize and return a progress bar.NF)totaldisableleavedesc) r/_forkedr PROGRESS_BARSlistr3lenr description)rrQrPs rr6MapReduce.init_progress_bars_##?6+?+?'? E /DM &E%))+ +rc4S[l[RS5 [ U5 [ UR [5HsnUR5(a[RS5 OH[RSU5 URU"U/UQ765 [RSU5 Mu UR[5 [RS5 g![a$nUR[U55 SnAgSnAff=f)z5A worker process, run by ``multiprocessing.Process``.TzWorker process starting...z&Worker received signal - exiting earlyz Worker got %szWorker finished %szWorker process exitingN) r/rTrdebugconfigure_worker_loggingiterget POISON_PILLis_setput Exceptionr)rHr8r9r:r?r4rGes rworkerMapReduce.workers 2 $I  II2 3 $Y /JNNK8??$$IIFG /3/  !7w!78 .49   [ ) II. / 2   -a0 1 1 2sC&C)) D3DDc [5Ul[R"[S9Ul[R"5Ul[R"5Ul[R"5Ul URUR UR URUR4UR-n[UR5Vs/sH#n[R"URUSS9PM% snUlURHnUR!5 M [#UR5UlUR$R!5 UR'5 gs snf)zIInitialize all queues and start the worker processes and the log thread. )maxsizeT)targetargsdaemonN)rr=r Queue Q_MAX_SIZEr8r9r:Eventr?rHr4rangeProcessrdr<start LogThreadr;initialize_tasks)rriiprocesss rstart_parallelMapReduce.start_parallels/0)// C+113(..0(--/  doot/@/@ /15>4--.0.  # #4;;T$ O.0~~G MMO&$DNN3  0s*E2c[UR[/UR-5Ul[ UR[ 5H4n[RSU5 URRU5 M6 g)zLoad the input queue to capacity. Overfilling causes a deadlock when `queue.put` blocks when full, so further tasks are enqueued as results are returned. Putting %s on queueN) rr3r_r=r>rrlrr[r8rartasks rrrMapReduce.initialize_taskssY4==;-$:L:L*LM 4::z2D II+T 2 OO   %3rc[UR5n[RSU5 URR U5 g![ a gf=f)z0Enqueue the next task, if there are any waiting.rxN)nextr>rr[r8ra StopIterationrys rmaybe_put_taskMapReduce.maybe_put_tasksL & #D II+T 2 OO   %    sA AAc0UR5 UR"UR6nURS:aURR 5nUR 5 U[LaU=RS-slO}[U[5(aUR5 OWURX!5nURRS5 UR(aURR!5 URS:aMUR#5 [&R)S5 URR+5 U$![$a ef=f![&R)S5 URR+5 f=f)zsPerform the computation in parallel, reading results from the output queue and passing them to ``process_result``. rr zRemoving progress bar)rurDr4r=r9r^rr_ isinstancerr%rMr7updater5r?setfinish_parallelrbrr[close)rresultrs r run_parallelMapReduce.run_parallels8 "    !&& 5F$$q(%%))+##% #&&!+&#344IIK"00;FMM((+yy ))+!$$q($  " II- . MM   !     II- . MM   !sDEE E!!E$$1FcURHnUR5 M [RS5 URR [ 5 URR5 URR5 [RS5 URR5 URR5 g)zOrderly shutdown of workers.zJoining log threadzClosing queuesN) r<joinrr[r:rar_r;rr8r9)rrts rrMapReduce.finish_parallels~~G LLN& &' ;'   "#  !rcUR"UR6nURH_nUR"U/URQ76nUR X15nUR R S5 UR(dM_ O UR R5 U$![anUeSnAff=f!UR R5 f=f)z]Perform the computation sequentially, only holding two computed objects in memory at a time. r N) rDr4r3rHrMr7rr5rbr)rrrGrrcs rrun_sequentialMapReduce.run_sequentials "&& 5F}}LL4t||4,,Q7 $$Q'999% MM   !  G  MM   !s<BB'B'B;B' B;' B81B33B88B;;CcPU(aUR5$UR5$)zPerform the computation. Keyword Args: parallel (boolean): If True, run the computation in parallel. Otherwise, operate sequentially. )rr)rparallels rrun MapReduce.run5s% $$& &""$$r) r?r4r5r3r:r;r=r<r7r9r8r>N)T)r'r(r)r*r+rXr rD staticmethodrHrMrTr6rdrurrrrrrrr,r-rrr/r/Gsu:K """ "G+ 220 8 &&"H" , %rr/c2^\rSrSrSrU4SjrSrSrU=r$)rqiBzThread which handles log records sent from ``MapReduce`` processes. It listens to an instance of ``multiprocessing.Queue``, rewriting log messages to the PyPhi log handler. c<>Xl[TU] 5 SUlg)NT)qsuperr rj)rr __class__s rr LogThread.__init__Is  rc[RS5 URR5nU[LaO3[ R "UR5nURU5 MW[RS5 g)NzLog thread startedzLog thread exiting) rr[rr^r_logging getLoggernamehandle)rrecordloggers rr LogThread.runNs` &'VVZZ\F$&&v{{3F MM& !  &'r)rjr) r'r(r)r*r+r rr, __classcell__)rs@rrqrqBs  ((rrqc^[RRSSSSUS.0SS/S.S.5 g ) zrs  # !"2DD$  ( ( 6 6 w%w%v(   (.r