piecrust2: piecrust/baking/baker.py comparison

comparison piecrust/baking/baker.py @ 411:e7b865f8f335

bake: Enable multiprocess baking. Baking is now done by running a worker per CPU, and sending jobs to them. This changes several things across the codebase: * Ability to not cache things related to pages other than the 'main' page (i.e. the page at the bottom of the execution stack). * Decouple the baking process from the bake records, so only the main process keeps track (and modifies) the bake record. * Remove the need for 'batch page getters' and loading a page directly from the page factories. There are various smaller changes too included here, including support for scope performance timers that are saved with the bake record and can be printed out to the console. Yes I got carried away. For testing, the in-memory 'mock' file-system doesn't work anymore, since we're spawning processes, so this is replaced by a 'tmpfs' file-system which is saved in temporary files on disk and deleted after tests have run.

author	Ludovic Chabant <ludovic@chabant.com>
date	Fri, 12 Jun 2015 17:09:19 -0700
parents	c2ca72fb7f0b
children	0e9a94b7fdfa

comparison

equal deleted inserted replaced

-:d1a472464e57
+:e7b865f8f335
+import copy
 import time
 import os.path
+import queue
 import hashlib
 import logging
-import threading
+import multiprocessing
 from piecrust.baking.records import (
-TransitionalBakeRecord, BakeRecordPageEntry)
+BakeRecordEntry, TransitionalBakeRecord, TaxonomyInfo, FirstRenderInfo)
-from piecrust.baking.scheduler import BakeScheduler
+from piecrust.baking.worker import (
-from piecrust.baking.single import (BakingError, PageBaker)
+BakeWorkerJob, LoadJobPayload, RenderFirstSubJobPayload,
-from piecrust.chefutil import format_timed, log_friendly_exception
+BakeJobPayload,
+JOB_LOAD, JOB_RENDER_FIRST, JOB_BAKE)
+from piecrust.chefutil import (
+format_timed_scope, format_timed)
 from piecrust.sources.base import (
 REALM_NAMES, REALM_USER, REALM_THEME)
 logger = logging.getLogger(__name__)
 def __init__(self, app, out_dir, force=False):
 assert app and out_dir
 self.app = app
 self.out_dir = out_dir
 self.force = force
-self.num_workers = app.config.get('baker/workers', 4)
+self.num_workers = app.config.get('baker/workers',
+multiprocessing.cpu_count())
 # Remember what taxonomy pages we should skip
 # (we'll bake them repeatedly later with each taxonomy term)
 self.taxonomy_pages = []
 logger.debug("Gathering taxonomy page paths:")
 for tax in self.app.taxonomies:
 for src in self.app.sources:
-path = tax.resolvePagePath(src.name)
+tax_page_ref = tax.getPageRef(src)
-if path is not None:
+for path in tax_page_ref.possible_paths:
 self.taxonomy_pages.append(path)
 logger.debug(" - %s" % path)
 def bake(self):
 logger.debug("  Bake Output: %s" % self.out_dir)
 logger.debug("  Root URL: %s" % self.app.config.get('site/root'))
 # Get into bake mode.
-start_time = time.clock()
+start_time = time.perf_counter()
 self.app.config.set('baker/is_baking', True)
 self.app.env.base_asset_url_format = '%uri%'
 # Make sure the output directory exists.
 if not os.path.isdir(self.out_dir):
 # Load/create the bake record.
 record = TransitionalBakeRecord()
 record_cache = self.app.cache.getCache('baker')
 record_id = hashlib.md5(self.out_dir.encode('utf8')).hexdigest()
 record_name = record_id + '.record'
+previous_record_path = None
 if not self.force and record_cache.has(record_name):
-t = time.clock()
+with format_timed_scope(logger, "loaded previous bake record",
-record.loadPrevious(record_cache.getCachePath(record_name))
+level=logging.DEBUG, colored=False):
-logger.debug(format_timed(
+previous_record_path = record_cache.getCachePath(record_name)
-t, 'loaded previous bake record',
+record.loadPrevious(previous_record_path)
-colored=False))
 record.current.success = True
 # Figure out if we need to clean the cache because important things
 # have changed.
 self._handleCacheValidity(record)
