From e26d4486fc1991dd6b68f6e8527259c07c659a79 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Mon, 27 Jan 2014 11:37:40 +0100 Subject: [PATCH] send multiple "to test" indices in one network message to a slave and improve heuristics for sending chunks where the chunksize depends on the number of remaining tests rather than fixed numbers. This reduces the number of master -> node messages (but not the reverse direction) --- CHANGELOG | 7 +++++ setup.py | 2 +- testing/test_dsession.py | 16 ++++------- testing/test_remote.py | 2 +- tox.ini | 2 +- xdist/dsession.py | 61 ++++++++++++++++++++++++---------------- xdist/remote.py | 17 ++++------- xdist/slavemanage.py | 4 +-- 8 files changed, 60 insertions(+), 51 deletions(-) diff --git a/CHANGELOG b/CHANGELOG index dc4d27e..6a631f4 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -11,6 +11,13 @@ collection instead of node ids (which are not neccessarily unique for functions parametrized with duplicate values) +- send multiple "to test" indices in one network message to a slave + and improve heuristics for sending chunks where the chunksize + depends on the number of remaining tests rather than fixed numbers. + This reduces the number of master -> node messages (but not the + reverse direction) + + 1.9 ------------------------- diff --git a/setup.py b/setup.py index bfc3e7f..f6c0abe 100644 --- a/setup.py +++ b/setup.py @@ -13,7 +13,7 @@ setup( packages = ['xdist'], entry_points = {'pytest11': ['xdist = xdist.plugin'],}, zip_safe=False, - install_requires = ['execnet>=1.1', 'pytest>=2.3.5'], + install_requires = ['execnet>=1.1', 'pytest>=2.4.2'], classifiers=[ 'Development Status :: 5 - Production/Stable', 'Intended Audience :: Developers', diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 34bfa52..10856e7 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -29,15 +29,12 @@ class MockNode: self.sent = [] self.gateway = MockGateway() - def send_runtest(self, nodeid): - self.sent.append(nodeid) + def send_runtest_some(self, indices): + self.sent.extend(indices) def send_runtest_all(self): self.sent.append("ALL") - def sendlist(self, items): - self.sent.extend(items) - def shutdown(self): self._shutdown=True @@ -117,17 +114,16 @@ class TestLoadScheduling: node2 = MockNode() sched.addnode(node1) sched.addnode(node2) - sched.ITEM_CHUNKSIZE = 2 - col = ["xyz"] * (2*sched.ITEM_CHUNKSIZE +1) + col = ["xyz"] * (3) sched.addnode_collection(node1, col) sched.addnode_collection(node2, col) sched.init_distribute() #assert not sched.tests_finished() sent1 = node1.sent sent2 = node2.sent - chunkitems = col[:sched.ITEM_CHUNKSIZE] - assert (sent1 == [0,2] and sent2 == [1,3]) or ( - sent1 == [1,3] and sent2 == [0,2]) + chunkitems = col[:1] + assert (sent1 == [0] and sent2 == [1]) or ( + sent1 == [1] and sent2 == [0]) assert sched.node2pending[node1] == sent1 assert sched.node2pending[node2] == sent2 assert len(sched.pending) == 1 diff --git a/testing/test_remote.py b/testing/test_remote.py index fe38465..6f3c273 100644 --- a/testing/test_remote.py +++ b/testing/test_remote.py @@ -154,7 +154,7 @@ class TestSlaveInteractor: assert ev.kwargs['topdir'] == slave.testdir.tmpdir ids = ev.kwargs['ids'] assert len(ids) == 1 - slave.sendcommand("runtests", indices=range(len(ids))) + slave.sendcommand("runtests", indices=list(range(len(ids)))) slave.sendcommand("shutdown") ev = slave.popevent("logstart") assert ev.kwargs["nodeid"].endswith("test_func") diff --git a/tox.ini b/tox.ini index 4b3c0fe..6a79a2c 100644 --- a/tox.ini +++ b/tox.ini @@ -3,7 +3,7 @@ envlist=py26,py32,py33,py27,py26,py26-old,py33-old [testenv] changedir=testing -deps=pytest>=2.4.2 +deps=pytest>=2.5.1 commands= py.test --junitxml={envlogdir}/junit-{envname}.xml [] [testenv:py27] diff --git a/xdist/dsession.py b/xdist/dsession.py index b315089..88f3ae7 100644 --- a/xdist/dsession.py +++ b/xdist/dsession.py @@ -59,9 +59,6 @@ class EachScheduling: pending[:] = range(len(self.node2collection[node])) class LoadScheduling: - LOAD_THRESHOLD_NEWITEMS = 5 - ITEM_CHUNKSIZE = 10 - def __init__(self, numnodes, log=None): self.numnodes = numnodes self.node2pending = {} @@ -96,15 +93,24 @@ class LoadScheduling: def remove_item(self, node, item_index): node_pending = self.node2pending[node] - assert item_index in node_pending, (item_index, node_pending) node_pending.remove(item_index) # pre-load items-to-test if the node may become ready - if self.pending and len(node_pending) < self.LOAD_THRESHOLD_NEWITEMS: - item_index = self.pending.pop(0) - node_pending.append(item_index) - node.send_runtest(item_index) - self.log("items waiting for node: %d" %(len(self.pending))) - #self.log("node2pending: %s" %(self.node2pending,)) + + if self.pending: + # how many nodes do we have remaining per node roughly? + num_nodes = len(self.node2pending) + # if our node goes below a heuristic minimum, fill it out to + # heuristic maximum + items_per_node_min = max( + 1, len(self.pending) // num_nodes // 4) + items_per_node_max = max( + 1, len(self.pending) // num_nodes // 2) + if len(node_pending) <= items_per_node_min: + num_send = items_per_node_max - len(node_pending) + 1 + self._send_tests(node, num_send) + + self.log("num items waiting for node:", len(self.pending)) + #self.log("node2pending:", self.node2pending) def remove_node(self, node): pending = self.node2pending.pop(node) @@ -118,8 +124,9 @@ class LoadScheduling: def init_distribute(self): assert self.collection_is_completed # XXX allow nodes to have different collections - first_node, col = list(self.node2collection.items())[0] - for node, collection in self.node2collection.items(): + node_collection_items = list(self.node2collection.items()) + first_node, col = node_collection_items[0] + for node, collection in node_collection_items[1:]: report_collection_diff( col, collection, @@ -130,21 +137,25 @@ class LoadScheduling: # all collections are the same, good. # we now create an index self.collection = col - self.pending = range(len(col)) + self.pending[:] = range(len(col)) if not col: return - available = list(self.node2pending.items()) - num_available = self.numnodes - max_one_round = num_available * self.ITEM_CHUNKSIZE - 1 - for i, item_index in enumerate(self.pending): - nodeindex = i % num_available - node, pending = available[nodeindex] - node.send_runtest(item_index) - pending.append(item_index) - if i >= max_one_round: - break - del self.pending[:i + 1] + # how many items per node do we have about? + items_per_node = len(self.collection) // len(self.node2pending) + # take half of it for initial distribution, at least 1 + node_chunksize = max(items_per_node // 2, 1) + # and initialize each node with a chunk of tests + for node in self.node2pending: + self._send_tests(node, node_chunksize) + #f = open("/tmp/sent", "w") + def _send_tests(self, node, num): + tests_per_node = self.pending[:num] + #print >>self.f, "sent", node, tests_per_node + if tests_per_node: + del self.pending[:num] + self.node2pending[node].extend(tests_per_node) + node.send_runtest_some(tests_per_node) def report_collection_diff(from_collection, to_collection, from_id, to_id): """Report the collected test difference between two nodes. @@ -243,7 +254,7 @@ class DSession: assert callname, kwargs method = "slave_" + callname call = getattr(self, method) - self.log("calling method: %s(**%s)" % (method, kwargs)) + self.log("calling method", method, kwargs) call(**kwargs) if self.sched.tests_finished(): self.triggershutdown() diff --git a/xdist/remote.py b/xdist/remote.py index bfce8f0..6cb3251 100644 --- a/xdist/remote.py +++ b/xdist/remote.py @@ -24,7 +24,7 @@ class SlaveInteractor: def pytest_internalerror(self, excrepr): for line in str(excrepr).split("\n"): - self.log("IERROR> " + line) + self.log("IERROR>", line) def pytest_sessionstart(self, session): self.session = session @@ -45,24 +45,19 @@ class SlaveInteractor: torun = [] while 1: name, kwargs = self.channel.receive() - self.log("received command %s(**%s)" % (name, kwargs)) + self.log("received command", name, kwargs) if name == "runtests": torun.extend(kwargs['indices']) elif name == "runtests_all": torun.extend(range(len(session.items))) - self.log("items to run: %s" % (torun,)) - while len(torun) >= 2: - # we store item_index so that we can pick it up from the - # runtest hooks - self.run_one_test(torun) - + self.log("items to run:", torun) + while torun: + self.run_tests(torun) if name == "shutdown": - while torun: - self.run_one_test(torun) break return True - def run_one_test(self, torun): + def run_tests(self, torun): items = self.session.items self.item_index = torun.pop(0) if torun: diff --git a/xdist/slavemanage.py b/xdist/slavemanage.py index 96f67fa..4f3b8c0 100644 --- a/xdist/slavemanage.py +++ b/xdist/slavemanage.py @@ -241,8 +241,8 @@ class SlaveController(object): self.gateway.exit() #del self.gateway - def send_runtest(self, index): - self.sendcommand("runtests", indices=[index]) + def send_runtest_some(self, indices): + self.sendcommand("runtests", indices=indices) def send_runtest_all(self): self.sendcommand("runtests_all",)