Merge lp:~zyga/checkbox/fix-1443911 into lp:checkbox

[Zygmunt Krynicki]

[Maciej Kisielewski (kissiel)]

+1, altough chunk size of 1 somewhat unsettles me.

review: Approve

[Zygmunt Krynicki (zyga)]

1 is the only that works as long as we are stuck in blocking mode.

I would like to switch to glibc-based code but we first need to solve
the waiting-for-process-from-a-thread issue.

On Wed, Apr 15, 2015 at 11:58 AM, Maciej Kisielewski
<email address hidden> wrote:
> Review: Approve
>
> +1, altough chunk size of 1 somewhat unsettles me.
>
> --
> https://code.launchpad.net/~zyga/checkbox/fix-1443911/+merge/256138
> You are the owner of lp:~zyga/checkbox/fix-1443911.

=== modified file 'plainbox/plainbox/impl/commands/inv_run.py'
--- plainbox/plainbox/impl/commands/inv_run.py	2015-03-31 07:26:46 +0000
+++ plainbox/plainbox/impl/commands/inv_run.py	2015-04-14 13:10:26 +0000
@@ -197,8 +197,8 @@
             'stdout': sys.stdout,
             'stderr': sys.stderr
         }[stream_name]
-        print(self.C.BLACK(line.decode("UTF-8", "ignore").rstrip('\n')),
-              file=stream)
+        print(self.C.BLACK(line.decode("UTF-8", "ignore")),
+              end='', flush=True, file=stream)
 
     def finished_executing_program(self, returncode):
         if self.show_cmd_output:

=== modified file 'plainbox/plainbox/impl/providers/manifest/units/manifest.pxu'
--- plainbox/plainbox/impl/providers/manifest/units/manifest.pxu	2015-03-30 17:36:35 +0000
+++ plainbox/plainbox/impl/providers/manifest/units/manifest.pxu	2015-04-14 13:10:26 +0000
@@ -8,7 +8,7 @@
 plugin: user-interact
 command: plainbox-manifest-collect
 estimated_duration: 30
-flags: preserve-locale
+flags: preserve-locale use-chunked-io
 
 unit: job
 id: manifest

=== modified file 'plainbox/plainbox/impl/runner.py'
--- plainbox/plainbox/impl/runner.py	2015-04-02 14:44:57 +0000
+++ plainbox/plainbox/impl/runner.py	2015-04-14 13:10:26 +0000
@@ -243,6 +243,15 @@
         if self.ui is not None:
             self.ui.got_program_output(stream_name, line)
 
+    def on_chunk(self, stream_name, chunk):
+        """
+        Internal method of extcmd.DelegateBase
+
+        Called for each chunk of output.
+        """
+        if self.ui is not None:
+            self.ui.got_program_output(stream_name, chunk)
+
 
 class JobRunner(IJobRunner):
     """
@@ -862,7 +871,11 @@
         # Create a subprocess.Popen() like object that uses the delegate
         # system to observe all IO as it occurs in real time.
         delegate_cls = self._get_delegate_cls(config)
-        extcmd_popen = delegate_cls(delegate)
+        flags = 0
+        # Use chunked IO for jobs that explicitly request this
+        if 'use-chunked-io' in job.get_flag_set():
+            flags |= extcmd.CHUNKED_IO
+        extcmd_popen = delegate_cls(delegate, flags=flags)
         # Stream all IOLogRecord entries to disk
         record_path = os.path.join(
             self._jobs_io_log_dir, "{}.record.gz".format(

=== modified file 'plainbox/plainbox/vendor/extcmd/__init__.py'
--- plainbox/plainbox/vendor/extcmd/__init__.py	2014-12-12 10:13:29 +0000
+++ plainbox/plainbox/vendor/extcmd/__init__.py	2015-04-14 13:10:26 +0000
@@ -291,6 +291,18 @@
     def on_line(self, stream_name, line):
         """
         Callback invoked for each line of the output