+# Pre-create all caches.
+for cache_name in ['app', 'baker', 'pages', 'renders']:
+self.app.cache.getCache(cache_name)
 # Gather all sources by realm -- we're going to bake each realm
-# separately so we can handle "overlaying" (i.e. one realm overrides
+# separately so we can handle "overriding" (i.e. one realm overrides
-# another realm's pages).
+# another realm's pages, like the user realm overriding the theme
+# realm).
 sources_by_realm = {}
 for source in self.app.sources:
 srclist = sources_by_realm.setdefault(source.realm, [])
 srclist.append(source)
+# Create the worker processes.
+pool = self._createWorkerPool()
 # Bake the realms.
 realm_list = [REALM_USER, REALM_THEME]
 for realm in realm_list:
 srclist = sources_by_realm.get(realm)
 if srclist is not None:
-self._bakeRealm(record, realm, srclist)
+self._bakeRealm(record, pool, realm, srclist)
 # Bake taxonomies.
-self._bakeTaxonomies(record)
+self._bakeTaxonomies(record, pool)
+# All done with the workers.
+self._terminateWorkerPool(pool)
+# Get the timing information from the workers.
+record.current.timers = {}
+for _ in range(len(pool.workers)):
+try:
+timers = pool.results.get(True, 0.1)
+except queue.Empty:
+logger.error("Didn't get timing information from all workers.")
+break
+for name, val in timers.items():
+main_val = record.current.timers.setdefault(name, 0)
+record.current.timers[name] = main_val + val
 # Delete files from the output.
 self._handleDeletetions(record)
 # Backup previous records.
 if os.path.exists(record_path_next):
 os.remove(record_path_next)
 os.rename(record_path, record_path_next)
 # Save the bake record.
-t = time.clock()
+with format_timed_scope(logger, "saved bake record.",
-record.current.bake_time = time.time()
+level=logging.DEBUG, colored=False):
-record.current.out_dir = self.out_dir
+record.current.bake_time = time.time()
-record.saveCurrent(record_cache.getCachePath(record_name))
+record.current.out_dir = self.out_dir
-logger.debug(format_timed(t, 'saved bake record', colored=False))
+record.saveCurrent(record_cache.getCachePath(record_name))
 # All done.
 self.app.config.set('baker/is_baking', False)
 logger.debug(format_timed(start_time, 'done baking'))
 return record.detach()
 def _handleCacheValidity(self, record):
-start_time = time.clock()
+start_time = time.perf_counter()
 reason = None
 if self.force:
 reason = "ordered to"
 elif not self.app.config.get('__cache_valid'):
 record.incremental_count += 1
 logger.debug(format_timed(
 start_time, "cache is assumed valid",
 colored=False))
-def _bakeRealm(self, record, realm, srclist):
+def _bakeRealm(self, record, pool, realm, srclist):
-# Gather all page factories from the sources and queue them
+start_time = time.perf_counter()
-# for the workers to pick up. Just skip taxonomy pages for now.
+try:
-logger.debug("Baking realm %s" % REALM_NAMES[realm])
+all_factories = []
-pool, queue, abort = self._createWorkerPool(record, self.num_workers)
+for source in srclist:
+factories = source.getPageFactories()
-for source in srclist:
+all_factories += [f for f in factories
-factories = source.getPageFactories()
+if f.path not in self.taxonomy_pages]
+self._loadRealmPages(record, pool, all_factories)
+self._renderRealmPages(record, pool, all_factories)
+self._bakeRealmPages(record, pool, all_factories)
+finally:
+page_count = len(all_factories)
+logger.info(format_timed(
+start_time,
+"baked %d %s pages" %
+(page_count, REALM_NAMES[realm].lower())))
+def _loadRealmPages(self, record, pool, factories):
+with format_timed_scope(logger,
+"loaded %d pages" % len(factories),
+level=logging.DEBUG, colored=False):
 for fac in factories:
