[svn r63203] add a method that allows to send an item to multiple nodes

--HG--
branch : trunk

py/test/dist/dsession.py

@@ -170,7 +170,6 @@ class DSession(Session):
             l.remove(node)
             if not l:
                 del self.item2nodes[item]
-
         return pending
 
     def triggertesting(self, colitems):
@@ -183,10 +182,29 @@ class DSession(Session):
                 self.bus.notify("collectionstart", event.CollectionStart(next))
                 self.queueevent("collectionreport", basic_collect_report(next))
         self.senditems(senditems)
+        #self.senditems_each(senditems)
 
     def queueevent(self, eventname, *args, **kwargs):
         self.queue.put((eventname, args, kwargs)) 
 
+    def senditems_each(self, tosend):
+        if not tosend:
+            return 
+        room = self.MAXITEMSPERHOST
+        for node, pending in self.node2pending.items():
+            room = min(self.MAXITEMSPERHOST - len(pending), room)
+        sending = tosend[:room]
+        for node, pending in self.node2pending.items():
+            node.sendlist(sending)
+            pending.extend(sending)
+            for item in sending:
+                self.item2nodes.setdefault(item, []).append(node)
+                self.bus.notify("itemstart", item, node)
+        tosend[:] = tosend[room:]  # update inplace
+        if tosend:
+            # we have some left, give it to the main loop
+            self.queueevent("rescheduleitems", event.RescheduleItems(tosend))
+
     def senditems(self, tosend):
         if not tosend:
             return 
@@ -221,6 +239,7 @@ class DSession(Session):
     def handle_crashitem(self, item, node):
         longrepr = "!!! Node %r crashed during running of test %r" %(node, item)
         rep = event.ItemTestReport(item, when="???", excinfo=longrepr)
+        rep.node = node
         self.bus.notify("itemtestreport", rep)
 
     def setup(self):

py/test/dist/testing/test_dsession.py

@@ -42,6 +42,24 @@ class TestDSession:
         l = session.removenode(node)
         assert not l 
 
+    def test_send_remove_to_two_nodes(self, testdir):
+        item = testdir.getitem("def test_func(): pass")
+        node1 = MockNode()
+        node2 = MockNode()
+        session = DSession(item.config)
+        session.addnode(node1)
+        session.addnode(node2)
+        session.senditems_each([item])
+        assert session.node2pending[node1] == [item]
+        assert session.node2pending[node2] == [item]
+        assert node1 in session.item2nodes[item]
+        assert node2 in session.item2nodes[item]
+        session.removeitem(item, node1)
+        assert session.item2nodes[item] == [node2]
+        session.removeitem(item, node2)
+        assert not session.node2pending[node1] 
+        assert not session.item2nodes
+
     def test_senditems_removeitems(self, testdir):
         item = testdir.getitem("def test_func(): pass")
         node = MockNode()