i3tSSKJr SSKJr SSKJr SSKrSSKJr SSK J r SSK J r SSK J r "S S 5rg) ) annotations)Sequence)cycleN)Producer)report_collection_diff)parse_tx_spec_config)WorkerControllerc \rSrSrSrSSSjjr\SSj5r\SSj5r\SSj5r \SSj5r SS jr SS jr SSS jjr SS jrSS jrSSSjjrSSjrS SjrS!SjrSSjrSrg)"LoadSchedulingaImplement load scheduling across nodes. This distributes the tests collected across all nodes so each test is run just once. All nodes collect and submit the test suite and when all collections are received it is verified they are identical collections. Then the collection gets divided up in chunks and chunks get submitted to nodes. Whenever a node finishes an item, it calls ``.mark_test_complete()`` which will trigger the scheduler to assign more tests if the number of pending tests for the node falls below a low-watermark. When created, ``numnodes`` defines how many nodes are expected to submit a collection. This is used to know when all nodes have finished collection or how large the chunks need to be created. 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. Nc[[U55Ul0Ul0Ul/UlSUlUc[S5UlOURUlXl URRS5Ul g)N loadsched maxschedchunk) lenrnumnodesnode2collection node2pendingpending collectionrlogrconfig getoptionr)selfrrs S/var/www/html/ai-image-ml/venv/lib/python3.13/site-packages/xdist/scheduler/load.py__init__LoadScheduling.__init__<sj089 BD?A"$ ,0 ; ,DH}}DH ![[22?CcH[URR55$)z%A list of all nodes in the scheduler.)listrkeysrs rnodesLoadScheduling.nodesIsD%%**,--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``. )rrrr!s rcollection_is_completed&LoadScheduling.collection_is_completedNs4''(DMM99rcUR(dgUR(agURR5Hn[ U5S:dM g g)z9Return True if all tests have been executed by the nodes.FT)r%rrvaluesrrrs rtests_finishedLoadScheduling.tests_finishedXsG++ <<((//1G7|q 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)rrr)r*s r has_pendingLoadScheduling.has_pendingds4 <<((//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)rnodes radd_nodeLoadScheduling.add_nodess&,,,,,"$$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) rr%rnextiterrr rgatewayidrr)rr1r other_nodemsgs radd_node_collection"LoadScheduling.add_node_collections(((((  ' '?? "?__,!$t';';'@'@'B"CD ,OOZ1C1C1F1F  %)*%5T"rc^URURU5 URXS9 g)zMark test item as completed by node. The duration it took to execute the item is used as a hint to the scheduler. This is called by the ``DSession.worker_testreport`` hook. )durationN)rremovecheck_schedule)rr1 item_indexr>s rmark_test_complete!LoadScheduling.mark_test_completes. $&&z2 D4rcURceURRSURRU55 URHnUR U5 M g)Nr)rrinsertindexrr@)ritemr1s rmark_test_pending LoadScheduling.mark_test_pendingsY***  OO ! !$ ' %%D    %&rc[5eN)NotImplementedError)rr1indicess rremove_pending_tests_from_node-LoadScheduling.remove_pending_tests_from_nodes "##rcrUR(agUR(a[UR5n[ S[UR5U-S-5n[ S[UR5U-S-5nURUn[U5U:aaUS:a[U5S:agU[U5- n[ S[U5- UR 5nUR U[Xx55 OUR5 URS[UR55 g)aMaybe schedule new items on the node. If there are any globally pending nodes left then this will check if the given node should be given any more tests. The ``duration`` of the last test is optionally used as a heuristic to influence how many tests the node is assigned. Nr(g?znum items waiting for node:) shutting_downrrrmaxr _send_testsminshutdownr) rr1r> num_nodesitems_per_node_minitems_per_node_max node_pendingnum_sendrs rr@LoadScheduling.check_schedules     <<D--.I"%QDLL(9Y(F!(K!L !$QDLL(9Y(F!(K!L ,,T2L< #55s?s<'8A'=-L0AA #AL(9$94;M;M N   s8'CD MMO .DLL0ABrc$URRU5nU(dgURceURURS5nURR U5 URHnUR U5 M 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)rpoprrextendr@)rr1r crashitems r remove_nodeLoadScheduling.remove_nodes~##''-***OOGKKN3  G$%%D    %&rcbUR(deURb%URHnURU5 M gUR 5(dUR S5 g[ [URR555Ul[[UR55URSS&UR(dgURc[UR5Ul [UR5S[UR5-:aV[UR5n[[UR55HnUR[ U5S5 M Ot[UR5[UR 5-n[#US-UR5n[%US5nURHnURX5 M UR(d$URHnUR'5 M gg)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**r(rQ)r%rr"r@!_check_nodes_have_same_collectionrr5r6rr)rangerrrrrTrrUrSrV)rr1r"_items_per_nodenode_chunksizes rscheduleLoadScheduling.schedules++++ ?? & ##D)# 5577 HHB C tD$8$8$?$?$ABCDOO 45 Q     %!$T__!5D  t|| q3tzz?2 2 $**%E3t||,-  ea0.!1S9J9J5KKN 1!4d6H6HIN 3N   6#||  #rcURSUnU(a?URSU2 URURU5 URU5 ggrK)rrr_send_runtest_some)rr1numtests_per_nodes rrTLoadScheduling._send_tests.sQds+  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. rTrdNFfailed)nodeidoutcomelongreprresult)report) rritemsrr7r8rrpytest CollectReporthookpytest_collectreport) rnode_collection_items first_nodecolsame_collectionr1rr:reps rre0LoadScheduling._check_nodes_have_same_collection5s!%T%9%9%?%?%A B/2  5ab 9 D(!3!3!6!6 Cs"' ;;* ..#|| (!$! C KK$$999E!: r)rrrrrrrrrK)rz pytest.ConfigrzProducer | NonereturnNone)rzlist[WorkerController])rbool)r1r rr)r1r rz Sequence[str]rr)r)r1r rAintr>floatrr)rGstrrr)r1r rMz Sequence[int]rr)r1r r>rrr)r1r rz str | None)rr)r1r rnrrr)__name__ __module__ __qualname____firstlineno____doc__rpropertyr"r%r+r.r2r;rBrHrNr@rarjrTre__static_attributes__rrr r s+Z D..::     %6$62?6 60JK 5$ 525 5AF 5  5&$$$  $ CD2> @3rr ) __future__rcollections.abcr itertoolsrry xdist.remoter xdist.reportrxdist.workermanagerr r rrrrs)"$ !/3/AAr