def schedule(self):
        """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 ``._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``.
        """
        assert self.collection_is_completed

        # Initial distribution already happened, reschedule on all nodes
        if self.collection is not None:
            for node in self.nodes:
                self._reschedule(node)
            return

        # Check that all nodes collected the same tests
        if not self._check_nodes_have_same_collection():
            self.log("**Different tests collected, aborting run**")
            return

        # Collections are identical, create the final list of items
        self.collection = list(next(iter(self.registered_collections.values())))
        if not self.collection:
            return

        # Determine chunks of work (scopes)
        for nodeid in self.collection:
            scope = self._split_scope(nodeid)
            work_unit = self.workqueue.setdefault(scope, default=OrderedDict())
            work_unit[nodeid] = False

        # Avoid having more workers than work
        extra_nodes = len(self.nodes) - len(self.workqueue)

        if extra_nodes > 0:
            self.log("Shuting down {0} nodes".format(extra_nodes))

            for _ in range(extra_nodes):
                unused_node, assigned = self.assigned_work.popitem(last=True)

                self.log("Shuting down unused node {0}".format(unused_node))
                unused_node.shutdown()

        # Assign initial workload
        for node in self.nodes:
            self._assign_work_unit(node)

        # Ensure nodes start with at least two work units if possible (#277)
        for node in self.nodes:
            self._reschedule(node)

        # Initial distribution sent all tests, start node shutdown
        if not self.workqueue:
            for node in self.nodes:
                node.shutdown()
def schedule ( self ) : assert self . collection_is_completed if self . collection is not None : for node in self . nodes : self . _reschedule ( node ) return if not self . _check_nodes_have_same_collection ( ) : self . log ( "**Different tests collected, aborting run**" ) return self . collection = list ( next ( iter ( self . registered_collections . values ( ) ) ) ) if not self . collection : return for nodeid in self . collection : scope = self . _split_scope ( nodeid ) work_unit = self . workqueue . setdefault ( scope , default = OrderedDict ( ) ) work_unit [ nodeid ] = False extra_nodes = len ( self . nodes ) - len ( self . workqueue ) if extra_nodes > 0 : self . log ( "Shuting down {0} nodes" . format ( extra_nodes ) ) for _ in range ( extra_nodes ) : unused_node , assigned = self . assigned_work . popitem ( last = True ) self . log ( "Shuting down unused node {0}" . format ( unused_node ) ) unused_node . shutdown ( ) for node in self . nodes : self . _assign_work_unit ( node ) for node in self . nodes : self . _reschedule ( node ) if not self . workqueue : for node in self . nodes : node . shutdown ( )
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 ``._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``.