-if fac.path in self.taxonomy_pages:
+job = BakeWorkerJob(
-logger.debug(
+JOB_LOAD,
-"Skipping taxonomy page: %s:%s" %
+LoadJobPayload(fac))
-(source.name, fac.ref_spec))
+pool.queue.put_nowait(job)
+def _handler(res):
+# Create the record entry for this page.
+record_entry = BakeRecordEntry(res.source_name, res.path)
+record_entry.config = res.config
+if res.errors:
+record_entry.errors += res.errors
+record.current.success = False
+record.addEntry(record_entry)
+self._waitOnWorkerPool(
+pool,
+expected_result_count=len(factories),
+result_handler=_handler)
+def _renderRealmPages(self, record, pool, factories):
+with format_timed_scope(logger,
+"prepared %d pages" % len(factories),
+level=logging.DEBUG, colored=False):
+expected_result_count = 0
+for fac in factories:
+record_entry = record.getCurrentEntry(fac.path)
+if record_entry.errors:
+logger.debug("Ignoring %s because it had previous "
+"errors." % fac.ref_spec)
 continue
-entry = BakeRecordPageEntry(fac.source.name, fac.rel_path,
+# Make sure the source and the route exist for this page,
-fac.path)
+# otherwise we add errors to the record entry and we'll skip
-record.addEntry(entry)
+# this page for the rest of the bake.
+source = self.app.getSource(fac.source.name)
-route = self.app.getRoute(source.name, fac.metadata,
+if source is None:
+record_entry.errors.append(
+"Can't get source for page: %s" % fac.ref_spec)
+logger.error(record_entry.errors[-1])
+continue
+route = self.app.getRoute(fac.source.name, fac.metadata,
 skip_taxonomies=True)
 if route is None:
-entry.errors.append(
+record_entry.errors.append(
 "Can't get route for page: %s" % fac.ref_spec)
-logger.error(entry.errors[-1])
+logger.error(record_entry.errors[-1])
 continue
-queue.addJob(BakeWorkerJob(fac, route, entry))
+# All good, queue the job.
+job = BakeWorkerJob(
-success = self._waitOnWorkerPool(pool, abort)
+JOB_RENDER_FIRST,
-record.current.success &= success
+RenderFirstSubJobPayload(fac))
+pool.queue.put_nowait(job)
-def _bakeTaxonomies(self, record):
+expected_result_count += 1
-logger.debug("Baking taxonomies")
+def _handler(res):
+entry = record.getCurrentEntry(res.path)
+entry.first_render_info = FirstRenderInfo()
+entry.first_render_info.used_assets = res.used_assets
+entry.first_render_info.used_pagination = \
+res.used_pagination
+entry.first_render_info.pagination_has_more = \
+res.pagination_has_more
+if res.errors:
+entry.errors += res.errors
+record.current.success = False
+self._waitOnWorkerPool(
+pool,
+expected_result_count=expected_result_count,
+result_handler=_handler)
+def _bakeRealmPages(self, record, pool, factories):
+with format_timed_scope(logger,
+"baked %d pages" % len(factories),
+level=logging.DEBUG, colored=False):
+expected_result_count = 0
+for fac in factories:
+if self._queueBakeJob(record, pool, fac):
+expected_result_count += 1
+def _handler(res):
+entry = record.getCurrentEntry(res.path, res.taxonomy_info)
+entry.bake_info = res.bake_info
+if res.errors:
+entry.errors += res.errors
+if entry.has_any_error:
+record.current.success = False
+self._waitOnWorkerPool(
+pool,
+expected_result_count=expected_result_count,
+result_handler=_handler)
+def _bakeTaxonomies(self, record, pool):
+with format_timed_scope(logger, 'built taxonomy buckets',
+level=logging.DEBUG, colored=False):
+buckets = self._buildTaxonomyBuckets(record)
+start_time = time.perf_counter()
+page_count = self._bakeTaxonomyBuckets(record, pool, buckets)
+logger.info(format_timed(start_time,
+"baked %d taxonomy pages." % page_count))
+def _buildTaxonomyBuckets(self, record):
 # Let's see all the taxonomy terms for which we must bake a
 # listing page... first, pre-populate our big map of used terms.
 # For each source name, we have a list of taxonomies, and for each
 # taxonomies, a list of terms, some being 'dirty', some used last
 # time, etc.
 tt_info = buckets[sn][tn]
 tt_info.all_terms.add(terms)
 if not tt_info.dirty_terms.isdisjoint(set(terms)):
 tt_info.dirty_terms.add(terms)
+return buckets
+def _bakeTaxonomyBuckets(self, record, pool, buckets):
 # Start baking those terms.
-pool, queue, abort = self._createWorkerPool(record, self.num_workers)
+expected_result_count = 0
 for source_name, source_taxonomies in buckets.items():
 for tax_name, tt_info in source_taxonomies.items():
 terms = tt_info.dirty_terms
 if len(terms) == 0:
 continue
 logger.debug(
 "Baking '%s' for source '%s': %s" %
 (tax_name, source_name, terms))
 tax = self.app.getTaxonomy(tax_name)
-route = self.app.getTaxonomyRoute(tax_name, source_name)
+source = self.app.getSource(source_name)
-tax_page_ref = tax.getPageRef(source_name)
+tax_page_ref = tax.getPageRef(source)
 if not tax_page_ref.exists:
 logger.debug(
 "No taxonomy page found at '%s', skipping." %
 tax.page_ref)
 continue
 logger.debug(
 "Using taxonomy page: %s:%s" %
 (tax_page_ref.source_name, tax_page_ref.rel_path))
+fac = tax_page_ref.getFactory()
 for term in terms:
-fac = tax_page_ref.getFactory()
 logger.debug(
 "Queuing: %s [%s=%s]" %
 (fac.ref_spec, tax_name, term))
-entry = BakeRecordPageEntry(
+tax_info = TaxonomyInfo(tax_name, source_name, term)
-fac.source.name, fac.rel_path, fac.path,
-(tax_name, term, source_name))
+cur_entry = BakeRecordEntry(
-record.addEntry(entry)
+fac.source.name, fac.path, tax_info)
-queue.addJob(BakeWorkerJob(fac, route, entry))
+record.addEntry(cur_entry)
-success = self._waitOnWorkerPool(pool, abort)
+if self._queueBakeJob(record, pool, fac, tax_info):
-record.current.success &= success
+expected_result_count += 1
+def _handler(res):
+entry = record.getCurrentEntry(res.path, res.taxonomy_info)
+entry.bake_info = res.bake_info
+if res.errors:
+entry.errors += res.errors
+if entry.has_any_error:
+record.current.success = False
+self._waitOnWorkerPool(
+pool,
+expected_result_count=expected_result_count,
+result_handler=_handler)
 # Now we create bake entries for all the terms that were *not* dirty.
 # This is because otherwise, on the next incremental bake, we wouldn't
 # find any entry for those things, and figure that we need to delete
 # their outputs.
 for prev_entry, cur_entry in record.transitions.values():
 # Only consider taxonomy-related entries that don't have any
 # current version.
 if (prev_entry and prev_entry.taxonomy_info and
 not cur_entry):
-sn = prev_entry.source_name
+ti = prev_entry.taxonomy_info
-tn, tt, tsn = prev_entry.taxonomy_info
+tt_info = buckets[ti.source_name][ti.taxonomy_name]
-tt_info = buckets[tsn][tn]
+if ti.term in tt_info.all_terms:
-if tt in tt_info.all_terms:
 logger.debug("Creating unbaked entry for taxonomy "
-"term '%s:%s'." % (tn, tt))
+"term '%s:%s'." % (ti.taxonomy_name, ti.term))
 record.collapseEntry(prev_entry)
 else:
 logger.debug("Taxonomy term '%s:%s' isn't used anymore." %
-(tn, tt))
+(ti.taxonomy_name, ti.term))
+return expected_result_count
+def _queueBakeJob(self, record, pool, fac, tax_info=None):
+# Get the previous (if any) and current entry for this page.
+pair = record.getPreviousAndCurrentEntries(fac.path, tax_info)
+assert pair is not None
+prev_entry, cur_entry = pair
+assert cur_entry is not None
+# Ignore if there were errors in the previous passes.
+if cur_entry.errors:
+logger.debug("Ignoring %s because it had previous "
+"errors." % fac.ref_spec)
+return False
+# Build the route metadata and find the appropriate route.
+route_metadata = copy.deepcopy(fac.metadata)
+if tax_info is not None:
+tax = self.app.getTaxonomy(tax_info.taxonomy_name)
+route = self.app.getTaxonomyRoute(tax_info.taxonomy_name,
+tax_info.source_name)
+slugified_term = route.slugifyTaxonomyTerm(tax_info.term)
+route_metadata[tax.term_name] = slugified_term
+else:
+route = self.app.getRoute(fac.source.name, route_metadata,
+skip_taxonomies=True)
+assert route is not None
+# Figure out if this page is overriden by another previously
+# baked page. This happens for example when the user has
+# made a page that has the same page/URL as a theme page.
+page = fac.buildPage()
+uri = route.getUri(route_metadata, provider=page)
+override_entry = record.getOverrideEntry(page.path, uri)
+if override_entry is not None:
+override_source = self.app.getSource(
+override_entry.source_name)
+if override_source.realm == fac.source.realm:
+cur_entry.errors.append(
+"Page '%s' maps to URL '%s' but is overriden "
+"by page '%s'." %
+(fac.ref_spec, uri, override_entry.path))
+logger.error(cur_entry.errors[-1])
+cur_entry.flags |= BakeRecordEntry.FLAG_OVERRIDEN
+return False
+job = BakeWorkerJob(
+JOB_BAKE,
+BakeJobPayload(fac, route_metadata, prev_entry,
+cur_entry.first_render_info,
+record.dirty_source_names,
+tax_info))
+pool.queue.put_nowait(job)
+return True
 def _handleDeletetions(self, record):
 for path, reason in record.getDeletions():
 logger.debug("Removing '%s': %s" % (path, reason))
 try:
 except OSError:
 # Not a big deal if that file had already been removed
 # by the user.
 pass
