From f323af122fcfbf31061a04309adc21fbcdae25fa Mon Sep 17 00:00:00 2001 From: Sam Thursfield Date: Tue, 7 Apr 2015 15:17:43 +0000 Subject: distbuild: Move SubprocessEventSource into its own module Previously it was only available in the distbuild-helper program. Moving it to its own module means we can test it and reuse it. This commit also adds a docstring to the class. Change-Id: Iaf7854048cf0ff463a87894f1f500cdcb6a34d8b --- distbuild-helper | 80 ++---------------------------- distbuild/__init__.py | 3 ++ distbuild/subprocess_eventsrc.py | 103 +++++++++++++++++++++++++++++++++++++++ without-test-modules | 1 + 4 files changed, 112 insertions(+), 75 deletions(-) create mode 100644 distbuild/subprocess_eventsrc.py diff --git a/distbuild-helper b/distbuild-helper index d77fcaea..76a39d5f 100755 --- a/distbuild-helper +++ b/distbuild-helper @@ -32,78 +32,6 @@ import urlparse import distbuild -class FileReadable(object): - - def __init__(self, request_id, p, f): - self.request_id = request_id - self.process = p - self.file = f - - -class FileWriteable(object): - - def __init__(self, request_id, p, f): - self.request_id = request_id - self.process = p - self.file = f - - -class SubprocessEventSource(distbuild.EventSource): - - def __init__(self): - self.procs = [] - self.closed = False - - def get_select_params(self): - r = [] - w = [] - for requst_id, p in self.procs: - if p.stdin_contents is not None: - w.append(p.stdin) - if p.stdout is not None: - r.append(p.stdout) - if p.stderr is not None: - r.append(p.stderr) - return r, w, [], None - - def get_events(self, r, w, x): - events = [] - - for request_id, p in self.procs: - if p.stdin in w: - events.append(FileWriteable(request_id, p, p.stdin)) - if p.stdout in r: - events.append(FileReadable(request_id, p, p.stdout)) - if p.stderr in r: - events.append(FileReadable(request_id, p, p.stderr)) - - return events - - def add(self, request_id, process): - - self.procs.append((request_id, process)) - distbuild.set_nonblocking(process.stdin) - distbuild.set_nonblocking(process.stdout) - distbuild.set_nonblocking(process.stderr) - - def remove(self, process): - self.procs = [t for t in self.procs if t[1] != process] - - def kill_by_id(self, request_id): - logging.debug('SES: Killing all processes for %s', request_id) - for id, process in self.procs: - if id == request_id: - logging.debug('SES: killing %s', repr(process)) - process.kill() - - def close(self): - self.procs = [] - self.closed = True - - def is_finished(self): - return self.closed - - class HelperMachine(distbuild.StateMachine): def __init__(self, conn): @@ -117,7 +45,7 @@ class HelperMachine(distbuild.StateMachine): jm = self.jm = distbuild.JsonMachine(self.conn) self.mainloop.add_state_machine(jm) - p = self.procsrc = SubprocessEventSource() + p = self.procsrc = distbuild.SubprocessEventSource() self.mainloop.add_event_source(p) self.send_helper_ready(jm) @@ -125,8 +53,10 @@ class HelperMachine(distbuild.StateMachine): spec = [ ('waiting', jm, distbuild.JsonNewMessage, 'waiting', self.do), ('waiting', jm, distbuild.JsonEof, None, self._eofed), - ('waiting', p, FileReadable, 'waiting', self._relay_exec_output), - ('waiting', p, FileWriteable, 'waiting', self._feed_stdin), + ('waiting', p, distbuild.FileReadable, 'waiting', + self._relay_exec_output), + ('waiting', p, distbuild.FileWriteable, 'waiting', + self._feed_stdin), ] self.add_transitions(spec) diff --git a/distbuild/__init__.py b/distbuild/__init__.py index 271b5def..e6ceda1f 100644 --- a/distbuild/__init__.py +++ b/distbuild/__init__.py @@ -62,4 +62,7 @@ from crashpoint import (crash_point, add_crash_condition, add_crash_conditions, from distbuild_socket import create_socket +from subprocess_eventsrc import (FileReadable, FileWriteable, + SubprocessEventSource) + __all__ = locals() diff --git a/distbuild/subprocess_eventsrc.py b/distbuild/subprocess_eventsrc.py new file mode 100644 index 00000000..52121502 --- /dev/null +++ b/distbuild/subprocess_eventsrc.py @@ -0,0 +1,103 @@ +# distbuild/subprocess_eventsrc.py -- for managing subprocesses +# +# Copyright (C) 2014-2015 Codethink Limited +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; version 2 of the License. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program. If not, see . + + +import logging + +import distbuild + + +class FileReadable(object): + + def __init__(self, request_id, p, f): + self.request_id = request_id + self.process = p + self.file = f + + +class FileWriteable(object): + + def __init__(self, request_id, p, f): + self.request_id = request_id + self.process = p + self.file = f + + +class SubprocessEventSource(distbuild.EventSource): + '''Event source for monitoring one or more subprocesses. + + This will send FileReadable and FileWritable events based on the + stdin and stdout and stderr handles of each subprocesses. + + When the subprocess terminates, you'll receive final FileReadable events + for stdout and for stderr. At that point, reading from those file + descriptors will return None, at which point you can be sure that the + subprocess is no longer running. + + ''' + + def __init__(self): + self.procs = [] + self.closed = False + + def get_select_params(self): + r = [] + w = [] + for requst_id, p in self.procs: + if p.stdin_contents is not None: + w.append(p.stdin) + if p.stdout is not None: + r.append(p.stdout) + if p.stderr is not None: + r.append(p.stderr) + return r, w, [], None + + def get_events(self, r, w, x): + events = [] + + for request_id, p in self.procs: + if p.stdin in w: + events.append(FileWriteable(request_id, p, p.stdin)) + if p.stdout in r: + events.append(FileReadable(request_id, p, p.stdout)) + if p.stderr in r: + events.append(FileReadable(request_id, p, p.stderr)) + + return events + + def add(self, request_id, process): + + self.procs.append((request_id, process)) + distbuild.set_nonblocking(process.stdin) + distbuild.set_nonblocking(process.stdout) + distbuild.set_nonblocking(process.stderr) + + def remove(self, process): + self.procs = [t for t in self.procs if t[1] != process] + + def kill_by_id(self, request_id): + logging.debug('SES: Killing all processes for %s', request_id) + for id, process in self.procs: + if id == request_id: + logging.debug('SES: killing %s', repr(process)) + process.kill() + + def close(self): + self.procs = [] + self.closed = True + + def is_finished(self): + return self.closed diff --git a/without-test-modules b/without-test-modules index caf10c8b..fb66ff69 100644 --- a/without-test-modules +++ b/without-test-modules @@ -50,6 +50,7 @@ distbuild/proxy_event_source.py distbuild/sockbuf.py distbuild/socketsrc.py distbuild/sockserv.py +distbuild/subprocess_eventsrc.py distbuild/timer_event_source.py distbuild/worker_build_scheduler.py # Not unit tested, since it needs a full system branch -- cgit v1.2.1