+
+        This method is only called when the ``CHUNKED_IO`` flag is **not**
+        passed to extcmd. Otherwise :meth:`on_chunk()` will be called instead.
+        """
+
+    @abc.abstractmethod
+    def on_chunk(self, stream_name, chunk):
+        """
+        Callback invoked on each chunk of the input
+
+        This method is only called when the ``CHUNKED_IO`` flag is passed to
+        extcmd. Otherwise :meth:`on_line()` will be called instead.
         """
 
     @abc.abstractmethod
@@ -327,6 +339,11 @@
         Do nothing
         """
 
+    def on_chunk(self, stream_name, chunk):
+        """
+        Do nothing
+        """
+
     def on_end(self, returncode):
         """
         Do nothing
@@ -379,6 +396,13 @@
         if hasattr(self._delegate, "on_line"):
             self._delegate.on_line(stream_name, line)
 
+    def on_chunk(self, stream_name, chunk):
+        """
+        Call on_chunk() on the wrapped delegate if supported
+        """
+        if hasattr(self._delegate, "on_chunk"):
+            self._delegate.on_chunk(stream_name, chunk)
+
     def on_end(self, returncode):
         """
         Call on_end() on the wrapped delegate if supported
@@ -411,6 +435,9 @@
             return cls(delegate)
 
 
+CHUNKED_IO = 1
+
+
 class ExternalCommandWithDelegate(ExternalCommand):
     """
     The actually interesting subclass of ExternalCommand.
@@ -426,17 +453,19 @@
         very heavyweight but (yay) works portably for windows. A unix-specific
         subclass implementing this with _just_ poll could be provided with the
         same interface.
-
     """
 
-    def __init__(self, delegate, killsig=signal.SIGINT):
+    def __init__(self, delegate, killsig=signal.SIGINT, flags=0):
         """
         Set the delegate helper. Technically it needs to have a 'on_line()'
         method. For actual example code look at :class:`Tee`.
         """
+        _logger.debug("ExternalCommandWithDelegate(%r, killsig=%r, flags=%x)",
+                      delegate, killsig, flags)
         self._queue = Queue()
         self._delegate = SafeDelegate.wrap_if_needed(delegate)
         self._killsig = killsig
+        self._flags = flags
 
     def call(self, *args, **kwargs):
         """
@@ -567,7 +596,10 @@
         _logger.debug("_read_stream(%r, %r) entering", stream, stream_name)
         while True:
             try:
-                line = stream.readline()
+                if self._flags & CHUNKED_IO:
+                    data = stream.read(1)
+                else:
+                    data = stream.readline()
             except (IOError, ValueError):
                 # Ignore IOError and ValueError that may be raised if
                 # the stream was closed this can happen if the process exits
@@ -575,9 +607,9 @@
                 # starts to close both of the streams
                 break
             else:
-                if len(line) == 0:
+                if len(data) == 0:
                     break
-                cmd = (stream_name, line)
+                cmd = (stream_name, data)
                 self._queue.put(cmd)
         _logger.debug("_read_stream(%r, %r) exiting", stream, stream_name)
 
@@ -587,7 +619,10 @@
             args = self._queue.get()
             if args is None:
                 break
-            self._delegate.on_line(*args)
+            if self._flags & CHUNKED_IO:
+                self._delegate.on_chunk(*args)
+            else:
+                self._delegate.on_line(*args)
         _logger.debug("_drain_queue() exiting")
 
 
@@ -627,6 +662,13 @@
         for delegate in self.delegate_list:
             delegate.on_line(stream_name, line)
 
+    def on_chunk(self, stream_name, chunk):
+        """
+        Call the on_line() method on each delegate in the list
+        """
+        for delegate in self.delegate_list:
+            delegate.on_chunk(stream_name, chunk)
+
     def on_end(self, returncode):
         """
         Call the on_end() method on each delegate in the list
