commit 7eac2854b2bfd504a99a4f0cdcefd41c52fe73c4 Author: Jiri Pirko jiri@resnulli.us Date: Tue Sep 11 17:02:26 2012 +0200
introduce command context
In this context, remember all background running commands. Kill them on cleanup.
Signed-off-by: Jiri Pirko jiri@resnulli.us
NetTest/NetTestCommand.py | 172 ++++++++++++++++++++++++----------------- NetTest/NetTestController.py | 7 +- NetTest/NetTestSlave.py | 18 +++-- 3 files changed, 117 insertions(+), 80 deletions(-) --- diff --git a/NetTest/NetTestCommand.py b/NetTest/NetTestCommand.py index e89f606..4fde1ee 100644 --- a/NetTest/NetTestCommand.py +++ b/NetTest/NetTestCommand.py @@ -37,42 +37,90 @@ class CommandException(Exception): def __str__(self): return "CommandException: " + str(self.command)
-class BgProcessException(Exception): - """Base class for client errors.""" +class BgCommandException(Exception): + """Base class for background command errors.""" def __init__(self, str): self._str = str
def __str__(self): - return "BgProcessError: " + self._str - -class BGProcesses: - def __init__(self): - self._dict = {} - - def add(self, bg_id, pid, pipe): - if bg_id in self._dict: - raise Exception - self._dict[bg_id] = {"pid": pid, "pipe": pipe} + return "BgCommandError: " + self._str
- def get_pid(self, bg_id): - return self._dict[bg_id]["pid"] +class BgCommand: + def __init__(self, bg_id, cmd_cls): + self._bg_id = bg_id + self._cmd_cls = cmd_cls + self._pid = None + self._read_pipe = None
- def get_pipe(self, bg_id): - return self._dict[bg_id]["pipe"] + def get_bg_id(self): + return self._bg_id
- def remove(self, bg_id): - del self._dict[bg_id] + def run(self): + read_pipe, write_pipe = os.pipe() + self._pid = os.fork() + if self._pid: + os.close(write_pipe) + logging.debug("Running in background with" + " id "%s", pid "%d"" % (self._bg_id, self._pid)) + self._read_pipe = read_pipe + return {"passed": True} + os.close(read_pipe) + os.setpgrp() + self._cmd_cls.set_handle_intr() + try: + self._cmd_cls.run() + result = self._cmd_cls.get_result() + except KeyboardInterrupt: + result = self._cmd_cls.get_result() + except: + type, value, tb = sys.exc_info() + result = {"Exception": ''.join(traceback.format_exception(type, value, tb))} + tmp = pickle.dumps(result) + os.write(write_pipe, tmp) + os.close(write_pipe) + os._exit(0) + + def wait_for(self): + logging.debug("Waiting for background command with id "%s", pid "%d"" % (self._bg_id, self._pid)) + os.waitpid(self._pid, 0) + + def interrupt(self): + logging.debug("Interrupting background command with id "%s", pid "%d"" % (self._bg_id, self._pid)) + os.killpg(os.getpgid(self._pid), signal.SIGINT) + os.waitpid(self._pid, 0) + + def kill(self): + logging.debug("Killing background command with id "%s", pid "%d"" % (self._bg_id, self._pid)) + os.killpg(os.getpgid(self._pid), signal.SIGKILL)
- def get_bg_process_result(self, bg_id): - pipe = self.get_pipe(bg_id) - tmp = os.read(pipe, 4096*10) + def get_result(self): + tmp = os.read(self._read_pipe, 4096*10) result = pickle.loads(tmp) if "Exception" in result: - raise BgProcessException(result["Exception"]) - os.close(pipe) + raise BgCommandException(result["Exception"]) + os.close(self._read_pipe) return result
-bg_processes = BGProcesses() +class NetTestCommandContext: + def __init__(self): + self._dict = {} + + def add_bg_cmd(self, bg_cmd): + self._dict[bg_cmd.get_bg_id()] = bg_cmd + + def del_bg_cmd(self, bg_cmd): + del self._dict[bg_cmd.get_bg_id()] + + def get_bg_cmd(self, bg_id): + return self._dict[bg_id] + + def _kill_all_bg_cmds(self): + for bg_id in self._dict: + self._dict[bg_id].kill() + + def cleanup(self): + self._kill_all_bg_cmds() + self._dict = {}
def NetTestCommandTest(command): test_name = command["value"] @@ -174,48 +222,49 @@ class NetTestCommandSystemConfig(NetTestCommandGeneric): res["res_data"] = res_data self.set_result(res)
-class NetTestCommandWait(NetTestCommandGeneric): +class NetTestCommandControl(NetTestCommandGeneric): + def __init__(self, command_context, command): + self._command_context = command_context + NetTestCommandGeneric.__init__(self, command) + +class NetTestCommandWait(NetTestCommandControl): def run(self): bg_id = self._command["value"] - pid = bg_processes.get_pid(bg_id) - logging.debug("Waiting for background id "%s", pid "%d"" % (bg_id, pid)) - os.waitpid(pid, 0) - result = bg_processes.get_bg_process_result(bg_id) - bg_processes.remove(bg_id) + bg_cmd = self._command_context.get_bg_cmd(bg_id) + bg_cmd.wait_for() + result = bg_cmd.get_result() + self._command_context.del_bg_cmd(bg_cmd) self.set_result(result)
-class NetTestCommandIntr(NetTestCommandGeneric): +class NetTestCommandIntr(NetTestCommandControl): def run(self): bg_id = self._command["value"] - pid = bg_processes.get_pid(bg_id) - logging.debug("Interrupting background id "%s", pid "%d"" % (bg_id, pid)) - os.killpg(os.getpgid(pid), signal.SIGINT) - os.waitpid(pid, 0) - result = bg_processes.get_bg_process_result(bg_id) - bg_processes.remove(bg_id) + bg_cmd = self._command_context.get_bg_cmd(bg_id) + bg_cmd.interrupt() + result = bg_cmd.get_result() + self._command_context.del_bg_cmd(bg_cmd) self.set_result(result)
-class NetTestCommandKill(NetTestCommandGeneric): +class NetTestCommandKill(NetTestCommandControl): def run(self): bg_id = self._command["value"] - pid = bg_processes.get_pid(bg_id) - logging.debug("Killing background id "%s", pid "%d"" % (bg_id, pid)) - os.killpg(os.getpgid(pid), signal.SIGKILL) - bg_processes.remove(bg_id) + bg_cmd = self._command_context.get_bg_cmd(bg_id) + bg_cmd.kill() + self._command_context.del_bg_cmd(bg_cmd) self.set_result({"passed": True})
-def get_command_class(command): +def get_command_class(command_context, command): cmd_type = command["type"] if cmd_type == "exec": return NetTestCommandExec(command) elif cmd_type == "test": return NetTestCommandTest(command) elif cmd_type == "wait": - return NetTestCommandWait(command) + return NetTestCommandWait(command_context, command) elif cmd_type == "intr": - return NetTestCommandIntr(command) + return NetTestCommandIntr(command_context, command) elif cmd_type == "kill": - return NetTestCommandKill(command) + return NetTestCommandKill(command_context, command) elif cmd_type == "system_config": return NetTestCommandSystemConfig(command) else: @@ -223,37 +272,18 @@ def get_command_class(command): raise Exception("Unknown command type "%s"" % cmd_type)
class NetTestCommand: - def __init__(self, command): - self._command_class = get_command_class(command) + def __init__(self, command_context, command): + self._command_class = get_command_class(command_context, command) + self._command_context = command_context self._command = command
def run(self): cmd_cls = self._command_class if "bg_id" in self._command: bg_id = self._command["bg_id"] - read_pipe, write_pipe = os.pipe() - pid = os.fork() - if pid: - os.close(write_pipe) - logging.debug("Running in background with" - " id "%s", pid "%d"" % (bg_id, pid)) - bg_processes.add(bg_id, pid, read_pipe) - return {"passed": True} - os.close(read_pipe) - os.setpgrp() - cmd_cls.set_handle_intr() - try: - cmd_cls.run() - result = cmd_cls.get_result() - except KeyboardInterrupt: - result = cmd_cls.get_result() - except: - type, value, tb = sys.exc_info() - result = {"Exception": ''.join(traceback.format_exception(type, value, tb))} - tmp = pickle.dumps(result) - os.write(write_pipe, tmp) - os.close(write_pipe) - os._exit(0) + bg_cmd = BgCommand(bg_id, cmd_cls) + self._command_context.add_bg_cmd(bg_cmd) + return bg_cmd.run() else: cmd_cls.run() return cmd_cls.get_result() diff --git a/NetTest/NetTestController.py b/NetTest/NetTestController.py index 2a6bbd6..01a5f9b 100644 --- a/NetTest/NetTestController.py +++ b/NetTest/NetTestController.py @@ -22,7 +22,7 @@ from NetTest.NetTestParse import NetTestParse from Common.SlaveUtils import prepare_client_session from Common.NetUtils import get_corespond_local_ip, MacPool from NetTest.NetTestSlave import DefaultRPCPort -from NetTest.NetTestCommand import NetTestCommand, str_command +from NetTest.NetTestCommand import NetTestCommandContext, NetTestCommand, str_command from Common.LoggingServer import LoggingServer from Common.VirtUtils import VirtNetCtl, VirtDomainCtl, BridgeCtl from Common.Utils import wait_for @@ -43,7 +43,7 @@ class NetTestController: self._docleanup = cleanup self._res_serializer = res_serializer self._remote_capture_files = {} - + self._command_context = NetTestCommandContext() self._machine_pool = MachinePool([])
self._recipe = {} @@ -314,7 +314,7 @@ class NetTestController: pass
if machine_id == "0": - cmd_res = NetTestCommand(command).run() + cmd_res = NetTestCommand(self._command_context, command).run() else: rpc = self._get_machinerpc(machine_id) if "timeout" in command: @@ -390,6 +390,7 @@ class NetTestController:
self._deconfigure_slaves() self._disconnect_slaves() + self._command_context.cleanup()
if not err: return res diff --git a/NetTest/NetTestSlave.py b/NetTest/NetTestSlave.py index 8065591..7c391f8 100644 --- a/NetTest/NetTestSlave.py +++ b/NetTest/NetTestSlave.py @@ -20,7 +20,7 @@ from Common.XmlRpc import Server from SimpleXMLRPCServer import SimpleXMLRPCRequestHandler from NetConfig.NetConfig import NetConfig from NetConfig.NetConfigDevice import NetConfigDeviceAllCleanup -from NetTest.NetTestCommand import NetTestCommand, CommandException +from NetTest.NetTestCommand import NetTestCommandContext, NetTestCommand, CommandException from Common.Utils import die_when_parent_die from Common.NetUtils import scan_netdevs
@@ -30,10 +30,11 @@ class NetTestSlaveXMLRPC: ''' Exported xmlrpc methods ''' - def __init__(self): + def __init__(self, command_context): self._netconfig = None self._packet_captures = {} self._netconfig = NetConfig() + self._command_context = command_context
def hello(self): return "hello" @@ -112,7 +113,7 @@ class NetTestSlaveXMLRPC:
def run_command(self, command): try: - return NetTestCommand(command).run() + return NetTestCommand(self._command_context, command).run() except: log_exc_traceback() raise CommandException(command) @@ -120,11 +121,13 @@ class NetTestSlaveXMLRPC: def machine_cleanup(self): NetConfigDeviceAllCleanup() self._netconfig.cleanup() + self._command_context.cleanup() return True
class MySimpleXMLRPCServer(Server): - def __init__(self, *args, **kwargs): + def __init__(self, command_context, *args, **kwargs): self._finished = False + self._command_context = command_context Server.__init__(self, *args, **kwargs)
def register_die_signal(self, signum): @@ -138,6 +141,7 @@ class MySimpleXMLRPCServer(Server): while True: try: if self._finished: + self._command_context.cleanup() import sys sys.exit() return @@ -149,12 +153,14 @@ class NetTestSlave: def __init__(self, port = DefaultRPCPort): die_when_parent_die()
- server = MySimpleXMLRPCServer(("", port), SimpleXMLRPCRequestHandler, + command_context = NetTestCommandContext() + server = MySimpleXMLRPCServer(command_context, + ("", port), SimpleXMLRPCRequestHandler, logRequests = False) server.register_die_signal(signal.SIGHUP) server.register_die_signal(signal.SIGINT) server.register_die_signal(signal.SIGTERM) - server.register_instance(NetTestSlaveXMLRPC()) + server.register_instance(NetTestSlaveXMLRPC(command_context)) self._server = server
def run(self):