piecrust2: piecrust/baking/baker.py comparison

comparison piecrust/baking/baker.py @ 854:08e02c2a2a1a

core: Keep refactoring, this time to prepare for generator sources. - Make a few APIs simpler. - Content pipelines create their own jobs, so that generator sources can keep aborting in `getContents`, but rely on their pipeline to generate pages for baking.

author	Ludovic Chabant <ludovic@chabant.com>
date	Sun, 04 Jun 2017 23:34:28 -0700
parents	f070a4fc033c
children	448710d84121

comparison

equal deleted inserted replaced

-:f070a4fc033c
+:08e02c2a2a1a
 import time
 import os.path
 import hashlib
 import logging
-from piecrust.baking.worker import BakeJob
 from piecrust.chefutil import (
 format_timed_scope, format_timed)
 from piecrust.environment import ExecutionStats
-from piecrust.pipelines.base import PipelineContext
+from piecrust.pipelines.base import (
+PipelineMergeRecordContext, PipelineManager,
+get_pipeline_name_for_source)
 from piecrust.pipelines.records import (
 MultiRecordHistory, MultiRecord, RecordEntry,
 load_records)
 from piecrust.sources.base import REALM_USER, REALM_THEME
 self.appfactory = appfactory
 self.app = app
 self.out_dir = out_dir
 self.force = force
-self._pipeline_classes = {}
-for pclass in app.plugin_loader.getPipelines():
-self._pipeline_classes[pclass.PIPELINE_NAME] = pclass
 self.allowed_pipelines = allowed_pipelines
 if allowed_pipelines is None:
 self.allowed_pipelines = list(self._pipeline_classes.keys())
-self._records = None
 def bake(self):
 start_time = time.perf_counter()
 logger.debug("  Bake Output: %s" % self.out_dir)
 logger.debug("  Root URL: %s" % self.app.config.get('site/root'))
 with format_timed_scope(logger, "loaded previous bake records",
 level=logging.DEBUG, colored=False):
 previous_records = load_records(records_path)
 else:
 previous_records = MultiRecord()
-self._records = MultiRecord()
+current_records = MultiRecord()
 # Figure out if we need to clean the cache because important things
 # have changed.
 is_cache_valid = self._handleCacheValidity(previous_records,
-self._records)
+current_records)
 if not is_cache_valid:
 previous_records = MultiRecord()
 # Create the bake records history which tracks what's up-to-date
 # or not since last time we baked to the given output folder.
-record_histories = MultiRecordHistory(previous_records, self._records)
+record_histories = MultiRecordHistory(
+previous_records, current_records)
 # Pre-create all caches.
 for cache_name in ['app', 'baker', 'pages', 'renders']:
 self.app.cache.getCache(cache_name)
 # separately so we can handle "overriding" (i.e. one realm overrides
 # another realm's pages, like the user realm overriding the theme
 # realm).
 #
 # Also, create and initialize each pipeline for each source.
-sources_by_realm = {}
+has_any_pp = False
+ppmngr = PipelineManager(
+self.app, self.out_dir, record_histories)
 for source in self.app.sources:
-pname = source.config['pipeline']
+pname = get_pipeline_name_for_source(source)
 if pname in self.allowed_pipelines:
-srclist = sources_by_realm.setdefault(
+ppinfo = ppmngr.createPipeline(source)
-source.config['realm'], [])
+logger.debug(
+"Created pipeline '%s' for source: %s" %
-pp = self._pipeline_classes[pname](source)
+(ppinfo.pipeline.PIPELINE_NAME, source.name))
+has_any_pp = True
-record_name = _get_record_name(source.name, pname)
-record_history = record_histories.getHistory(record_name)
-ppctx = PipelineContext(self.out_dir, record_history,
-force=self.force)
-pp.initialize(ppctx)
-srclist.append((source, pp, ppctx))
 else:
 logger.debug(
 "Skip source '%s' because pipeline '%s' is ignored." %
 (source.name, pname))
+if not has_any_pp:
+raise Exception("The website has no content sources, or the bake "
+"command was invoked with all pipelines filtered "
+"out. There's nothing to do.")
 # Create the worker processes.
-pool = self._createWorkerPool(records_path)
+pool_userdata = _PoolUserData(self, ppmngr, current_records)
+pool = self._createWorkerPool(records_path, pool_userdata)
+realm_list = [REALM_USER, REALM_THEME]
 # Bake the realms -- user first, theme second, so that a user item
 # can override a theme item.
