o
    Zh-                     @   s`   d dl mZ d dlmZ d dlmZ d dlmZ d dlm	Z	 edddgZ
d	ZG d
d dZdS )    )
namedtuple)CollectReport)Producer)parse_spec_config)report_collection_diffNodePendingnodepending   c                   @   s   e Zd ZdZd!ddZedd Zedd Zed	d
 Zedd Z	dd Z
dd Zd!ddZdd Zdd Zdd Zdd Zdd Zdd Zdd  ZdS )"WorkStealingSchedulinga  Implement 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                 C   sR   t t|| _i | _i | _g | _d | _|d u rtd| _n|j	| _|| _
d | _d S )Nworkstealsched)lenr   numnodesnode2collectionnode2pendingr	   
collectionr   logr   configsteal_requested_from_node)selfr   r    r   P/var/www/html/lang_env/lib/python3.10/site-packages/xdist/scheduler/worksteal.py__init__<   s   
zWorkStealingScheduling.__init__c                 C   s   t | j S )z%A list of all nodes in the scheduler.)listr   keysr   r   r   r   nodesI   s   zWorkStealingScheduling.nodesc                 C   s   t | j| jkS )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``.
        )r   r   r   r   r   r   r   collection_is_completedN   s   z.WorkStealingScheduling.collection_is_completedc                 C   sH   | j sdS | jr
dS | jdurdS | j D ]}t|tkr! dS qdS )z9Return True if all tests have been executed by the nodes.FNT)r   r	   r   r   valuesr   MIN_PENDINGr   r	   r   r   r   tests_finishedX   s   
z%WorkStealingScheduling.tests_finishedc                 C   s(   | j rdS | j D ]}|r dS q
dS )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)r	   r   r   r    r   r   r   has_pendingf   s   z"WorkStealingScheduling.has_pendingc                 C   s   || j vsJ g | j |< dS )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   r   r   r   add_nodeu   s   	zWorkStealingScheduling.add_nodec                 C   sr   || j v sJ | jr0| jsJ || jkr0tt| j }t| j||jj	|jj	}| 
| dS t|| j|< dS )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   r   r   nextiterr   r   r   gatewayidr   r   )r   r   r   Z
other_nodemsgr   r   r   add_node_collection   s   


z*WorkStealingScheduling.add_node_collectionc                 C   s   | j | | |   dS )zpMark test item as completed by node

        This is called by the ``DSession.worker_testreport`` hook.
        N)r   removecheck_schedule)r   r   Z
item_indexdurationr   r   r   mark_test_complete   s   z)WorkStealingScheduling.mark_test_completec                 C   s"   | j d| j| |   d S )Nr   )r	   insertr   indexr+   )r   itemr   r   r   mark_test_pending   s
   
z(WorkStealingScheduling.mark_test_pendingc                    sR   || j u sJ d| _ t|  fdd| j| D | j|< | j| |   dS )zNode returned some test indices back in response to 'steal' command.

        This is called by ``DSession.worker_unscheduled``.
        Nc                    s   g | ]}| vr|qS r   r   ).0iZindices_setr   r   
<listcomp>   s    zIWorkStealingScheduling.remove_pending_tests_from_node.<locals>.<listcomp>)r   setr   r	   extendr+   )r   r   indicesr   r4   r   remove_pending_tests_from_node   s   
z5WorkStealingScheduling.remove_pending_tests_from_nodec           
         s  dd | j  D   fdd}| }|sdS | jr=t|D ]\}}t|| }t| j| }| || q| }|s=dS | jdurDdS t dd dd}|du rTd	}ntd	t|jt }	t	t|jd
 |	}|d	krw|D ]}|
  qndS |j|j| d  |j| _dS )z(Reschedule tests/perform load balancing.c                 S   s    g | ]\}}|j st||qS r   )Zshutting_downr   r2   r   r	   r   r   r   r5      s    z9WorkStealingScheduling.check_schedule.<locals>.<listcomp>c                      s   dd  D S )Nc                 S   s    g | ]\}}t |tk r|qS r   )r   r   r:   r   r   r   r5      s     zQWorkStealingScheduling.check_schedule.<locals>.get_idle_nodes.<locals>.<listcomp>r   r   Znodes_upr   r   get_idle_nodes   s   z=WorkStealingScheduling.check_schedule.<locals>.get_idle_nodesNc                 S   s
   t | jS N)r   r	   )Znode_pendingr   r   r   <lambda>   s   
 z7WorkStealingScheduling.check_schedule.<locals>.<lambda>)keydefaultr   r
   )r   itemsr	   	enumerater   _send_testsr   maxr   minshutdownr   Z
send_steal)
r   r<   Z
idle_nodesr3   r   Znodes_remainingZnum_sendZ
steal_fromZ	num_stealZ	max_stealr   r;   r   r+      s<   


z%WorkStealingScheduling.check_schedulec                 C   sN   | j |}|r| j|d }nd}| j| | j|u r!d| _|   |S )a  Remove 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.

        r   N)r   popr   r	   r7   r   r+   )r   r   r	   Z	crashitemr   r   r   remove_node   s   
z"WorkStealingScheduling.remove_nodec                 C   sx   | j sJ | jdur|   dS |  s| d dS t| j d | _tt	| j| j
dd< | js6dS |   dS )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   )r   r   r+   !_check_nodes_have_same_collectionr   r   r   r   ranger   r	   r   r   r   r   schedule	  s   


zWorkStealingScheduling.schedulec                 C   s@   | j d | }|r| j d |= | j| | || d S d S r=   )r	   r   r7   Zsend_runtest_some)r   r   numZtests_per_noder   r   r   rC   '  s   z"WorkStealingScheduling._send_testsc           	      C   s   t | j }|d \}}d}|dd D ]/\}}t|||jj|jj}|rDd}| | | jdurDt|jjd|g d}| jj	j
|d q|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.
        r   T   NFZfailed)Zlongreprresult)report)r   r   rA   r   r&   r'   r   r   r   hookZpytest_collectreport)	r   Znode_collection_itemsZ
first_nodecolZsame_collectionr   r   r(   repr   r   r   rI   .  s"   

z8WorkStealingScheduling._check_nodes_have_same_collectionr=   )__name__
__module____qualname____doc__r   propertyr   r   r!   r"   r#   r)   r-   r1   r9   r+   rH   rK   rC   rI   r   r   r   r   r      s*    
+

	


7r   N)collectionsr   Z_pytest.runnerr   Zxdist.remoter   Zxdist.workermanager   Zxdist.reportr   r   r   r   r   r   r   r   <module>   s    