[oe-commits] [openembedded-core] 24/28: commands.py: live output logging + result.error encoding fix

git at git.openembedded.org git at git.openembedded.org
Wed Jun 28 19:56:28 UTC 2017


This is an automated email from the git hooks/post-receive script.

rpurdie pushed a commit to branch master
in repository openembedded-core.

commit 00b8c7ff17cd8f1920728fdc2653068e63d71724
Author: Patrick Ohly <patrick.ohly at intel.com>
AuthorDate: Tue Jun 27 13:03:40 2017 +0200

    commands.py: live output logging + result.error encoding fix
    
    Tests that use bitbake("my-test-image") can run for a long time
    without any indication to the user of oe-selftest about what's going
    on. The test author has to log the bitbake output explicitly,
    otherwise it is lost in case of test failures.
    
    Now it is possible to use bitbake("my-test-image",
    output_log=self.logger) to get more output both on the console and in
    the XML output (when xmlrunner is installed). Example output:
    
    2017-06-23 12:23:14,144 - oe-selftest - INFO - Running tests...
    2017-06-23 12:23:14,145 - oe-selftest - INFO - ----------------------------------------------------------------------
    2017-06-23 12:23:14,151 - oe-selftest - INFO - Running: bitbake my-test-image
    2017-06-23 12:23:16,363 - oe-selftest - INFO - Loading cache...done.
    2017-06-23 12:23:17,575 - oe-selftest - INFO - Loaded 3529 entries from dependency cache.
    2017-06-23 12:23:18,811 - oe-selftest - INFO - Parsing recipes...done.
    2017-06-23 12:23:19,659 - oe-selftest - INFO - Parsing of 2617 .bb files complete (2612 cached, 5 parsed). 3533 targets, 460 skipped, 0 masked, 0 errors.
    2017-06-23 12:23:19,659 - oe-selftest - INFO - NOTE: Resolving any missing task queue dependencies
    
    Because the implementation was already using threading, the same is
    done to decouple reading and writing the different pipes instead of
    trying to multiplex IO in a single thread. Previously the helper
    thread waited for command completion, now that is done in the main
    thread.
    
    The most common case (no input data, joined stdout/stderr) still uses
    one extra thread and a single read(), so performance should be roughly
    the same as before.
    
    Probably unintentionally, result.error was left as byte string when
    migrating to Python3. OE-core doesn't seem to use runCmd() with split
    output at the moment, so changing result.error to be treated the same
    as result.output (i.e. decoded to a normal strings) seems like a
    relatively safe API change (or rather, implementation fix).
    
    Signed-off-by: Patrick Ohly <patrick.ohly at intel.com>
    
    merge: wait()
    Signed-off-by: Ross Burton <ross.burton at intel.com>
---
 meta/lib/oeqa/utils/commands.py | 107 +++++++++++++++++++++++++++++++---------
 1 file changed, 85 insertions(+), 22 deletions(-)

diff --git a/meta/lib/oeqa/utils/commands.py b/meta/lib/oeqa/utils/commands.py
index 57286fc..5e53454 100644
--- a/meta/lib/oeqa/utils/commands.py
+++ b/meta/lib/oeqa/utils/commands.py
@@ -13,6 +13,7 @@ import sys
 import signal
 import subprocess
 import threading
+import time
 import logging
 from oeqa.utils import CommandError
 from oeqa.utils import ftools
@@ -25,7 +26,7 @@ except ImportError:
     pass
 
 class Command(object):
