[gnome-continuous-yocto/gnomeostree-3.28-rocko: 3436/8267] bitbake: bitbake-worker: Handle cooker/worker IO deadlocking

[Emmanuele Bassi <ebassi src gnome org>]

commit 3d34ae6c11bc2e800f784740c54b2dd2029658ed
Author: Richard Purdie <richard purdie linuxfoundation org>
Date:   Thu Nov 24 21:41:30 2016 +0000

    bitbake: bitbake-worker: Handle cooker/worker IO deadlocking
    
    I noiced builds where tasks seemed to be taking a surprisingly long time.
    When I looked at the output of top/pstree, these tasks were no longer
    running despite being listed in knotty. Some were in D/Z state waiting for
    their exit code to be collected, others were simply not present at all.
    
    strace showed communication problems between the worker and cooker, each
    was trying to write to the other and nearly deadlocking. Eventually, timeouts
    would allow them to echange 64kb of data but this was only happening every
    few seconds.
    
    Whilst this particularly affected builds on machines with large numbers
    of cores (and hence highly parallal task execution) and in cases where
    I had a lot of debug enabled, this situation is clearly bad in general.
    
    This patch introduces a thread to the worker which is used to write data
    back to cooker. This means that the deadlock can't occur and data flows
    much more freely and effectively.
    
    (Bitbake rev: 3cb0d1c78b4c2e4f251a59b86c8da583828ad08b)
    
    Signed-off-by: Richard Purdie <richard purdie linuxfoundation org>

 bitbake/bin/bitbake-worker |   52 ++++++++++++++++++++++++++++++-------------
 1 files changed, 36 insertions(+), 16 deletions(-)
---
diff --git a/bitbake/bin/bitbake-worker b/bitbake/bin/bitbake-worker
index af66ff0..db3c4b1 100755
--- a/bitbake/bin/bitbake-worker
+++ b/bitbake/bin/bitbake-worker
@@ -12,7 +12,9 @@ import errno
 import signal
 import pickle
 import traceback
+import queue
 from multiprocessing import Lock
+from threading import Thread
 
 if sys.getfilesystemencoding() != "utf-8":
     sys.exit("Please use a locale setting which supports utf-8.\nPython can't change the filesystem locale 
after loading so we need a utf-8 when python starts or things won't work.")
@@ -64,7 +66,7 @@ if 0:
     consolelog.setFormatter(conlogformat)
     logger.addHandler(consolelog)
 
-worker_queue = b""
+worker_queue = queue.Queue()
 
 def worker_fire(event, d):
     data = b"<event>" + pickle.dumps(event) + b"</event>"
@@ -73,21 +75,38 @@ def worker_fire(event, d):
 def worker_fire_prepickled(event):
     global worker_queue
 
-    worker_queue = worker_queue + event
-    worker_flush()
+    worker_queue.put(event)
 
-def worker_flush():
-    global worker_queue, worker_pipe
+#
+# We can end up with write contention with the cooker, it can be trying to send commands
+# and we can be trying to send event data back. Therefore use a separate thread for writing 
+# back data to cooker.
+#
+worker_thread_exit = False
 
-    if not worker_queue:
-        return
+def worker_flush(worker_queue):
+    worker_queue_int = b""
+    global worker_pipe, worker_thread_exit
 
-    try:
-        written = os.write(worker_pipe, worker_queue)
-        worker_queue = worker_queue[written:]
-    except (IOError, OSError) as e:
-        if e.errno != errno.EAGAIN and e.errno != errno.EPIPE:
-            raise
+    while True:
+        try:
+            worker_queue_int = worker_queue_int + worker_queue.get(True, 1)
+        except queue.Empty:
+            pass
+        while (worker_queue_int or not worker_queue.empty()):
+            try:
+                if not worker_queue.empty():
+                    worker_queue_int = worker_queue_int + worker_queue.get()
+                written = os.write(worker_pipe, worker_queue_int)
+                worker_queue_int = worker_queue_int[written:]
+            except (IOError, OSError) as e:
+                if e.errno != errno.EAGAIN and e.errno != errno.EPIPE:
+                    raise
+        if worker_thread_exit and worker_queue.empty() and not worker_queue_int:
+            return
+
+worker_thread = Thread(target=worker_flush, args=(worker_queue,))
+worker_thread.start()
 
 def worker_child_fire(event, d):
     global worker_pipe
@@ -353,7 +372,6 @@ class BitbakeWorker(object):
                 self.build_pipes[pipe].read()
             if len(self.build_pids):
                 self.process_waitpid()
-            worker_flush()
 
 
     def handle_item(self, item, func):
@@ -458,8 +476,10 @@ except BaseException as e:
         import traceback
         sys.stderr.write(traceback.format_exc())
         sys.stderr.write(str(e))
-while len(worker_queue):
-    worker_flush()
+
+worker_thread_exit = True
+worker_thread.join()
+
 workerlog_write("exitting")
 sys.exit(0)