i0SSKJr SSKJr SSKJr SSKrSSKJr SSK J r SSK J r SSK J r "S S \5rS r"S S 5rg)) annotations)Sequence) NamedTupleN)Producer)report_collection_diff)parse_tx_spec_config)WorkerControllerc*\rSrSr%S\S'S\S'Srg) NodePendingr nodez list[int]pendingN)__name__ __module__ __qualname____firstlineno____annotations____static_attributes__rX/var/www/html/ai-image-ml/venv/lib/python3.13/site-packages/xdist/scheduler/worksteal.pyr r s  rr c\rSrSrSrSSSjjr\SSj5r\SSj5r\SSj5r \SSj5r SS jr SS jr SSS jjr SS jrSS jrSSjrSSjrSSjrSSjrSSjrSrg) WorkStealingSchedulingaImplement work-stealing scheduling. Initially, tests are distributed evenly among all nodes. When some node completes most of its assigned tests (when only one pending test remains), an attempt is made to reassign ("steal") some tests from other nodes to this node. Attributes:: :numnodes: The expected number of nodes taking part. The actual number of nodes will vary during the scheduler's lifetime as nodes are added by the DSession as they are brought up and removed either because of a dead node or normal shutdown. This number is primarily used to know when the initial collection is completed. :node2collection: Map of nodes and their test collection. All collections should always be identical. :node2pending: Map of nodes and the indices of their pending tests. The indices are an index into ``.pending`` (which is identical to their own collection stored in ``.node2collection``). :collection: The one collection once it is validated to be identical between all the nodes. It is initialised to None until ``.schedule()`` is called. :pending: List of indices of globally pending tests. These are tests which have not yet been allocated to a chunk for a node to process. :log: A py.log.Producer instance. :config: Config object, used for handling hooks. :steal_requested_from_node: The node to which the current "steal" request was sent. ``None`` if there is no request in progress. Only one request can be in progress at any time, the scheduler doesn't send multiple simultaneous requests. Nc[[U55Ul0Ul0Ul/UlSUlUc[S5UlOURUlXl SUl g)Nworkstealsched) lenrnumnodesnode2collection node2pendingr collectionrlogrconfigsteal_requested_from_node)selfr$r#s r__init__WorkStealingScheduling.__init__Cs^089 BD?A"$ ,0 ; 01DH))DH BF&rcH[URR55$)z%A list of all nodes in the scheduler.)listr!keysr&s rnodesWorkStealingScheduling.nodesPsD%%**,--rcF[UR5UR:$)zBoolean indication initial test collection is complete. This is a boolean indicating all initial participating nodes have finished collection. The required number of initial nodes is defined by ``.numnodes``. )rr rr,s rcollection_is_completed.WorkStealingScheduling.collection_is_completedUs4''(DMM99rcUR(dgUR(agURbgURR 5Hn[ U5[ :dM g g)z9Return True if all tests have been executed by the nodes.FT)r0rr%r!valuesr MIN_PENDINGr&rs rtests_finished%WorkStealingScheduling.tests_finished_sV++ <<  ) ) 5((//1G7|{*2rc|UR(agURR5H nU(dM g g)zReturn True if there are pending test items. This indicates that collection has finished and nodes are still processing test items, so this can be thought of as "the scheduler is active". TF)rr!r3r5s r has_pending"WorkStealingScheduling.has_pendingms4 <<((//1Gw2rcDXR;de/URU'g)zAdd a new node to the scheduler. From now on the node will be allocated chunks of tests to execute. Called by the ``DSession.worker_workerready`` hook when it successfully bootstraps a new node. N)r!)r&r s radd_nodeWorkStealingScheduling.add_node|s&,,,,,"$$rcXR;deUR(aUR(deX R:wa}[[ UR R 555n[URX#RRURR5nURU5 g[U5UR U'g)zAdd the collected test items from a node. The collection is stored in the ``.node2collection`` map. Called by the ``DSession.worker_collectionfinish`` hook. N) r!r0r"nextiterr r+rgatewayidr#r*)r&r r" other_nodemsgs radd_node_collection*WorkStealingScheduling.add_node_collections(((((  ' '?? "?__,!$t';';'@'@'B"CD ,OOZ1C1C1F1F  %)*%5T"rc`URURU5 UR5 g)zaMark test item as completed by node. This is called by the ``DSession.worker_testreport`` hook. N)r!removecheck_schedule)r&r item_indexdurations rmark_test_complete)WorkStealingScheduling.mark_test_completes) $&&z2 rcURceURRSURRU55 UR 5 g)Nr)r"rinsertindexrI)r&items rmark_test_pending(WorkStealingScheduling.mark_test_pendingsG***  OO ! !$ '  rcXRLdeSUl[U5nURUVs/sH oDU;dM UPM snURU'URR U5 UR 5 gs snf)zyNode returned some test indices back in response to 'steal' command. This is called by ``DSession.worker_unscheduled``. N)r%setr!rextendrI)r&r indices indices_setis rremove_pending_tests_from_node5WorkStealingScheduling.remove_pending_tests_from_nodes55555)-&'l ((.# .!;2FA.# $ G$  # s BBc:^ URR5VVs/sH#upUR(aM[X5PM% snnm SU 4SjjnU"5nU(dgUR(aZ[ U5H<upQ[ U5U- n[ UR5U-nURX5 M> U"5nU(dgURbg[T SSS9nUcSn OH[S[ UR5[- 5n [[ UR5S-U 5n U S:XaUHnUR5 M gUceURRURU *S5 URUlgs snnf)z(Reschedule tests/perform load balancing.ch>TVVs/sHup[U5[:dMUPM snn$s snnfN)rr4)r rnodes_ups rget_idle_nodes=WorkStealingScheduling.check_schedule..get_idle_nodess(.6Uh]T#g,:TDhU UUs..Nc,[UR5$r^)rr) node_pendings r7WorkStealingScheduling.check_schedule..ss<3G3G/Hr)keydefaultrrreturnzlist[WorkerController])r!items shutting_downr r enumerater _send_testsr%maxr4minshutdownr send_steal) r&r rr` idle_nodesrYnodes_remainingnum_send steal_from num_steal max_stealr_s @rrI%WorkStealingScheduling.check_schedules{"&!2!2!8!8!: !: %% 'K &!:   V$%   <<$Z0"%j/A"5t||,?  01 ()J  ) ) 5  HRV   I As:#5#56DEIC 2 23q8)DI ># # %%%"":#5#5yjk#BC)3&i s FFc,URRU5nU(a.URceURURS5nOSnURR U5 UR ULaSUlUR 5 U$)aRemove a node from the scheduler. This should be called either when the node crashed or at shutdown time. In the former case any pending items assigned to the node will be re-scheduled. Called by the ``DSession.worker_workerfinished`` and ``DSession.worker_errordown`` hooks. Return the item which was being executing while the node crashed or None if the node has no more pending items. Nr)r!popr"rrVr%rI)r&r r crashitems r remove_node"WorkStealingScheduling.remove_nodes##''- ??. .. A7II G$  ) )T 1-1D * rcUR(deURbUR5 gUR5(dUR S5 g[ [ URR555Ul[[UR55URSS&UR(dgUR5 g)a`Initiate distribution of the test collection. Initiate scheduling of the items across the nodes. If this gets called again later it behaves the same as calling ``.check_schedule()`` on all nodes so that newly added nodes will start to be used. This is called by the ``DSession.worker_collectionfinish`` hook if ``.collection_is_completed`` is True. Nz+**Different tests collected, aborting run**) r0r"rI!_check_nodes_have_same_collectionr#r?r@r r3rangerrr,s rscheduleWorkStealingScheduling.schedules++++ ?? &    ! 5577 HHB C tD$8$8$?$?$ABCDOO 45 Q  rcURSUnU(a?URSU2 URURU5 URU5 ggr^)rr!rVsend_runtest_some)r&r numtests_per_nodes rrm"WorkStealingScheduling._send_tests8sQds+  TcT"   d # * *> :  " "> 2 rc[URR55nUSup#SnUSSHupV[X6URR URR 5nU(dMCSnUR U5 URcMe[R"URR SU/S9nURRRUS9 M U$) zReturn True if all nodes have collected the same items. If collections differ, this method returns False while logging the collection differences and posting collection errors to pytest_collectreport hook. rTNFfailed)nodeidoutcomelongreprresult)report) r*r rjrrArBr#r$pytest CollectReporthookpytest_collectreport) r&node_collection_items first_nodecolsame_collectionr r"rDreps rr8WorkStealingScheduling._check_nodes_have_same_collection?s!%T%9%9%?%?%A B/2  5ab 9 D(!3!3!6!6 Cs"' ;;* ..#|| (!$! C KK$$999E!: r)r"r$r#r r!rrr%r^)r$z pytest.Configr#zProducer | NoneriNonerh)ribool)r r rir)r r r"z Sequence[str]rir)r r rJintrKz float | Nonerir)rQstrrir)r r rWz Sequence[int]rir)rir)r r riz str | None)r r rrrir)rrrr__doc__r'propertyr-r0r6r9r<rErLrRrZrIr|rrmrrrrrrrs)V G..::     %6$62?6 60QU$25AM   &69p<<3rr) __future__rcollections.abcrtypingrr xdist.remoter xdist.reportrxdist.workermanagerr r r4rrrrrs<"$ !/3/*  BBr