-realm_list = [REALM_USER, REALM_THEME]
+# Do this for as many times as we have pipeline passes left to do.
-for realm in realm_list:
+pp_by_pass_and_realm = {}
-srclist = sources_by_realm.get(realm)
+for ppinfo in ppmngr.getPipelines():
-if srclist is not None:
+pp_by_realm = pp_by_pass_and_realm.setdefault(
-self._bakeRealm(pool, srclist)
+ppinfo.pipeline.PASS_NUM, {})
+pplist = pp_by_realm.setdefault(
-# Handle deletions.
+ppinfo.pipeline.source.config['realm'], [])
-for realm in realm_list:
+pplist.append(ppinfo)
-srclist = sources_by_realm.get(realm)
-if srclist is not None:
+for pp_pass in sorted(pp_by_pass_and_realm.keys()):
-self._deleteStaleOutputs(pool, srclist)
+logger.debug("Pipelines pass %d" % pp_pass)
+pp_by_realm = pp_by_pass_and_realm[pp_pass]
-# Collapse records.
+for realm in realm_list:
-for realm in realm_list:
+pplist = pp_by_realm.get(realm)
-srclist = sources_by_realm.get(realm)
+if pplist is not None:
-if srclist is not None:
+self._bakeRealm(pool, pplist)
-self._collapseRecords(srclist)
+# Handle deletions, collapse records, etc.
+ppmngr.buildHistoryDiffs()
+ppmngr.deleteStaleOutputs()
+ppmngr.collapseRecords()
 # All done with the workers. Close the pool and get reports.
 pool_stats = pool.close()
 total_stats = ExecutionStats()
 for ps in pool_stats:
 if ps is not None:
 total_stats.mergeStats(ps)
-record_histories.current.stats = total_stats
+current_records.stats = total_stats
 # Shutdown the pipelines.
-for realm in realm_list:
+ppmngr.shutdownPipelines()
-srclist = sources_by_realm.get(realm)
-if srclist is not None:
-for _, pp, ppctx in srclist:
-pp.shutdown(ppctx)
 # Backup previous records.
 records_dir, records_fn = os.path.split(records_path)
 records_id, _ = os.path.splitext(records_fn)
 for i in range(8, -1, -1):
 os.rename(records_path_i, records_path_next)
 # Save the bake records.
 with format_timed_scope(logger, "saved bake records.",
 level=logging.DEBUG, colored=False):
-record_histories.current.bake_time = time.time()
+current_records.bake_time = time.time()
-record_histories.current.out_dir = self.out_dir
+current_records.out_dir = self.out_dir
-record_histories.current.save(records_path)
+current_records.save(records_path)
 # All done.
 self.app.config.set('baker/is_baking', False)
 logger.debug(format_timed(start_time, 'done baking'))
-self._records = None
+return current_records
-return record_histories.current
 def _handleCacheValidity(self, previous_records, current_records):
 start_time = time.perf_counter()
 reason = None
 current_records.incremental_count += 1
 logger.debug(format_timed(
 start_time, "cache is assumed valid", colored=False))
 return True
-def _bakeRealm(self, pool, srclist):
+def _bakeRealm(self, pool, pplist):
-for source, pp, ppctx in srclist:
+# Start with the first pass, where we iterate on the content sources'
-logger.debug("Queuing jobs for source '%s' using pipeline '%s'." %
+# items and run jobs on those.
-(source.name, pp.PIPELINE_NAME))
+pool.userdata.cur_pass = 0
-jobs = [BakeJob(source.name, item.spec, item.metadata)
+next_pass_jobs = {}
-for item in source.getAllContents()]
+pool.userdata.next_pass_jobs = next_pass_jobs
+for ppinfo in pplist:
+src = ppinfo.source
+pp = ppinfo.pipeline
+logger.debug(
+"Queuing jobs for source '%s' using pipeline '%s' (pass 0)." %
+(src.name, pp.PIPELINE_NAME))
+next_pass_jobs[src.name] = []
+jobs = pp.createJobs()
 pool.queueJobs(jobs)
 pool.wait()
