From c90c4a254c956902d6500514a7fbe3b341147827 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Mon, 27 Sep 2010 16:13:55 +0200 Subject: [PATCH] make rsyncing work again, some streamlining in termination detection --- testing/test_dsession.py | 12 +++---- xdist/dsession.py | 41 ++++++++++++---------- xdist/remote.py | 73 +++++++++++++++++++++++----------------- 3 files changed, 71 insertions(+), 55 deletions(-) diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 9dc18b6..748cf65 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -37,11 +37,11 @@ class TestLoadScheduling: sched.addnode(node1) sched.addnode(node2) collection = ["a.py::test_1", "a.py::test_2"] - assert not sched.collection_is_completed() + assert not sched.collection_is_completed sched.addnode_collection(node1, collection) - assert not sched.collection_is_completed() + assert not sched.collection_is_completed sched.addnode_collection(node2, collection) - assert sched.collection_is_completed() + assert sched.collection_is_completed assert sched.node2collection[node1] == collection assert sched.node2collection[node2] == collection sched.init_distribute() @@ -87,12 +87,12 @@ class TestLoadScheduling: sched.addnode(node) collection = ["test_file.py::test_func"] sched.addnode_collection(node, collection) - assert sched.collection_is_completed() + assert sched.collection_is_completed sched.init_distribute() sched.triggertesting() assert not sched.pending - sched.remove_node(node) - assert sched.pending == collection + crashitem = sched.remove_node(node) + assert crashitem == collection[0] class TestDSession: diff --git a/xdist/dsession.py b/xdist/dsession.py index b2c47bf..3b812c2 100644 --- a/xdist/dsession.py +++ b/xdist/dsession.py @@ -2,7 +2,6 @@ import py import sys from xdist.nodemanage import NodeManager from py._test import session -import kwlog queue = py.builtin._tryimport('queue', 'Queue') def dsession_main(config): @@ -24,9 +23,10 @@ class LoadScheduling: self.node2collection = {} self.pending = [] if log is None: - self.log = kwlog.Producer("loadsched") + self.log = py.log.Producer("loadsched") else: self.log = log.loadsched + self.collection_is_completed = False def hasnodes(self): return bool(self.node2pending) @@ -34,11 +34,8 @@ class LoadScheduling: def addnode(self, node): self.node2pending[node] = [] - def collection_is_completed(self): - return len(self.node2collection) == self.numnodes - def tests_finished(self): - if not self.collection_is_completed() or self.pending: + if not self.collection_is_completed or self.pending: return False for items in self.node2pending.values(): if items: @@ -46,8 +43,11 @@ class LoadScheduling: return True def addnode_collection(self, node, collection): + assert not self.collection_is_completed assert node in self.node2pending self.node2collection[node] = list(collection) + if len(self.node2collection) >= self.numnodes: + self.collection_is_completed = True def remove_item(self, node, item): if item not in self.item2nodes: @@ -65,6 +65,9 @@ class LoadScheduling: pending.append(item) self.item2nodes.setdefault(item, []).append(node) node.send_runtest(item) + #self.log("items waiting for node: %d" %(len(self.pending))) + #self.log("item2pending still executing: %s" %(self.item2nodes,)) + #self.log("node2pending: %s" %(self.node2pending,)) def remove_node(self, node): pending = self.node2pending.pop(node) @@ -81,7 +84,7 @@ class LoadScheduling: return crashitem def init_distribute(self): - assert self.collection_is_completed() + assert self.collection_is_completed assert not hasattr(self, 'item2nodes') self.item2nodes = {} # XXX allow nodes to have different collections @@ -111,7 +114,7 @@ class LoadScheduling: break del self.pending[:i+1] if self.pending: - self.log.debug("triggertesting remaining:", len(self.pending)) + self.log("triggertesting remaining:", len(self.pending)) class Interrupted(KeyboardInterrupt): """ signals an immediate interruption. """ @@ -119,9 +122,9 @@ class Interrupted(KeyboardInterrupt): class DSession: def __init__(self, config): self.config = config - self.log = kwlog.Producer("dsession") - #kwlog.setconsumer(self.log, kwlog.Path("/tmp/x.log")) - kwlog.setconsumer(self.log, None) + self.log = py.log.Producer("dsession") + if not config.option.debug: + py.log.setconsumer(self.log._keywords, None) self.shuttingdown = False self.countfailures = 0 self.maxfail = config.getvalue("maxfail") @@ -156,7 +159,7 @@ class DSession: def slave_slaveready(self, node): self.sched.addnode(node) if self.shuttingdown: - node.sendcommand("shutdown") + node.shutdown() def slave_slavefinished(self, node): crashitem = self.sched.remove_node(node) @@ -178,7 +181,7 @@ class DSession: self.report_line("[%s] collected %d test items" %( node.gateway.id, len(ids))) - if self.sched.collection_is_completed(): + if self.sched.collection_is_completed: self.sched.init_distribute() self.sched.triggertesting() @@ -204,15 +207,16 @@ class DSession: self.countfailures) def loop(self): - self.sched = LoadScheduling(len(self.config.option.tx), log=self.log) + numnodes = len(self.nodemanager.gwmanager.specs) + self.sched = LoadScheduling(numnodes, log=self.log) self.shouldstop = False self.session_finished = False exitstatus = 0 try: - while not (self.session_finished or self.shouldstop): + while not self.session_finished: self.loop_once() - if self.shouldstop: - raise Interrupted(str(self.shouldstop)) + if self.shouldstop: + raise Interrupted(str(self.shouldstop)) except KeyboardInterrupt: excinfo = py.code.ExceptionInfo() self.config.hook.pytest_keyboard_interrupt(excinfo=excinfo) @@ -236,12 +240,13 @@ class DSession: assert callname, kwargs method = "slave_" + callname call = getattr(self, method) - self.log.debug("calling method: %s(**%s)" % (method, kwargs)) + self.log("calling method: %s(**%s)" % (method, kwargs)) call(**kwargs) if self.sched.tests_finished(): self.triggershutdown() def triggershutdown(self): + self.log("triggering shutdown") self.shuttingdown = True for node in self.sched.node2pending: node.shutdown() diff --git a/xdist/remote.py b/xdist/remote.py index 421f273..e057158 100644 --- a/xdist/remote.py +++ b/xdist/remote.py @@ -5,38 +5,50 @@ import py import sys import execnet -import kwlog from py._plugin import pytest_runner as runner # XXX load dynamically +def make_reltoroot(roots, args): + # XXX introduce/use public API for splitting args + splitcode = "::" + l = [] + for arg in args: + parts = arg.split(splitcode) + fspath = py.path.local(parts[0]) + for root in roots: + x = fspath.relto(root) + if x or fspath == root: + parts[0] = root.basename + "/" + x + break + else: + raise ValueError("arg %s not relative to an rsync root" % (arg,)) + l.append(splitcode.join(parts)) + return l + class SlaveController(object): ENDMARK = -1 def __init__(self, nodemanager, gateway, config, putevent): - #self.nodemanager = nodemanager + self.nodemanager = nodemanager self.putevent = putevent self.gateway = gateway self.config = config - self.status = None self._down = False - self.status = "gateway-init" + self.log = py.log.Producer("slavectl-%s" % gateway.id) + if not self.config.option.debug: + py.log.setconsumer(self.log._keywords, None) def __repr__(self): - return "<%s id=%s status=%s>" %(self.__class__.__name__, - self.gateway.id, self.status) - - def trace(self, *args): - if self.config.option.debug: - msg = " ".join([str(x) for x in args]) - py.builtin.print_("SlaveController:", msg) + return "<%s %s>" %(self.__class__.__name__, self.gateway.id,) def setup(self): - self.trace("setting up slave session") - assert self.status == "gateway-init" + self.log("setting up slave session") + spec = self.gateway.spec + args = self.config.args + if not spec.popen or spec.chdir: + args = make_reltoroot(self.nodemanager.roots, args) self.channel = self.gateway.remote_exec(init_slave_session, - args=self.config.args, - option_dict=vars(self.config.option), + args=args, option_dict=vars(self.config.option), ) - self.status = "slave-init" if self.putevent: self.channel.setcallback(self.process_from_remote, endmarker=self.ENDMARK) @@ -44,11 +56,11 @@ class SlaveController(object): def ensure_teardown(self): if hasattr(self, 'channel'): if not self.channel.isclosed(): - self.trace("closing", self.channel) + self.log("closing", self.channel) self.channel.close() #del self.channel if hasattr(self, 'gateway'): - self.trace("exiting", self.gateway) + self.log("exiting", self.gateway) self.gateway.exit() #del self.gateway @@ -61,11 +73,11 @@ class SlaveController(object): def sendcommand(self, name, **kwargs): """ send a named parametrized command to the other side. """ - self.trace("sending command %s(**%s)" % (name, kwargs)) + self.log("sending command %s(**%s)" % (name, kwargs)) self.channel.send((name, kwargs)) def notify_inproc(self, eventname, **kwargs): - self.trace("queuing %s(**%s)" % (eventname, kwargs)) + self.log("queuing %s(**%s)" % (eventname, kwargs)) self.putevent((eventname, kwargs)) def process_from_remote(self, eventcall): @@ -87,7 +99,7 @@ class SlaveController(object): return eventname, kwargs = eventcall if eventname in ("collectionstart"): - self.trace("ignoring %s(%s)" %(eventname, kwargs)) + self.log("ignoring %s(%s)" %(eventname, kwargs)) elif eventname == "slaveready": self.notify_inproc(eventname, node=self) elif eventname == "slavefinished": @@ -130,26 +142,25 @@ def remote_initconfig(config, option_dict, args): config.option.dist = "no" config.option.distload = False config.option.numprocesses = None - #kwlog.Producer("slave").DEBUG("option dict", config.option.__dict__) config.args = args return config - + class SlaveInteractor: def __init__(self, config, channel): self.config = config - self.log = kwlog.Producer("slave") - kwlog.setconsumer(self.log, None) - self.log.info("initializing SlaveInteractor") + self.log = py.log.Producer("slave") + if not config.option.debug: + py.log.setconsumer(self.log._keywords, None) self.channel = channel config.pluginmanager.register(self) def sendevent(self, name, **kwargs): - self.log.debug("sending", name, kwargs) + self.log("sending", name, kwargs) self.channel.send((name, kwargs)) def pytest_internalerror(self, excrepr): for line in str(excrepr).split("\n"): - self.log.debug("IERROR> " + line) + self.log("IERROR> " + line) def pytest_sessionstart(self, session): self.session = session @@ -163,10 +174,10 @@ class SlaveInteractor: self.sendevent("collectionstart") def pytest_runtest_mainloop(self, session): - self.log.debug("entering main loop") + self.log("entering main loop") while 1: name, kwargs = self.channel.receive() - self.log.debug("received command %s(**%s)" % (name, kwargs)) + self.log("received command %s(**%s)" % (name, kwargs)) if name == "runtests": ids = kwargs['ids'] for nodeid in ids: @@ -177,7 +188,7 @@ class SlaveInteractor: return True def pytest_log_finishcollection(self, collection): - self.log.debug("pytest_log_finishcollection") + self.log("pytest_log_finishcollection") ids = [collection.getid(item) for item in collection.items] self.sendevent("collectionfinish", topdir=str(collection.topdir),