[PATCH 0/2] Resolve parsing hang issues

[PATCH 0/2] Resolve parsing hang issues

[Christopher Larson]

From: Christopher Larson <chris_larson@mentor.com>

*** BLURB HERE ***
The following changes since commit db689a99beffea1a285cdfc74a58fe73f1666987:

  Revert the switch to futures for now (2012-02-03 08:12:56 -0700)

are available in the git repository at:
  git://github.com/kergoth/bitbake.git hang-issue-fix
  https://github.com/kergoth/bitbake/tree/hang-issue-fix

Christopher Larson (2):
  cooker: roll our own process pool
  knotty: shut down more cleanly

 lib/bb/cooker.py    |  155 ++++++++++++++++++++++++++++++++++++++++++---------
 lib/bb/ui/knotty.py |   15 ++---
 2 files changed, 136 insertions(+), 34 deletions(-)

-- 
1.7.7

[PATCH 1/2] cooker: roll our own process pool

[Christopher Larson]

From: Christopher Larson <chris_larson@mentor.com>

This fixes the hang issue encountered with parse errors. The underlying issue
seems to have been the pool.terminate(). This sends SIGTERM to each of the
multiprocessing pool's processes, however, a python process terminating in
this fashion can corrupt any queues it's interacting with, causing a number of
problems for us (e.g. the queue that sends events to the UI).

So instead of using multiprocessing's pool, we roll our own, with the ability
to cancel the work. In the very long term, the python concurrent.futures
module introduced in python 3.2 could be used to resolve this as well.

Signed-off-by: Christopher Larson <chris_larson@mentor.com>
---
 lib/bb/cooker.py |  155 +++++++++++++++++++++++++++++++++++++++++++++---------
 1 files changed, 129 insertions(+), 26 deletions(-)

diff --git a/lib/bb/cooker.py b/lib/bb/cooker.py
index bb09dff..295441b 100644
--- a/lib/bb/cooker.py
+++ b/lib/bb/cooker.py
@@ -36,6 +36,7 @@ from functools import wraps
 from collections import defaultdict
 import bb, bb.exceptions, bb.command
 from bb import utils, data, parse, event, cache, providers, taskdata, runqueue
+import Queue
 import prserv.serv
 
 logger      = logging.getLogger("BitBake")
@@ -1402,20 +1403,87 @@ class ParsingFailure(Exception):
         self.recipe = recipe
         Exception.__init__(self, realexception, recipe)
 
-def parse_file(task):
-    filename, appends, caches_array = task
-    try:
-        return True, bb.cache.Cache.parse(filename, appends, parse_file.cfg, caches_array)
-    except Exception as exc:
-        tb = sys.exc_info()[2]
-        exc.recipe = filename
-        exc.traceback = list(bb.exceptions.extract_traceback(tb, context=3))
-        raise exc
-    # Need to turn BaseExceptions into Exceptions here so we gracefully shutdown
-    # and for example a worker thread doesn't just exit on its own in response to
-    # a SystemExit event for example.
-    except BaseException as exc:
-        raise ParsingFailure(exc, filename)
+class Feeder(multiprocessing.Process):
+    def __init__(self, jobs, to_parsers, quit):
+        self.quit = quit
+        self.jobs = jobs
+        self.to_parsers = to_parsers
+        multiprocessing.Process.__init__(self)
+
+    def run(self):
+        while True:
+            try:
+                quit = self.quit.get_nowait()
+            except Queue.Empty:
+                pass
+            else:
+                if quit == 'cancel':
+                    self.to_parsers.cancel_join_thread()
+                break
+
+            try:
+                job = self.jobs.pop()
+            except IndexError:
+                break
+
+            try:
+                self.to_parsers.put(job, timeout=0.5)
+            except Queue.Full:
+                self.jobs.insert(0, job)
+                continue
+
+class Parser(multiprocessing.Process):
+    def __init__(self, jobs, results, quit, init):
+        self.jobs = jobs
+        self.results = results
+        self.quit = quit
+        self.init = init
+        multiprocessing.Process.__init__(self)
+
+    def run(self):
+        if self.init:
+            self.init()
+
+        pending = []
+        while True:
+            try:
+                self.quit.get_nowait()
+            except Queue.Empty:
+                pass
+            else:
+                self.results.cancel_join_thread()
+                break
+
+            if pending:
+                result = pending.pop()
+            else:
+                try:
+                    job = self.jobs.get(0.25)
+                except Queue.Empty:
+                    continue
+
+                if job is None:
+                    break
+                result = self.parse(*job)
+
+            try:
+                self.results.put(result, 0.25)
+            except Queue.Full:
+                pending.append(result)
+
+    def parse(self, filename, appends, caches_array):
+        try:
+            return True, bb.cache.Cache.parse(filename, appends, self.cfg, caches_array)
+        except Exception as exc:
+            tb = sys.exc_info()[2]
+            exc.recipe = filename
+            exc.traceback = list(bb.exceptions.extract_traceback(tb, context=3))
+            return True, exc
+        # Need to turn BaseExceptions into Exceptions here so we gracefully shutdown
+        # and for example a worker thread doesn't just exit on its own in response to
+        # a SystemExit event for example.
+        except BaseException as exc:
+            return True, ParsingFailure(exc, filename)
 
 class CookerParser(object):
     def __init__(self, cooker, filelist, masked):