-def _deleteStaleOutputs(self, pool, srclist):
+# Now let's see if any job created a follow-up job. Let's keep
-for source, pp, ppctx in srclist:
+# processing those jobs as long as they create new ones.
-ppctx.record_history.build()
+pool.userdata.cur_pass = 1
+while True:
-to_delete = pp.getDeletions(ppctx)
+had_any_job = False
-if to_delete is not None:
-for path, reason in to_delete:
+# Make a copy of out next pass jobs and reset the list, so
-logger.debug("Removing '%s': %s" % (path, reason))
+# the first jobs to be processed don't mess it up as we're
-ppctx.current_record.deleted_out_paths.append(path)
+# still iterating on it.
-try:
+next_pass_jobs = pool.userdata.next_pass_jobs
-os.remove(path)
+pool.userdata.next_pass_jobs = {}
-except FileNotFoundError:
-pass
+for sn, jobs in next_pass_jobs.items():
-logger.info('[delete] %s' % path)
+if jobs:
+logger.debug(
-def _collapseRecords(self, srclist):
+"Queuing jobs for source '%s' (pass %d)." %
-for source, pp, ppctx in srclist:
+(sn, pool.userdata.cur_pass))
-pp.collapseRecords(ppctx)
+pool.userdata.next_pass_jobs[sn] = []
+pool.queueJobs(jobs)
+had_any_job = True
+if not had_any_job:
+break
+pool.wait()
+pool.userdata.cur_pass += 1
 def _logErrors(self, item_spec, errors):
 logger.error("Errors found in %s:" % item_spec)
 for e in errors:
 logger.error("  " + e)
-def _createWorkerPool(self, previous_records_path):
+def _createWorkerPool(self, previous_records_path, pool_userdata):
 from piecrust.workerpool import WorkerPool
 from piecrust.baking.worker import BakeWorkerContext, BakeWorker
 worker_count = self.app.config.get('baker/workers')
 batch_size = self.app.config.get('baker/batch_size')
 worker_count=worker_count,
 batch_size=batch_size,
 worker_class=BakeWorker,
 initargs=(ctx,),
 callback=self._handleWorkerResult,
-error_callback=self._handleWorkerError)
+error_callback=self._handleWorkerError,
+userdata=pool_userdata)
 return pool
-def _handleWorkerResult(self, job, res):
+def _handleWorkerResult(self, job, res, userdata):
-record_name = _get_record_name(job.source_name, res.pipeline_name)
+cur_pass = userdata.cur_pass
-record = self._records.getRecord(record_name)
+record = userdata.records.getRecord(job.record_name)
-record.entries.append(res.record_entry)
+if cur_pass == 0:
+record.addEntry(res.record_entry)
+else:
+ppinfo = userdata.ppmngr.getPipeline(job.source_name)
+ppmrctx = PipelineMergeRecordContext(
+record, job, cur_pass)
+ppinfo.pipeline.mergeRecordEntry(res.record_entry, ppmrctx)
+npj = res.next_pass_job
+if npj is not None:
+npj.data['pass'] = cur_pass + 1
+userdata.next_pass_jobs[job.source_name].append(npj)
 if not res.record_entry.success:
 record.success = False
-self._records.success = False
+userdata.records.success = False
-self._logErrors(job.item_spec, res.record_entry.errors)
+self._logErrors(job.content_item.spec, res.record_entry.errors)
-def _handleWorkerError(self, job, exc_data):
+def _handleWorkerError(self, job, exc_data, userdata):
-e = RecordEntry()
+cur_pass = userdata.cur_pass
-e.item_spec = job.item_spec
+record = userdata.records.getRecord(job.record_name)
-e.errors.append(str(exc_data))
+if cur_pass == 0:
-ppname = self.app.getSource(job.source_name).config['pipeline']
+ppinfo = userdata.ppmngr.getPipeline(job.source_name)
-record_name = _get_record_name(job.source_name, ppname)
+entry_class = ppinfo.pipeline.RECORD_ENTRY_CLASS or RecordEntry
-record_name = self._getRecordName(job)
+e = entry_class()
-record = self._records.getRecord(record_name)
+e.item_spec = job.content_item.spec
-record.entries.append(e)
+e.errors.append(str(exc_data))
+record.addEntry(e)
+else:
+e = record.getEntry(job.content_item.spec)
+e.errors.append(str(exc_data))
 record.success = False
-self._records.success = False
+userdata.records.success = False
-self._logErrors(job.item_spec, e.errors)
+self._logErrors(job.content_item.spec, e.errors)
 if self.app.debug:
 logger.error(exc_data.traceback)
-def _get_record_name(source_name, pipeline_name):
+class _PoolUserData:
-return '%s@%s' % (source_name, pipeline_name)
+def __init__(self, baker, ppmngr, current_records):
+self.baker = baker
+self.ppmngr = ppmngr
+self.records = current_records
+self.cur_pass = 0
+self.next_pass_jobs = {}

Mercurial > piecrust2

comparison piecrust/baking/baker.py @ 854:08e02c2a2a1a