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

@@ -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
 -------------------------
 

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',

testing/test_dsession.py

@@ -29,15 +29,12 @@ class MockNode:
         self.sent = []
         self.gateway = MockGateway()
 
-    def send_runtest(self, nodeid):
+    def send_runtest_some(self, indices):
-        self.sent.append(nodeid)
+        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"] * (3)
-        col = ["xyz"] * (2*sched.ITEM_CHUNKSIZE +1)
         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]
+        chunkitems = col[:1]
-        assert (sent1 == [0,2] and sent2 == [1,3]) or (
+        assert (sent1 == [0] and sent2 == [1]) or (
-                sent1 == [1,3] and sent2 == [0,2])
+                sent1 == [1] and sent2 == [0])
         assert sched.node2pending[node1] == sent1
         assert sched.node2pending[node2] == sent2
         assert len(sched.pending) == 1

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")

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]

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)
+        if self.pending:
-            node_pending.append(item_index)
+            # how many nodes do we have remaining per node roughly?
-            node.send_runtest(item_index)
+            num_nodes = len(self.node2pending)
-        self.log("items waiting for node: %d" %(len(self.pending)))
+            # if our node goes below a heuristic minimum, fill it out to
-        #self.log("node2pending: %s" %(self.node2pending,))
+            # 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]
+        node_collection_items = list(self.node2collection.items())
-        for node, collection in 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())
+        # how many items per node do we have about?
-        num_available = self.numnodes
+        items_per_node = len(self.collection) // len(self.node2pending)
-        max_one_round = num_available * self.ITEM_CHUNKSIZE - 1
+        # take half of it for initial distribution, at least 1
-        for i, item_index in enumerate(self.pending):
+        node_chunksize = max(items_per_node // 2, 1)
-            nodeindex = i % num_available
+        # and initialize each node with a chunk of tests
-            node, pending = available[nodeindex]
+        for node in self.node2pending:
-            node.send_runtest(item_index)
+            self._send_tests(node, node_chunksize)
-            pending.append(item_index)
-            if i >= max_one_round:
-                break
-        del self.pending[:i + 1]
 
+    #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()

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,))
+            self.log("items to run:", torun)
-            while len(torun) >= 2:
+            while torun:
-                # we store item_index so that we can pick it up from the
+                self.run_tests(torun)
-                # runtest hooks
-                self.run_one_test(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:

xdist/slavemanage.py

@@ -241,8 +241,8 @@ class SlaveController(object):
             self.gateway.exit()
             #del self.gateway
 
-    def send_runtest(self, index):
+    def send_runtest_some(self, indices):
-        self.sendcommand("runtests", indices=[index])
+        self.sendcommand("runtests", indices=indices)
 
     def send_runtest_all(self):
         self.sendcommand("runtests_all",)