@@ -677,6 +719,16 @@
         else:
             self._stderr.write(line)
 
+    def on_chunk(self, stream_name, chunk):
+        """
+        Write each chunk, verbatim, to the desired stream.
+        """
+        assert stream_name == 'stdout' or stream_name == 'stderr'
+        if stream_name == 'stdout':
+            self._stdout.write(chunk)
+        else:
+            self._stderr.write(chunk)
+
     def on_end(self, returncode):
         """
         Close the output streams if requested
@@ -724,6 +776,14 @@
         transformed_line = self._callback(stream_name, line)
         self._delegate.on_line(stream_name, transformed_line)
 
+    def on_chunk(self, stream_name, chunk):
+        """
+        Transform each chunk by calling callback(stream_name, chunk) and pass
+        it down to the subsequent delegate.
+        """
+        transformed_chunk = self._callback(stream_name, chunk)
+        self._delegate.on_chunk(stream_name, transformed_chunk)
+
     def on_begin(self, args, kwargs):
         self._delegate.on_begin(args, kwargs)
 

=== modified file 'plainbox/plainbox/vendor/extcmd/glibc.py'
--- plainbox/plainbox/vendor/extcmd/glibc.py	2014-12-15 20:50:24 +0000
+++ plainbox/plainbox/vendor/extcmd/glibc.py	2015-04-14 13:10:26 +0000
@@ -46,6 +46,7 @@
 
 from plainbox.vendor.extcmd import ExternalCommand
 from plainbox.vendor.extcmd import SafeDelegate
+from plainbox.vendor.extcmd import CHUNKED_IO
 
 _logger = logging.getLogger("plainbox.vendor.extcmd")
 _bug_logger = logging.getLogger("plainbox.bug")
@@ -79,9 +80,10 @@
     multi-threaded application.
     """
 
-    def __init__(self, delegate, killsig=signal.SIGINT):
+    def __init__(self, delegate, killsig=signal.SIGINT, flags=0):
         self._delegate = SafeDelegate.wrap_if_needed(delegate)
         self._killsig = killsig
+        self._flags = flags
 
     def call(self, args, bufsize=-1, executable=None,
              stdin=None, stdout=None, stderr=None,
@@ -215,7 +217,18 @@
         _logger.debug("Sending SIGQUIT to process %d", pid)
         os.kill(pid, signal.SIGQUIT)
 
-    def _read_pipe(self, fd, name, buffer_map, force_last):
+    def _read_pipe_chunked(self, fd, name, force_last):
+        assert name in ('stdout', 'stderr')
+        pipe_size = fcntl.fcntl(fd, F_GETPIPE_SZ)
+        _logger.debug("Reading at most %d bytes of data from %s pipe",
+                      pipe_size, name)
+        data = os.read(fd, pipe_size)
+        done_reading = force_last or len(data) == 0
+        _logger.debug("Read %d bytes of data from %s", len(data), name)
+        self._delegate.on_chunk(name, data)
+        return done_reading
+
+    def _read_pipe_lines(self, fd, name, buffer_map, force_last):
         assert name in ('stdout', 'stderr')
         pipe_size = fcntl.fcntl(fd, F_GETPIPE_SZ)
         _logger.debug("Reading at most %d bytes of data from %s pipe",
@@ -291,8 +304,13 @@
                         # could have written and don't wait forever if the pipe
                         # has leaked.
                         force_last = 'proc' not in waiting_for
-                        if self._read_pipe(key.fd, key.data, buffer_map,
-                                           force_last):
+                        if self._flags & CHUNKED_IO:
+                            is_done = self._read_pipe_chunked(
+                                key.fd, key.data, force_last)
+                        else:
+                            is_done = self._read_pipe_lines(
+                                key.fd, key.data, buffer_map, force_last)
+                        if is_done:
                             _logger.debug(
                                 "pipe %s depleted, unregistering and closing",
                                 key.data)