-def _createWorkerPool(self, record, pool_size=4):
+def _createWorkerPool(self):
-pool = []
+from piecrust.baking.worker import BakeWorkerContext, worker_func
-queue = BakeScheduler(record)
-abort = threading.Event()
+pool = _WorkerPool()
-for i in range(pool_size):
+for i in range(self.num_workers):
 ctx = BakeWorkerContext(
-self.app, self.out_dir, self.force,
+self.app.root_dir, self.out_dir,
-record, queue, abort)
+pool.queue, pool.results, pool.abort_event,
-worker = BakeWorker(i, ctx)
+force=self.force, debug=self.app.debug)
-pool.append(worker)
+w = multiprocessing.Process(
-return pool, queue, abort
+target=worker_func, args=(i, ctx))
-def _waitOnWorkerPool(self, pool, abort):
-for w in pool:
 w.start()
+pool.workers.append(w)
-success = True
+return pool
+def _terminateWorkerPool(self, pool):
+pool.abort_event.set()
+for w in pool.workers:
+w.join()
+def _waitOnWorkerPool(self, pool,
+expected_result_count=-1, result_handler=None):
+assert result_handler is None or expected_result_count >= 0
+abort_with_exception = None
 try:
-for w in pool:
+if result_handler is None:
-w.join()
+pool.queue.join()
-success &= w.success
+else:
-except KeyboardInterrupt:
+got_count = 0
+while got_count < expected_result_count:
+try:
+res = pool.results.get(True, 10)
+except queue.Empty:
+logger.error(
+"Got %d results, expected %d, and timed-out "
+"for 10 seconds. A worker might be stuck?" %
+(got_count, expected_result_count))
+abort_with_exception = Exception("Worker time-out.")
+break
+got_count += 1
+result_handler(res)
+except KeyboardInterrupt as kiex:
 logger.warning("Bake aborted by user... "
 "waiting for workers to stop.")