@@ -1452,22 +1520,28 @@ class CookerParser(object):
         self.start()
 
     def start(self):
-        def init(cfg):
-            parse_file.cfg = cfg
-            multiprocessing.util.Finalize(None, bb.codeparser.parser_cache_save, args=(self.cooker.configuration.data, ), exitpriority=1)
-
         self.results = self.load_cached()
-
+        self.processes = []
         if self.toparse:
             bb.event.fire(bb.event.ParseStarted(self.toparse), self.cfgdata)
+            def init():
+                Parser.cfg = self.cfgdata
+                multiprocessing.util.Finalize(None, bb.codeparser.parser_cache_save, args=(self.cfgdata,), exitpriority=1)
 
-            self.pool = multiprocessing.Pool(self.num_processes, init, [self.cfgdata])
-            parsed = self.pool.imap(parse_file, self.willparse)
-            self.pool.close()
+            self.feeder_quit = multiprocessing.Queue(maxsize=1)
+            self.parser_quit = multiprocessing.Queue(maxsize=self.num_processes)
+            self.jobs = multiprocessing.Queue(maxsize=self.num_processes)
+            self.result_queue = multiprocessing.Queue()
+            self.feeder = Feeder(self.willparse, self.jobs, self.feeder_quit)
+            self.feeder.start()
+            for i in range(1, self.num_processes):
+                parser = Parser(self.jobs, self.result_queue, self.parser_quit, init)
+                parser.start()
+                self.processes.append(parser)
 
-            self.results = itertools.chain(self.results, parsed)
+            self.results = itertools.chain(self.results, self.parse_generator())
 
-    def shutdown(self, clean=True):
+    def shutdown(self, clean=True, force=False):
         if not self.toparse:
             return
 
@@ -1477,9 +1551,22 @@ class CookerParser(object):
                                             self.virtuals, self.error,
                                             self.total)
             bb.event.fire(event, self.cfgdata)
+            self.feeder_quit.put(None)
+            for process in self.processes:
+                self.jobs.put(None)
         else:
-            self.pool.terminate()
-        self.pool.join()
+            self.feeder_quit.put('cancel')
+
+            self.parser_quit.cancel_join_thread()
+            for process in self.processes:
+                self.parser_quit.put(None)
+
+            self.jobs.cancel_join_thread()
+            sys.exit(1)
+
+        for process in self.processes:
+            process.join()
+        self.feeder.join()
 
         sync = threading.Thread(target=self.bb_cache.sync)
         sync.start()
@@ -1491,6 +1578,22 @@ class CookerParser(object):
             cached, infos = self.bb_cache.load(filename, appends, self.cfgdata)
             yield not cached, infos
 
