o %if;@szddlmZddlmZddlmZddlmZddlZddlm Z ddl m Z ddl m Z dd l mZGd d d ZdS) ) annotations) OrderedDict)NoReturn)SequenceN)Producer)report_collection_diff)parse_spec_config)WorkerControllerc@seZdZdZd?d@d d ZedAd d ZedBddZedBddZedBddZ dCddZ dDddZ dEddZ dFdGd%d&Z dHd*d+ZdId.d/ZdCd0d1ZdJd3d4ZdKd7d8ZdCd9d:ZdLd;d<ZdBd=d>ZdS)MLoadScopeSchedulinga Implement load scheduling across nodes, but grouping test by scope. This distributes the tests collected across all nodes so each test is run just once. All nodes collect and submit the list of tests and when all collections are received it is verified they are identical collections. Then the collection gets divided up in work units, grouped by test scope, and those work units get submitted to nodes. Whenever a node finishes an item, it calls ``.mark_test_complete()`` which will trigger the scheduler to assign more work units 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. 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. :collection: The final list of tests collected by all nodes once it is validated to be identical between all the nodes. It is initialised to None until ``.schedule()`` is called. :workqueue: Ordered dictionary that maps all available scopes with their associated tests (nodeid). Nodeids are in turn associated with their completion status. One entry of the workqueue is called a work unit. In turn, a collection of work unit is called a workload. :: workqueue = { '///test_module.py': { '///test_module.py::test_case1': False, '///test_module.py::test_case2': False, (...) }, (...) } :assigned_work: Ordered dictionary that maps worker nodes with their assigned work units. :: assigned_work = { '': { '///test_module.py': { '///test_module.py::test_case1': False, '///test_module.py::test_case2': False, (...) }, (...) }, (...) } :registered_collections: Ordered dictionary that maps worker nodes with their collection of tests gathered during test discovery. :: registered_collections = { '': [ '///test_module.py::test_case1', '///test_module.py::test_case2', ], (...) } :log: A py.log.Producer instance. :config: Config object, used for handling hooks. Nconfig pytest.ConfiglogProducer | NonereturnNonecCsNtt||_d|_t|_i|_i|_|durtd|_ n|j |_ ||_ dS)Nloadscopesched) lenrnumnodes collectionr workqueue assigned_workregistered_collectionsrr rr )selfr r rR/var/www/html/corbot_env/lib/python3.10/site-packages/xdist/scheduler/loadscope.py__init__]s  zLoadScopeScheduling.__init__list[WorkerController]cCst|jS)z,A list of all active nodes in the scheduler.)listrkeysrrrrnodeslszLoadScopeScheduling.nodesboolcCst|j|jkS)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``. )rrrrrrrcollection_is_completedqsz+LoadScopeScheduling.collection_is_completedcCs<|jsdS|jr dS|jD] }||dkrdSqdS)z9Return True if all tests have been executed by the nodes.FT)r"rrvalues _pending_ofr assigned_unitrrrtests_finished{sz"LoadScopeScheduling.tests_finishedcCs2|jrdS|jD] }||dkrdSq dS)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". TrF)rrr$r%r&rrr has_pendingszLoadScopeScheduling.has_pendingnoder cCs||jvsJi|j|<dS)zAdd a new node to the scheduler. From now on the node will be assigned work units to be executed. Called by the ``DSession.worker_workerready`` hook when it successfully bootstraps a new node. N)rrr*rrradd_nodeszLoadScopeScheduling.add_node str | NonecCsx|j|}||s dS|D]}|D] \}}|s!|}nqqntd|j||jD]}||q2|S)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 hooks: - ``DSession.worker_workerfinished``. - ``DSession.worker_errordown``. Return the item being executed while the node crashed or None if the node has no more pending items. Nz=Unable to identify crashitem on a workload with pending items) rpopr%r$items RuntimeErrorrupdate _reschedule)rr*workload work_unitnodeid completed crashitemrrr remove_nodes$      zLoadScopeScheduling.remove_noder Sequence[str]cCsr||jvsJ|jr0|jsJ||jkr0tt|j}t|j||jj |jj }| |dSt ||j|<dS)zAdd the collected test items from a node. The collection is stored in the ``.registered_collections`` dictionary. Called by the hook: - ``DSession.worker_collectionfinish``. N) rr"rnextiterrrrgatewayidr r)rr*r other_nodemsgrrradd_node_collections    z'LoadScopeScheduling.add_node_collectionr item_indexintdurationfloatcCs8|j||}||}d|j|||<||dS)zuMark test item as completed by node. Called by the hook: - ``DSession.worker_testreport``. TN)r _split_scoperr2)rr*rArCr5scoperrrmark_test_completes z&LoadScopeScheduling.mark_test_completeitemstrrcCtNNotImplementedError)rrHrrrmark_test_pendingsz%LoadScopeScheduling.mark_test_pendingindices Sequence[int]cCrJrKrL)rr*rOrrrremove_pending_tests_from_nodesz2LoadScopeScheduling.remove_pending_tests_from_nodecs`|jsJ|jjdd\}}|j|i}|||<|j|fdd|D}||dS)zAssign a work unit to a node.F)lastcsg|] \}}|s|qSr)index).0r5r6worker_collectionrr s z9LoadScopeScheduling._assign_work_unit..N)rpopitemr setdefaultrr/send_runtest_some)rr*rFr4assigned_to_nodenodeids_indexesrrUr_assign_work_units   z%LoadScopeScheduling._assign_work_unitr5cCs|dddS)avDetermine the scope (grouping) of a nodeid. There are usually 3 cases for a nodeid:: example/loadsuite/test/test_beta.py::test_beta0 example/loadsuite/test/test_delta.py::Delta1::test_delta0 example/loadsuite/epsilon/__init__.py::epsilon.epsilon #. Function in a test module. #. Method of a class in a test module. #. Doctest in a function in a package. This function will group tests with the scope determined by splitting the first ``::`` from the right. That is, classes will be grouped in a single work unit, and functions from a test module will be grouped by their module. In the above example, scopes will be:: example/loadsuite/test/test_beta.py example/loadsuite/test/test_delta.py::Delta1 example/loadsuite/epsilon/__init__.py z::r)rsplit)rr5rrrrEsz LoadScopeScheduling._split_scoper3dict[str, dict[str, bool]]cCstdd|D}|S)z1Return the number of pending tests in a workload.css"|] }t|dVqdS)FN)rr$count)rTrFrrr 6s z2LoadScopeScheduling._pending_of..)sumr$)rr3pendingrrrr%4szLoadScopeScheduling._pending_ofcCsT|jrdS|js|dS|dt|j||j|dkr#dS||dS)zMaybe schedule new items on the node. If there are any globally pending work units left then this will check if the given node should be given any more tests. Nz!Number of units waiting for node:r#) shutting_downrshutdownr rr%rr]r+rrrr29szLoadScopeScheduling._reschedulec Csl|jsJ|jdur|jD]}||q dS|s"|ddSttt|j |_|js3dSi}|jD]}| |}| |i}d||<q8t |dddD] \}}||j|<qTt|jt|j}|dkr|d|d t|D]}|j\} } |d | | qy|jD]}||q|jD]}||q|js|jD]}|qdSdS) aInitiate 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 ``._reschedule()`` on all nodes so that newly added nodes will start to be used. If ``.collection_is_completed`` is True, this is called by the hook: - ``DSession.worker_collectionfinish``. Nz+**Different tests collected, aborting run**FcSst|d S)Nr^)r)rHrrrxsz.LoadScopeScheduling.schedule..)keyrzShutting down z nodeszShutting down unused node )r"rr r2!_check_nodes_have_same_collectionr rr:r;rr$rErYsortedr/rrrangerrXrfr]) rr*unsorted_workqueuer5rFr4nodeids extra_nodes_ unused_nodeassignedrrrscheduleRsH                  zLoadScopeScheduling.schedulec Cst|j}|d\}}d}|ddD],\}}t|||jj|jj}|s'qd}||tj|jjd|gd}|j j j |dq|S) 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. rTr^NFfailed)r5outcomelongreprresult)report) rrr/rr<r=r pytest CollectReportr hookpytest_collectreport) rnode_collection_items first_nodecolsame_collectionr*rr?reprrrris&  z5LoadScopeScheduling._check_nodes_have_same_collectionrK)r r r rrr)rr)rr!)r*r rr)r*r rr-)r*r rr9rr)r)r*r rArBrCrDrr)rHrIrr)r*r rOrPrr)r5rIrrI)r3r`rrB)rr)__name__ __module__ __qualname____doc__rpropertyr r"r(r)r,r8r@rGrNrQr]rEr%r2rrrirrrrr s0 M      )       Cr ) __future__r collectionsrtypingrrrx xdist.remoter xdist.reportrxdist.workermanagerr r rrrrs