-abort.set()
+abort_with_exception = kiex
-for w in pool:
-w.join()
+if abort_with_exception:
-raise
+pool.abort_event.set()
+for w in pool.workers:
-if abort.is_set():
+w.join(2)
-excs = [w.abort_exception for w in pool
+raise abort_with_exception
-if w.abort_exception is not None]
-logger.error("Baking was aborted due to %s error(s):" % len(excs))
-if self.app.debug:
+class _WorkerPool(object):
-for e in excs:
+def __init__(self):
-logger.exception(e)
+self.queue = multiprocessing.JoinableQueue()
-else:
+self.results = multiprocessing.Queue()
-for e in excs:
+self.abort_event = multiprocessing.Event()
-log_friendly_exception(logger, e)
+self.workers = []
-raise BakingError("Baking was aborted due to errors.")
-return success
-class BakeWorkerContext(object):
-def __init__(self, app, out_dir, force, record, work_queue,
-abort_event):
-self.app = app
-self.out_dir = out_dir
-self.force = force
-self.record = record
-self.work_queue = work_queue
-self.abort_event = abort_event
-class BakeWorkerJob(object):
-def __init__(self, factory, route, record_entry):
-self.factory = factory
-self.route = route
-self.record_entry = record_entry
-@property
-def source(self):
-return self.factory.source
-class BakeWorker(threading.Thread):
-def __init__(self, wid, ctx):
-super(BakeWorker, self).__init__(name=('worker%d' % wid))
-self.wid = wid
-self.ctx = ctx
-self.abort_exception = None
-self.success = True
-self._page_baker = PageBaker(
-ctx.app, ctx.out_dir, ctx.force,
-ctx.record)
-def run(self):
-while(not self.ctx.abort_event.is_set()):
-try:
-job = self.ctx.work_queue.getNextJob(wait_timeout=1)
-if job is None:
-logger.debug(
-"[%d] No more work... shutting down." %
-self.wid)
-break
-success = self._unsafeRun(job)
-logger.debug("[%d] Done with page." % self.wid)
-self.ctx.work_queue.onJobFinished(job)
-self.success &= success
-except Exception as ex:
-self.ctx.abort_event.set()
-self.abort_exception = ex
-self.success = False
-logger.debug("[%d] Critical error, aborting." % self.wid)
-if self.ctx.app.debug:
-logger.exception(ex)
-break
-def _unsafeRun(self, job):
-start_time = time.clock()
-entry = job.record_entry
-try:
-self._page_baker.bake(job.factory, job.route, entry)
-except BakingError as ex:
-logger.debug("Got baking error. Adding it to the record.")
-while ex:
-entry.errors.append(str(ex))
-ex = ex.__cause__
-has_error = False
-for e in entry.getAllErrors():
-has_error = True
-logger.error(e)
-if has_error:
-return False
-if entry.was_any_sub_baked:
-first_sub = entry.subs[0]
-friendly_uri = first_sub.out_uri
-if friendly_uri == '':
-friendly_uri = '[main page]'
-friendly_count = ''
-if entry.num_subs > 1:
-friendly_count = ' (%d pages)' % entry.num_subs
-logger.info(format_timed(
-start_time, '[%d] %s%s' %
-(self.wid, friendly_uri, friendly_count)))
-return True
 class _TaxonomyTermsInfo(object):
 def __init__(self):
 self.dirty_terms = set()
 def __str__(self):
 return 'dirty:%s, all:%s' % (self.dirty_terms, self.all_terms)
 def __repr__(self):
 return 'dirty:%s, all:%s' % (self.dirty_terms, self.all_terms)

Mercurial > piecrust2

comparison piecrust/baking/baker.py @ 411:e7b865f8f335