diff --git a/setup.py b/setup.py index 9e6d9c0..b085333 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,4 @@ -from setuptools import setup +from setuptools import setup, find_packages setup( name="pytest-xdist", @@ -11,7 +11,7 @@ setup( author_email='pytest-dev@python.org,holger@merlinux.eu', url='https://github.com/pytest-dev/pytest-xdist', platforms=['linux', 'osx', 'win32'], - packages=['xdist'], + packages=find_packages(exclude=['testing', 'example']), entry_points={ 'pytest11': [ 'xdist = xdist.plugin', diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 5de6c50..e05d10f 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -1,6 +1,10 @@ -from xdist.dsession import ( - DSession, LoadScheduling, EachScheduling, report_collection_diff, +from xdist.dsession import DSession +from xdist.report import report_collection_diff +from xdist.scheduler import ( + EachScheduling, + LoadScheduling, ) + import py import pytest import execnet diff --git a/xdist/dsession.py b/xdist/dsession.py index 9ddbc2d..2a5e38c 100644 --- a/xdist/dsession.py +++ b/xdist/dsession.py @@ -1,445 +1,16 @@ -import difflib -import itertools -from _pytest.runner import CollectReport - -import pytest import py -from xdist.slavemanage import NodeManager, parse_spec_config +import pytest + +from xdist.slavemanage import NodeManager +from xdist.scheduler import ( + EachScheduling, + LoadScheduling, +) queue = py.builtin._tryimport('queue', 'Queue') -class EachScheduling: - """Implement scheduling of test items on all nodes - - If a node gets added after the test run is started then it is - assumed to replace a node which got removed before it finished - its collection. In this case it will only be used if a node - with the same spec got removed earlier. - - Any nodes added after the run is started will only get items - assigned if a node with a matching spec was removed before it - finished all its pending items. The new node will then be - assigned the remaining items from the removed node. - """ - - def __init__(self, config, log=None): - self.config = config - self.numnodes = len(parse_spec_config(config)) - self.node2collection = {} - self.node2pending = {} - self._started = [] - self._removed2pending = {} - if log is None: - self.log = py.log.Producer("eachsched") - else: - self.log = log.eachsched - self.collection_is_completed = False - - @property - def nodes(self): - """A list of all nodes in the scheduler.""" - return list(self.node2pending.keys()) - - @property - def tests_finished(self): - if not self.collection_is_completed: - return False - if self._removed2pending: - return False - for pending in self.node2pending.values(): - if len(pending) >= 2: - return False - return True - - @property - def has_pending(self): - """Return 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". - """ - for pending in self.node2pending.values(): - if pending: - return True - return False - - def add_node(self, node): - assert node not in self.node2pending - self.node2pending[node] = [] - - def add_node_collection(self, node, collection): - """Add the collected test items from a node - - Collection is complete once all nodes have submitted their - collection. In this case its pending list is set to an empty - list. When the collection is already completed this - submission is from a node which was restarted to replace a - dead node. In this case we already assign the pending items - here. In either case ``.schedule()`` will instruct the - node to start running the required tests. - """ - assert node in self.node2pending - if not self.collection_is_completed: - self.node2collection[node] = list(collection) - self.node2pending[node] = [] - if len(self.node2collection) >= self.numnodes: - self.collection_is_completed = True - elif self._removed2pending: - for deadnode in self._removed2pending: - if deadnode.gateway.spec == node.gateway.spec: - dead_collection = self.node2collection[deadnode] - if collection != dead_collection: - msg = report_collection_diff(dead_collection, - collection, - deadnode.gateway.id, - node.gateway.id) - self.log(msg) - return - pending = self._removed2pending.pop(deadnode) - self.node2pending[node] = pending - break - - def mark_test_complete(self, node, item_index, duration=0): - self.node2pending[node].remove(item_index) - - def remove_node(self, node): - # KeyError if we didn't get an add_node() yet - pending = self.node2pending.pop(node) - if not pending: - return - crashitem = self.node2collection[node][pending.pop(0)] - if pending: - self._removed2pending[node] = pending - return crashitem - - def schedule(self): - """Schedule the test items on the nodes - - If the node's pending list is empty it is a new node which - needs to run all the tests. If the pending list is already - populated (by ``.add_node_collection()``) then it replaces a - dead node and we only need to run those tests. - """ - assert self.collection_is_completed - for node, pending in self.node2pending.items(): - if node in self._started: - continue - if not pending: - pending[:] = range(len(self.node2collection[node])) - node.send_runtest_all() - else: - node.send_runtest_some(pending) - self._started.append(node) - - -class LoadScheduling: - """Implement 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. - """ - - def __init__(self, config, log=None): - self.numnodes = len(parse_spec_config(config)) - self.node2collection = {} - self.node2pending = {} - self.pending = [] - self.collection = None - if log is None: - self.log = py.log.Producer("loadsched") - else: - self.log = log.loadsched - self.config = config - - @property - def nodes(self): - """A list of all nodes in the scheduler.""" - return list(self.node2pending.keys()) - - @property - def collection_is_completed(self): - """Boolean 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``. - """ - return len(self.node2collection) >= self.numnodes - - @property - def tests_finished(self): - """Return True if all tests have been executed by the nodes.""" - if not self.collection_is_completed: - return False - if self.pending: - return False - for pending in self.node2pending.values(): - if len(pending) >= 2: - return False - return True - - @property - def has_pending(self): - """Return 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". - """ - if self.pending: - return True - for pending in self.node2pending.values(): - if pending: - return True - return False - - def add_node(self, node): - """Add a new node to the scheduler. - - From now on the node will be allocated chunks of tests to - execute. - - Called by the ``DSession.slave_slaveready`` hook when it - successfully bootstraps a new node. - """ - assert node not in self.node2pending - self.node2pending[node] = [] - - def add_node_collection(self, node, collection): - """Add the collected test items from a node - - The collection is stored in the ``.node2collection`` map. - Called by the ``DSession.slave_collectionfinish`` hook. - """ - assert node in self.node2pending - if self.collection_is_completed: - # A new node has been added later, perhaps an original one died. - # .schedule() should have - # been called by now - assert self.collection - if collection != self.collection: - other_node = next(iter(self.node2collection.keys())) - msg = report_collection_diff(self.collection, - collection, - other_node.gateway.id, - node.gateway.id) - self.log(msg) - return - self.node2collection[node] = list(collection) - - def mark_test_complete(self, node, item_index, duration=0): - """Mark 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.slave_testreport`` hook. - """ - self.node2pending[node].remove(item_index) - self.check_schedule(node, duration=duration) - - def check_schedule(self, node, duration=0): - """Maybe 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. - """ - if node.shutting_down: - return - - if self.pending: - # how many nodes do we have? - num_nodes = len(self.node2pending) - # if our node goes below a heuristic minimum, fill it out to - # heuristic maximum - items_per_node_min = max(2, len(self.pending) // num_nodes // 4) - items_per_node_max = max(2, len(self.pending) // num_nodes // 2) - node_pending = self.node2pending[node] - if len(node_pending) < items_per_node_min: - if duration >= 0.1 and len(node_pending) >= 2: - # seems the node is doing long-running tests - # and has enough items to continue - # so let's rather wait with sending new items - return - num_send = items_per_node_max - len(node_pending) - self._send_tests(node, num_send) - self.log("num items waiting for node:", len(self.pending)) - - def remove_node(self, node): - """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.slave_slavefinished`` and - ``DSession.slave_errordown`` hooks. - - Return the item which was being executing while the node - crashed or None if the node has no more pending items. - - """ - pending = self.node2pending.pop(node) - if not pending: - return - - # The node crashed, reassing pending items - crashitem = self.collection[pending.pop(0)] - self.pending.extend(pending) - for node in self.node2pending: - self.check_schedule(node) - return crashitem - - 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 - ``.check_schedule()`` on all nodes so that newly added nodes - will start to be used. - - This is called by the ``DSession.slave_collectionfinish`` hook - if ``.collection_is_completed`` is True. - """ - 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.check_schedule(node) - return - - # XXX allow nodes to have different collections - if not self._check_nodes_have_same_collection(): - self.log('**Different tests collected, aborting run**') - return - - # Collections are identical, create the index of pending items. - self.collection = list(self.node2collection.values())[0] - self.pending[:] = range(len(self.collection)) - if not self.collection: - return - - # Send a batch of tests to run. If we don't have at least two - # tests per node, we have to send them all so that we can send - # shutdown signals and get all nodes working. - initial_batch = max(len(self.pending) // 4, - 2 * len(self.nodes)) - - # distribute tests round-robin up to the batch size - # (or until we run out) - nodes = itertools.cycle(self.nodes) - for i in range(initial_batch): - self._send_tests(next(nodes), 1) - - if not self.pending: - # initial distribution sent all tests, start node shutdown - for node in self.nodes: - node.shutdown() - - def _send_tests(self, node, num): - tests_per_node = self.pending[:num] - if tests_per_node: - del self.pending[:num] - self.node2pending[node].extend(tests_per_node) - node.send_runtest_some(tests_per_node) - - def _check_nodes_have_same_collection(self): - """Return 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. - """ - node_collection_items = list(self.node2collection.items()) - first_node, col = node_collection_items[0] - same_collection = True - for node, collection in node_collection_items[1:]: - msg = report_collection_diff( - col, - collection, - first_node.gateway.id, - node.gateway.id, - ) - if msg: - same_collection = False - self.log(msg) - if self.config is not None: - rep = CollectReport( - node.gateway.id, 'failed', - longrepr=msg, result=[]) - self.config.hook.pytest_collectreport(report=rep) - - return same_collection - - -def report_collection_diff(from_collection, to_collection, from_id, to_id): - """Report the collected test difference between two nodes. - - :returns: detailed message describing the difference between the given - collections, or None if they are equal. - """ - if from_collection == to_collection: - return None - - diff = difflib.unified_diff( - from_collection, - to_collection, - fromfile=from_id, - tofile=to_id, - ) - error_message = py.builtin._totext( - 'Different tests were collected between {from_id} and {to_id}. ' - 'The difference is:\n' - '{diff}' - ).format(from_id=from_id, to_id=to_id, diff='\n'.join(diff)) - msg = "\n".join([x.rstrip() for x in error_message.split("\n")]) - return msg - - class Interrupted(KeyboardInterrupt): """ signals an immediate interruption. """ @@ -457,6 +28,7 @@ class DSession: automatically starts collecting tests. Once tests are collected it will wait for instructions. """ + def __init__(self, config): self.config = config self.log = py.log.Producer("dsession") @@ -521,11 +93,12 @@ class DSession: @pytest.mark.trylast def pytest_xdist_make_scheduler(self, config, log): - dist = config.getvalue("dist") - if dist == "load": - return LoadScheduling(config, log) - elif dist == "each": - return EachScheduling(config, log) + dist = config.getvalue('dist') + schedulers = { + 'each': EachScheduling, + 'load': LoadScheduling, + } + return schedulers[dist](config, log) def pytest_runtestloop(self): self.sched = self.config.hook.pytest_xdist_make_scheduler( @@ -746,7 +319,7 @@ class TerminalDistReporter: return " / ".join(parts) def rewrite(self, line, newline=False): - pline = line + " " * max(self._lastlen-len(line), 0) + pline = line + " " * max(self._lastlen - len(line), 0) if newline: self._lastlen = 0 pline += "\n" diff --git a/xdist/plugin.py b/xdist/plugin.py index b47cbe8..aa20def 100644 --- a/xdist/plugin.py +++ b/xdist/plugin.py @@ -31,11 +31,12 @@ def pytest_addoption(parser): "when crashed (set to zero to disable this feature)") group._addoption( '--dist', metavar="distmode", - action="store", choices=['load', 'each', 'no'], + action="store", choices=['each', 'load', 'no'], dest="dist", default="no", help=("set mode for distributing tests to exec environments.\n\n" - "each: send each test to each available environment.\n\n" - "load: send each test to available environment.\n\n" + "each: send each test to all available environments.\n\n" + "load: load balance by sending any pending test to any" + " available environment.\n\n" "(default) no: run tests inprocess, don't distribute.")) group._addoption( '--tx', dest="tx", action="append", default=[], diff --git a/xdist/report.py b/xdist/report.py new file mode 100644 index 0000000..ae26ff7 --- /dev/null +++ b/xdist/report.py @@ -0,0 +1,26 @@ +import py +from difflib import unified_diff + + +def report_collection_diff(from_collection, to_collection, from_id, to_id): + """Report the collected test difference between two nodes. + + :returns: detailed message describing the difference between the given + collections, or None if they are equal. + """ + if from_collection == to_collection: + return None + + diff = unified_diff( + from_collection, + to_collection, + fromfile=from_id, + tofile=to_id, + ) + error_message = py.builtin._totext( + 'Different tests were collected between {from_id} and {to_id}. ' + 'The difference is:\n' + '{diff}' + ).format(from_id=from_id, to_id=to_id, diff='\n'.join(diff)) + msg = "\n".join([x.rstrip() for x in error_message.split("\n")]) + return msg diff --git a/xdist/scheduler/__init__.py b/xdist/scheduler/__init__.py new file mode 100644 index 0000000..45e632e --- /dev/null +++ b/xdist/scheduler/__init__.py @@ -0,0 +1,2 @@ +from xdist.scheduler.each import EachScheduling # noqa +from xdist.scheduler.load import LoadScheduling # noqa diff --git a/xdist/scheduler/each.py b/xdist/scheduler/each.py new file mode 100644 index 0000000..d455ede --- /dev/null +++ b/xdist/scheduler/each.py @@ -0,0 +1,129 @@ +from py.log import Producer + +from xdist.slavemanage import parse_spec_config +from xdist.report import report_collection_diff + + +class EachScheduling: + """Implement scheduling of test items on all nodes + + If a node gets added after the test run is started then it is + assumed to replace a node which got removed before it finished + its collection. In this case it will only be used if a node + with the same spec got removed earlier. + + Any nodes added after the run is started will only get items + assigned if a node with a matching spec was removed before it + finished all its pending items. The new node will then be + assigned the remaining items from the removed node. + """ + + def __init__(self, config, log=None): + self.config = config + self.numnodes = len(parse_spec_config(config)) + self.node2collection = {} + self.node2pending = {} + self._started = [] + self._removed2pending = {} + if log is None: + self.log = Producer("eachsched") + else: + self.log = log.eachsched + self.collection_is_completed = False + + @property + def nodes(self): + """A list of all nodes in the scheduler.""" + return list(self.node2pending.keys()) + + @property + def tests_finished(self): + if not self.collection_is_completed: + return False + if self._removed2pending: + return False + for pending in self.node2pending.values(): + if len(pending) >= 2: + return False + return True + + @property + def has_pending(self): + """Return 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". + """ + for pending in self.node2pending.values(): + if pending: + return True + return False + + def add_node(self, node): + assert node not in self.node2pending + self.node2pending[node] = [] + + def add_node_collection(self, node, collection): + """Add the collected test items from a node + + Collection is complete once all nodes have submitted their + collection. In this case its pending list is set to an empty + list. When the collection is already completed this + submission is from a node which was restarted to replace a + dead node. In this case we already assign the pending items + here. In either case ``.schedule()`` will instruct the + node to start running the required tests. + """ + assert node in self.node2pending + if not self.collection_is_completed: + self.node2collection[node] = list(collection) + self.node2pending[node] = [] + if len(self.node2collection) >= self.numnodes: + self.collection_is_completed = True + elif self._removed2pending: + for deadnode in self._removed2pending: + if deadnode.gateway.spec == node.gateway.spec: + dead_collection = self.node2collection[deadnode] + if collection != dead_collection: + msg = report_collection_diff(dead_collection, + collection, + deadnode.gateway.id, + node.gateway.id) + self.log(msg) + return + pending = self._removed2pending.pop(deadnode) + self.node2pending[node] = pending + break + + def mark_test_complete(self, node, item_index, duration=0): + self.node2pending[node].remove(item_index) + + def remove_node(self, node): + # KeyError if we didn't get an add_node() yet + pending = self.node2pending.pop(node) + if not pending: + return + crashitem = self.node2collection[node][pending.pop(0)] + if pending: + self._removed2pending[node] = pending + return crashitem + + def schedule(self): + """Schedule the test items on the nodes + + If the node's pending list is empty it is a new node which + needs to run all the tests. If the pending list is already + populated (by ``.add_node_collection()``) then it replaces a + dead node and we only need to run those tests. + """ + assert self.collection_is_completed + for node, pending in self.node2pending.items(): + if node in self._started: + continue + if not pending: + pending[:] = range(len(self.node2collection[node])) + node.send_runtest_all() + else: + node.send_runtest_some(pending) + self._started.append(node) diff --git a/xdist/scheduler/load.py b/xdist/scheduler/load.py new file mode 100644 index 0000000..aabbac1 --- /dev/null +++ b/xdist/scheduler/load.py @@ -0,0 +1,288 @@ +from itertools import cycle + +from py.log import Producer +from _pytest.runner import CollectReport + +from xdist.slavemanage import parse_spec_config +from xdist.report import report_collection_diff + + +class LoadScheduling: + """Implement 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. + """ + + def __init__(self, config, log=None): + self.numnodes = len(parse_spec_config(config)) + self.node2collection = {} + self.node2pending = {} + self.pending = [] + self.collection = None + if log is None: + self.log = Producer("loadsched") + else: + self.log = log.loadsched + self.config = config + + @property + def nodes(self): + """A list of all nodes in the scheduler.""" + return list(self.node2pending.keys()) + + @property + def collection_is_completed(self): + """Boolean 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``. + """ + return len(self.node2collection) >= self.numnodes + + @property + def tests_finished(self): + """Return True if all tests have been executed by the nodes.""" + if not self.collection_is_completed: + return False + if self.pending: + return False + for pending in self.node2pending.values(): + if len(pending) >= 2: + return False + return True + + @property + def has_pending(self): + """Return 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". + """ + if self.pending: + return True + for pending in self.node2pending.values(): + if pending: + return True + return False + + def add_node(self, node): + """Add a new node to the scheduler. + + From now on the node will be allocated chunks of tests to + execute. + + Called by the ``DSession.slave_slaveready`` hook when it + successfully bootstraps a new node. + """ + assert node not in self.node2pending + self.node2pending[node] = [] + + def add_node_collection(self, node, collection): + """Add the collected test items from a node + + The collection is stored in the ``.node2collection`` map. + Called by the ``DSession.slave_collectionfinish`` hook. + """ + assert node in self.node2pending + if self.collection_is_completed: + # A new node has been added later, perhaps an original one died. + # .schedule() should have + # been called by now + assert self.collection + if collection != self.collection: + other_node = next(iter(self.node2collection.keys())) + msg = report_collection_diff(self.collection, + collection, + other_node.gateway.id, + node.gateway.id) + self.log(msg) + return + self.node2collection[node] = list(collection) + + def mark_test_complete(self, node, item_index, duration=0): + """Mark 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.slave_testreport`` hook. + """ + self.node2pending[node].remove(item_index) + self.check_schedule(node, duration=duration) + + def check_schedule(self, node, duration=0): + """Maybe 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. + """ + if node.shutting_down: + return + + if self.pending: + # how many nodes do we have? + num_nodes = len(self.node2pending) + # if our node goes below a heuristic minimum, fill it out to + # heuristic maximum + items_per_node_min = max(2, len(self.pending) // num_nodes // 4) + items_per_node_max = max(2, len(self.pending) // num_nodes // 2) + node_pending = self.node2pending[node] + if len(node_pending) < items_per_node_min: + if duration >= 0.1 and len(node_pending) >= 2: + # seems the node is doing long-running tests + # and has enough items to continue + # so let's rather wait with sending new items + return + num_send = items_per_node_max - len(node_pending) + self._send_tests(node, num_send) + self.log("num items waiting for node:", len(self.pending)) + + def remove_node(self, node): + """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.slave_slavefinished`` and + ``DSession.slave_errordown`` hooks. + + Return the item which was being executing while the node + crashed or None if the node has no more pending items. + + """ + pending = self.node2pending.pop(node) + if not pending: + return + + # The node crashed, reassing pending items + crashitem = self.collection[pending.pop(0)] + self.pending.extend(pending) + for node in self.node2pending: + self.check_schedule(node) + return crashitem + + 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 + ``.check_schedule()`` on all nodes so that newly added nodes + will start to be used. + + This is called by the ``DSession.slave_collectionfinish`` hook + if ``.collection_is_completed`` is True. + """ + 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.check_schedule(node) + return + + # XXX allow nodes to have different collections + if not self._check_nodes_have_same_collection(): + self.log('**Different tests collected, aborting run**') + return + + # Collections are identical, create the index of pending items. + self.collection = list(self.node2collection.values())[0] + self.pending[:] = range(len(self.collection)) + if not self.collection: + return + + # Send a batch of tests to run. If we don't have at least two + # tests per node, we have to send them all so that we can send + # shutdown signals and get all nodes working. + initial_batch = max(len(self.pending) // 4, + 2 * len(self.nodes)) + + # distribute tests round-robin up to the batch size + # (or until we run out) + nodes = cycle(self.nodes) + for i in range(initial_batch): + self._send_tests(next(nodes), 1) + + if not self.pending: + # initial distribution sent all tests, start node shutdown + for node in self.nodes: + node.shutdown() + + def _send_tests(self, node, num): + tests_per_node = self.pending[:num] + if tests_per_node: + del self.pending[:num] + self.node2pending[node].extend(tests_per_node) + node.send_runtest_some(tests_per_node) + + def _check_nodes_have_same_collection(self): + """Return 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. + """ + node_collection_items = list(self.node2collection.items()) + first_node, col = node_collection_items[0] + same_collection = True + for node, collection in node_collection_items[1:]: + msg = report_collection_diff( + col, + collection, + first_node.gateway.id, + node.gateway.id, + ) + if msg: + same_collection = False + self.log(msg) + if self.config is not None: + rep = CollectReport( + node.gateway.id, 'failed', + longrepr=msg, result=[]) + self.config.hook.pytest_collectreport(report=rep) + + return same_collection