+    def parse_generator(self):
+        while True:
+            if self.parsed >= self.toparse:
+                break
+
+            try:
+                result = self.result_queue.get(0.25)
+            except Queue.Empty:
+                pass
+            else:
+                value = result[1]
+                if isinstance(value, BaseException):
+                    raise value
+                else:
+                    yield result
+
     def parse_next(self):
         try:
             parsed, result = self.results.next()
-- 
1.7.7

[PATCH 2/2] knotty: shut down more cleanly

[Christopher Larson]

From: Christopher Larson <chris_larson@mentor.com>

This ensures that when we are shutting down, we drain the queue of events from
the server. This ensures that the server never gets hung up on the join of the
feeder thread associated with the event queue, thereby avoiding a bitbake hang
issue.

Also, since the third keyboard interrupt now behaves the same as the second,
removed that message.

Signed-off-by: Christopher Larson <chris_larson@mentor.com>
---
 lib/bb/ui/knotty.py |   15 +++++++--------
 1 files changed, 7 insertions(+), 8 deletions(-)

diff --git a/lib/bb/ui/knotty.py b/lib/bb/ui/knotty.py
index e1d42f7..22370ef 100644
--- a/lib/bb/ui/knotty.py
+++ b/lib/bb/ui/knotty.py
@@ -118,6 +118,8 @@ def main(server, eventHandler):
         try:
             event = eventHandler.waitEvent(0.25)
             if event is None:
+                if shutdown > 1:
+                    break
                 continue
             helper.eventHandler(event)
             if isinstance(event, bb.runqueue.runQueueExitWait):
@@ -199,18 +201,18 @@ def main(server, eventHandler):
                 print("Loaded %d entries from dependency cache." % event.num_entries)
                 continue
 
-            if isinstance(event, bb.command.CommandCompleted):
-                break
             if isinstance(event, bb.command.CommandFailed):
                 return_value = event.exitcode
                 logger.error("Command execution failed: %s", event.error)
-                break
+                shutdown = 2
+                continue
             if isinstance(event, bb.command.CommandExit):
                 if not return_value:
                     return_value = event.exitcode
                 continue
-            if isinstance(event, bb.cooker.CookerExit):
-                break
+            if isinstance(event, (bb.command.CommandCompleted, bb.cooker.CookerExit)):
+                shutdown = 2
+                continue
             if isinstance(event, bb.event.MultipleProviders):
                 logger.info("multiple providers are available for %s%s (%s)", event._is_runtime and "runtime " or "",
                             event._item,
@@ -269,9 +271,6 @@ def main(server, eventHandler):
             if ioerror.args[0] == 4:
                 pass
         except KeyboardInterrupt:
-            if shutdown == 2:
-                print("\nThird Keyboard Interrupt, exit.\n")
-                break
             if shutdown == 1:
                 print("\nSecond Keyboard Interrupt, stopping...\n")
                 server.runCommand(["stateStop"])
-- 
1.7.7

Re: [PATCH 0/2] Resolve parsing hang issues

[Chris Larson]

Bah, sent to the wrong list. Hate when I do that.

On Tue, Feb 7, 2012 at 9:13 PM, Christopher Larson <kergoth@gmail.com> wrote:
> From: Christopher Larson <chris_larson@mentor.com>
>
> *** BLURB HERE ***
> The following changes since commit db689a99beffea1a285cdfc74a58fe73f1666987:
>
>  Revert the switch to futures for now (2012-02-03 08:12:56 -0700)
>
> are available in the git repository at:
>  git://github.com/kergoth/bitbake.git hang-issue-fix
>  https://github.com/kergoth/bitbake/tree/hang-issue-fix
>
> Christopher Larson (2):
>  cooker: roll our own process pool
>  knotty: shut down more cleanly
>
>  lib/bb/cooker.py    |  155 ++++++++++++++++++++++++++++++++++++++++++---------
>  lib/bb/ui/knotty.py |   15 ++---
>  2 files changed, 136 insertions(+), 34 deletions(-)
>
> --
> 1.7.7
>
-- 
Christopher Larson