-    def __init__(self, command, bg=False, timeout=None, data=None, **options):
+    def __init__(self, command, bg=False, timeout=None, data=None, output_log=None, **options):
 
         self.defaultopts = {
             "stdout": subprocess.PIPE,
@@ -48,41 +49,103 @@ class Command(object):
         self.options.update(options)
 
         self.status = None
+        # We collect chunks of output before joining them at the end.
+        self._output_chunks = []
+        self._error_chunks = []
         self.output = None
         self.error = None
-        self.thread = None
+        self.threads = []
 
+        self.output_log = output_log
         self.log = logging.getLogger("utils.commands")
 
     def run(self):
         self.process = subprocess.Popen(self.cmd, **self.options)
 
-        def commThread():
-            self.output, self.error = self.process.communicate(self.data)
-
-        self.thread = threading.Thread(target=commThread)
-        self.thread.start()
+        def readThread(output, stream, logfunc):
+            if logfunc:
+                for line in stream:
+                    output.append(line)
+                    logfunc(line.decode("utf-8", errors='replace').rstrip())
+            else:
+                output.append(stream.read())
+
+        def readStderrThread():
+            readThread(self._error_chunks, self.process.stderr, self.output_log.error if self.output_log else None)
+
+        def readStdoutThread():
+            readThread(self._output_chunks, self.process.stdout, self.output_log.info if self.output_log else None)
+
+        def writeThread():
+            try:
+                self.process.stdin.write(self.data)
+                self.process.stdin.close()
+            except OSError as ex:
+                # It's not an error when the command does not consume all
+                # of our data. subprocess.communicate() also ignores that.
+                if ex.errno != EPIPE:
+                    raise
+
+        # We write in a separate thread because then we can read
+        # without worrying about deadlocks. The additional thread is
+        # expected to terminate by itself and we mark it as a daemon,
+        # so even it should happen to not terminate for whatever
+        # reason, the main process will still exit, which will then
+        # kill the write thread.
+        if self.data:
+            threading.Thread(target=writeThread, daemon=True).start()
+        if self.process.stderr:
+            thread = threading.Thread(target=readStderrThread)
+            thread.start()
+            self.threads.append(thread)
+        if self.output_log:
+            self.output_log.info('Running: %s' % self.cmd)
+        thread = threading.Thread(target=readStdoutThread)
+        thread.start()
+        self.threads.append(thread)
 
         self.log.debug("Running command '%s'" % self.cmd)
 
         if not self.bg:
-            self.thread.join(self.timeout)
+            if self.timeout is None:
+                for thread in self.threads:
+                    thread.join()
+            else:
+                deadline = time.time() + self.timeout
+                for thread in self.threads:
+                    timeout = deadline - time.time() 
+                    if timeout < 0:
+                        timeout = 0
+                    thread.join(timeout)
             self.stop()
 
     def stop(self):
-        if self.thread.isAlive():
-            self.process.terminate()
+        for thread in self.threads:
+            if thread.isAlive():
+                self.process.terminate()
             # let's give it more time to terminate gracefully before killing it
-            self.thread.join(5)
-            if self.thread.isAlive():
+            thread.join(5)
+            if thread.isAlive():
                 self.process.kill()
-                self.thread.join()
+                thread.join()
 
-        if not self.output:
-            self.output = ""
-        else:
-            self.output = self.output.decode("utf-8", errors='replace').rstrip()
-        self.status = self.process.poll()
+        def finalize_output(data):
+            if not data:
+                data = ""
+            else:
+                data = b"".join(data)
+                data = data.decode("utf-8", errors='replace').rstrip()
+            return data
+
+        self.output = finalize_output(self._output_chunks)
+        self._output_chunks = None
+        # self.error used to be a byte string earlier, probably unintentionally.
+        # Now it is a normal string, just like self.output.
+        self.error = finalize_output(self._error_chunks)
+        self._error_chunks = None
+        # At this point we know that the process has closed stdout/stderr, so
+        # it is safe and necessary to wait for the actual process completion.
+        self.status = self.process.wait()
 
         self.log.debug("Command '%s' returned %d as exit code." % (self.cmd, self.status))
         # logging the complete output is insane
@@ -98,7 +161,7 @@ class Result(object):
 
 
 def runCmd(command, ignore_status=False, timeout=None, assert_error=True,
-          native_sysroot=None, limit_exc_output=0, **options):
+          native_sysroot=None, limit_exc_output=0, output_log=None, **options):
     result = Result()
 
     if native_sysroot:
@@ -108,7 +171,7 @@ def runCmd(command, ignore_status=False, timeout=None, assert_error=True,
         nenv['PATH'] = extra_paths + ':' + nenv.get('PATH', '')
         options['env'] = nenv
 
-    cmd = Command(command, timeout=timeout, **options)
+    cmd = Command(command, timeout=timeout, output_log=output_log, **options)
     cmd.run()
 
     result.command = command
@@ -132,7 +195,7 @@ def runCmd(command, ignore_status=False, timeout=None, assert_error=True,
     return result
 
 
-def bitbake(command, ignore_status=False, timeout=None, postconfig=None, **options):
+def bitbake(command, ignore_status=False, timeout=None, postconfig=None, output_log=None, **options):
 
     if postconfig:
         postconfig_file = os.path.join(os.environ.get('BUILDDIR'), 'oeqa-post.conf')
@@ -147,7 +210,7 @@ def bitbake(command, ignore_status=False, timeout=None, postconfig=None, **optio
         cmd = [ "bitbake" ] + [a for a in (command + extra_args.split(" ")) if a not in [""]]
 
     try:
-        return runCmd(cmd, ignore_status, timeout, **options)
+        return runCmd(cmd, ignore_status, timeout, output_log=output_log, **options)
     finally:
         if postconfig:
             os.remove(postconfig_file)

-- 
To stop receiving notification emails like this one, please contact
the administrator of this repository.


More information about the Openembedded-commits mailing list