Re: [bitbake-devel] [PATCH 2/2] cooker: Use a queue to feed parsing jobs

[Richard Purdie <richard.purdie@linuxfoundation.org>]

On Thu, 2025-07-03 at 08:50 -0600, Joshua Watt wrote:
> On Thu, Jul 3, 2025 at 8:30 AM Richard Purdie
> <richard.purdie@linuxfoundation.org> wrote:
> > 
> > On Thu, 2025-07-03 at 08:27 -0600, Joshua Watt wrote:
> > > On Wed, Jul 2, 2025 at 4:24 PM Richard Purdie via
> > > lists.openembedded.org
> > > <richard.purdie=linuxfoundation.org@lists.openembedded.org>
> > > wrote:
> > > > 
> > > > Curerntly, recipes to parse are split into equal groups and
> > > > passed to
> > > > each parse thread at the start of parsing. We can replace this
> > > > with
> > > > a queue and collect a new job as each parsing process becomes
> > > > idle
> > > > to better spread load in the case of slow parsing jobs.
> > > > 
> > > > Some of the data we need has to be passed in at fork time since
> > > > it
> > > > can't be pickled, so the job to parse is only referenced as an
> > > > index
> > > > in that list.
> > > > 
> > > > This should better spread load for slow to parse recipes such
> > > > as those
> > > > with many class extensions.
> > > > 
> > > > Signed-off-by: Richard Purdie
> > > > <richard.purdie@linuxfoundation.org>
> > > > ---
> > > >  lib/bb/cooker.py | 29 ++++++++++++++++++-----------
> > > >  1 file changed, 18 insertions(+), 11 deletions(-)
> > > > 
> > > > diff --git a/lib/bb/cooker.py b/lib/bb/cooker.py
> > > > index 91e3ee025ea..e88ad24cf61 100644
> > > > --- a/lib/bb/cooker.py
> > > > +++ b/lib/bb/cooker.py
> > > > @@ -1998,8 +1998,9 @@ class ParsingFailure(Exception):
> > > >          Exception.__init__(self, realexception, recipe)
> > > > 
> > > >  class Parser(multiprocessing.Process):
> > > > -    def __init__(self, jobs, results, quit, profile):
> > > > +    def __init__(self, jobs, jobid_queue, results, quit,
> > > > profile):
> > > >          self.jobs = jobs
> > > > +        self.jobid_queue = jobid_queue
> > > >          self.results = results
> > > >          self.quit = quit
> > > >          multiprocessing.Process.__init__(self)
> > > > @@ -2064,12 +2065,14 @@ class Parser(multiprocessing.Process):
> > > >                  if self.quit.is_set():
> > > >                      break
> > > > 
> > > > -                job = None
> > > > +                jobid = None
> > > >                  try:
> > > > -                    job = self.jobs.pop()
> > > > -                except IndexError:
> > > > +                    jobid = self.jobid_queue.get(True, 0.5)
> > > > +                except (ValueError, OSError):
> > > >                      havejobs = False
> > > > -                if job:
> > > > +
> > > > +                if jobid is not None:
> > > > +                    job = self.jobs[jobid]
> > > >                      result = self.parse(*job)
> > > >                      # Clear the siggen cache after parsing to
> > > > control memory usage, its huge
> > > >                      bb.parse.siggen.postparsing_clean_cache()
> > > > @@ -2082,6 +2085,7 @@ class Parser(multiprocessing.Process):
> > > >                      except queue.Full:
> > > >                          pending.append(result)
> > > >          finally:
> > > > +            self.jobs.close()
> > > >              self.results.close()
> > > >              self.results.join_thread()
> > > > 
> > > > @@ -2134,13 +2138,13 @@ class CookerParser(object):
> > > > 
> > > >          self.bb_caches =
> > > > bb.cache.MulticonfigCache(self.cfgbuilder, self.cfghash,
> > > > cooker.caches_array)
> > > >          self.fromcache = set()
> > > > -        self.willparse = set()
> > > > +        self.willparse = []
> > > >          for mc in self.cooker.multiconfigs:
> > > >              for filename in self.mcfilelist[mc]:
> > > >                  appends =
> > > > self.cooker.collections[mc].get_file_appends(filename)
> > > >                  layername =
> > > > self.cooker.collections[mc].calc_bbfile_priority(filename)[2]
> > > >                  if not self.bb_caches[mc].cacheValid(filename,
> > > > appends):
> > > > -                    self.willparse.add((mc,
> > > > self.bb_caches[mc], filename, appends, layername))
> > > > +                    self.willparse.append((mc,
> > > > self.bb_caches[mc], filename, appends, layername))
> > > >                  else:
> > > >                      self.fromcache.add((mc,
> > > > self.bb_caches[mc], filename, appends, layername))
> > > > 
> > > > @@ -2159,22 +2163,25 @@ class CookerParser(object):
> > > >      def start(self):
> > > >          self.results = self.load_cached()
> > > >          self.processes = []
> > > > +
> > > >          if self.toparse:
> > > >              bb.event.fire(bb.event.ParseStarted(self.toparse),
> > > > self.cfgdata)
> > > > 
> > > > +            self.toparse_queue =
> > > > multiprocessing.Queue(len(self.willparse))
> > > >              self.parser_quit = multiprocessing.Event()
> > > >              self.result_queue = multiprocessing.Queue()
> > > > 
> > > > -            def chunkify(lst,n):
> > > > -                return [lst[i::n] for i in range(n)]
> > > > -            self.jobs = chunkify(list(self.willparse),
> > > > self.num_processes)
> > > > +            for jobid in range(len(self.willparse)):
> > > > +                self.toparse_queue.put(jobid)
> > > 
> > > It's generally not a good idea to assume you can push all the
> > > items on
> > > the queue before there are any consumers; if the queue fills,
> > > this
> > > will deadlock. It would be better to push the items to the queue
> > > after
> > > creating the processes.
> > 
> > True. I did make sure the queue was large enough above however it
> > would
> > depend on the size of the job entries.
> > 
> > >  If you do that, I also don't see any reason to
> > > push the indexes instead of the jobs directly; basically instead
> > > of
> > > "chunkifying" and pre-determining the assignment of jobs to
> > > processes,
> > > have them all pull from one queue of jobs.
> > 
> > I added this comment to the patch in master next:
> > 
> > # Have to pass in willparse at fork time so all parsing processes
> > have the unpickleable data
> > # then access it by index from the parse queue.
> > 
> > 
> > As if you don't do that, it can't rebuild the data as the cache
> > objects
> > aren't pickleable :(
> 
> Ah, OK. We must be relying on the multiprocess being implemented with
> fork() then to pass the data. That's reasonable and the indexes make
> sense (but we should still push them after creating the process to
> avoid the possibility of deadlock)

I changed it to queue after starting the parser processes however I
think they exit early due to no work and it causes occasional build
failures that way around so the code is going to need more extensive
changes.

Cheers,

Richard