[oe-commits] [bitbake] branch master updated: bitbake-worker: Handle cooker/worker IO deadlocking

[git at git.openembedded.org]

rpurdie pushed a commit to branch master
in repository bitbake.

The following commit(s) were added to refs/heads/master by this push:
       new  3cb0d1c   bitbake-worker: Handle cooker/worker IO deadlocking
3cb0d1c is described below

commit 3cb0d1c78b4c2e4f251a59b86c8da583828ad08b
Author: Richard Purdie <richard.purdie at linuxfoundation.org>
AuthorDate: Thu Nov 24 21:41:30 2016 +0000

    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.
    
    Signed-off-by: Richard Purdie <richard.purdie at linuxfoundation.org>
---
 bin/bitbake-worker | 52 ++++++++++++++++++++++++++++++++++++----------------
 1 file changed, 36 insertions(+), 16 deletions(-)

diff --git a/bin/bitbake-worker b/bin/bitbake-worker
index af66ff0..db3c4b1 100755
--- a/bin/bitbake-worker
+++ b/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)
 

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