# HG changeset patch # User cmlenz # Date 1127922567 0 # Node ID a8c9dd7e3f7180a87307623befa3bbe99e5b71e2 # Parent 014bc6c29dffea00fea003b1d22e32a80f86c230 * Cleanup and documentation for the `BuildQueue` class added in [236]. * Moved duplicate code from the master and the web UI into `collect_changes` function. diff --git a/bitten/master.py b/bitten/master.py --- a/bitten/master.py +++ b/bitten/master.py @@ -67,15 +67,12 @@ available_slaves = set([name for name in self.handlers if not self.handlers[name].building]) - for queue in self.queues[:]: + for idx, queue in enumerate(self.queues[:]): build, slave = queue.get_next_pending_build(available_slaves) if build: self.handlers[slave].send_initiation(queue, build) available_slaves.discard(slave) - - # Round robin - self.queues.remove(queue) - self.queues.append(queue) + self.queues.append(self.queues.pop(idx)) # Round robin def register(self, handler): any_match = False @@ -410,6 +407,9 @@ envs = [] for arg in args: + if not os.path.isdir(arg): + log.warning('Ignoring %s: not a directory', arg) + continue env = Environment(arg) if BuildSystem(env): if env.needs_upgrade(): diff --git a/bitten/queue.py b/bitten/queue.py --- a/bitten/queue.py +++ b/bitten/queue.py @@ -18,14 +18,62 @@ log = logging.getLogger('bitten.queue') +def collect_changes(repos, config): + """Collect all changes for a build configuration that either have already + been built, or still need to be built. + + This function is a generator that yields `(platform, rev, build)` tuples, + where `platform` is a `TargetPlatform` object, `rev` is the identifier of + the changeset, and `build` is a `Build` object or `None`. + """ + env = config.env + node = repos.get_node(config.path) + + for path, rev, chg in node.get_history(): + + # Don't follow moves/copies + if path != repos.normalize_path(config.path): + break + + # Make sure the repository directory isn't empty at this + # revision + old_node = repos.get_node(path, rev) + is_empty = True + for entry in old_node.get_entries(): + is_empty = False + break + if is_empty: + continue + + # For every target platform, check whether there's a build + # of this revision + for platform in TargetPlatform.select(env, config.name): + builds = list(Build.select(env, config.name, rev, platform.id)) + if builds: + build = builds[0] + else: + build = None + + yield platform, rev, build + + class BuildQueue(object): - """Enapsulates the build queue of an environment.""" + """Enapsulates the build queue of an environment. + + A build queue manages the the registration of build slaves, creation and + removal of snapshot archives, and detection of repository revisions that + need to be built. + """ def __init__(self, env): + """Create the build queue. + + @param env: The Trac environment + """ self.env = env self.slaves = {} # Sets of slave names keyed by target platform ID - # path to generated snapshot archives, key is (config name, revision) + # Paths to generated snapshot archives, key is (config name, revision) self.snapshots = {} for config in BuildConfig.select(self.env): snapshots = archive.index(self.env, prefix=config.name) @@ -59,65 +107,49 @@ # Find a slave for the build platform that is not already building # something else slaves = self.slaves.get(build.platform, []) - for slave in [name for name in slaves if name in available_slaves]: - slaves.remove(slave) - slaves.append(slave) + for idx, slave in enumerate([name for name in slaves if name + in available_slaves]): + slaves.append(slaves.pop(idx)) # Round robin return build, slave return None, None def populate(self): + """Add a build for the next change on each build configuration to the + queue. + + The next change is the latest repository check-in for which there isn't + a corresponding build on each target platform. Repeatedly calling this + method will eventually result in the entire change history of the build + configuration being in the build queue. + """ repos = self.env.get_repository() try: repos.sync() - db = self.env.get_db_cnx() - for config in BuildConfig.select(self.env, db=db): - log.debug('Checking for changes to "%s" at %s', config.label, - config.path) - node = repos.get_node(config.path) - for path, rev, chg in node.get_history(): - - # Don't follow moves/copies - if path != repos.normalize_path(config.path): - break - - # Make sure the repository directory isn't empty at this - # revision - old_node = repos.get_node(path, rev) - is_empty = True - for entry in old_node.get_entries(): - is_empty = False - break - if is_empty: - continue - - enqueued = False - for platform in TargetPlatform.select(self.env, - config.name, db=db): - # Check whether this revision of the configuration has - # already been built on this platform - builds = Build.select(self.env, config.name, rev, - platform.id, db=db) - if not list(builds): - log.info('Enqueuing build of configuration "%s" at ' - 'revision [%s] on %s', config.name, rev, - platform.name) - build = Build(self.env) - build.config = config.name - build.rev = str(rev) - build.rev_time = repos.get_changeset(rev).date - build.platform = platform.id - build.insert(db) - enqueued = True - if enqueued: - db.commit() + for config in BuildConfig.select(self.env): + for platform, rev, build in collect_changes(repos, config): + if build is None: + log.info('Enqueuing build of configuration "%s" at ' + 'revision [%s] on %s', config.name, rev, + platform.name) + build = Build(self.env) + build.config = config.name + build.rev = str(rev) + build.rev_time = repos.get_changeset(rev).date + build.platform = platform.id + build.insert() break finally: repos.close() def reset_orphaned_builds(self): - # Reset all in-progress builds + """Reset all in-progress builds to `PENDING` state. + + This is used to cleanup after a crash of the build master process, + which would leave in-progress builds in the database that aren't + actually being built because the slaves have disconnected. + """ db = self.env.get_db_cnx() for build in Build.select(self.env, status=Build.IN_PROGRESS, db=db): build.status = Build.PENDING @@ -132,6 +164,16 @@ # Snapshot management def get_snapshot(self, build, format, create=False): + """Return the absolute path to a snapshot archive for the given build. + The archive can be created if it doesn't exist yet. + + @param build: The `Build` object + @param format: The archive format (one of `gzip`, `bzip2` or `zip`) + @param create: Whether the archive should be created if it doesn't exist + yet + @return: The absolute path to the create archive file, or None if the + snapshot doesn't exist and wasn't created + """ snapshot = self.snapshots.get((build.config, build.rev, format)) if create and snapshot is None: config = BuildConfig.fetch(self.env, build.config) @@ -142,7 +184,14 @@ return snapshot def remove_unused_snapshots(self): + """Find any previously created snapshot archives that are no longer + needed because all corresponding builds have already been completed. + + This method should be called in regular intervals to keep the total + disk space occupied by the snapshot archives to a minimum. + """ log.debug('Checking for unused snapshot archives...') + for (config, rev, format), path in self.snapshots.items(): keep = False for build in Build.select(self.env, config=config, rev=rev): @@ -157,6 +206,16 @@ # Slave registry def register_slave(self, name, properties): + """Register a build slave with the queue. + + @param name: The name of the slave + @param properties: A `dict` containing the properties of the slave + @return: whether the registration was successful + + This method tries to match the slave against the configured target + platforms. Only if it matches at least one platform will the + registration be successful. + """ any_match = False for config in BuildConfig.select(self.env): for platform in TargetPlatform.select(self.env, config=config.name): @@ -182,6 +241,13 @@ return any_match def unregister_slave(self, name): + """Unregister a build slave. + + @param name: The name of the slave + + This method removes the slave from the registry, and also resets any + in-progress builds by this slave to `PENDING` state. + """ for slaves in self.slaves.values(): if name in slaves: slaves.remove(name) diff --git a/bitten/trac_ext/tests/web_ui.py b/bitten/trac_ext/tests/web_ui.py --- a/bitten/trac_ext/tests/web_ui.py +++ b/bitten/trac_ext/tests/web_ui.py @@ -39,7 +39,8 @@ 'DefaultPermissionStore') # Hook up a dummy repository - repos = Mock(get_node=lambda path: Mock(get_history=lambda: [])) + repos = Mock(get_node=lambda path: Mock(get_history=lambda: []), + sync=lambda: None) self.env.get_repository = lambda x: repos def tearDown(self): diff --git a/bitten/trac_ext/web_ui.py b/bitten/trac_ext/web_ui.py --- a/bitten/trac_ext/web_ui.py +++ b/bitten/trac_ext/web_ui.py @@ -20,6 +20,7 @@ from trac.wiki import wiki_to_html from bitten.model import BuildConfig, TargetPlatform, Build, BuildStep, \ BuildLog, Report +from bitten.queue import collect_changes from bitten.trac_ext.api import ILogFormatter, IReportSummarizer _status_label = {Build.IN_PROGRESS: 'in progress', @@ -325,7 +326,7 @@ } req.hdf['page.mode'] = 'view_config' - platforms = TargetPlatform.select(self.env, config=config_name) + platforms = list(TargetPlatform.select(self.env, config=config_name)) req.hdf['config.platforms'] = [ {'name': platform.name, 'id': platform.id} for platform in platforms ] @@ -348,52 +349,35 @@ more = False req.hdf['page.number'] = page + builds_per_page = 12 * len(platforms) repos = self.env.get_repository(req.authname) - try: - root = repos.get_node(config.path) - idx = 0 - for path, rev, chg in root.get_history(): - # Don't follow moves/copies - if path != repos.normalize_path(config.path): - break - # If the directory was empty at that revision, it isn't built - old_node = repos.get_node(path, rev) - is_empty = True - for entry in old_node.get_entries(): - is_empty = False - break - if is_empty: - continue - - if idx < (page - 1) * 12: - idx += 1 - continue + repos.sync() + idx = 0 + for platform, rev, build in collect_changes(repos, config): + if idx < (page - 1) * builds_per_page: + idx += 1 + continue - prefix = 'config.builds.%d' % rev - req.hdf[prefix + '.href'] = self.env.href.changeset(rev) - for build in Build.select(self.env, config=config.name, rev=rev): - if build.status == Build.PENDING: - continue - build_hdf = _build_to_hdf(self.env, req, build) - req.hdf['%s.%s' % (prefix, build.platform)] = build_hdf + prefix = 'config.builds.%d' % rev + req.hdf[prefix + '.href'] = self.env.href.changeset(rev) + if build and build.status != Build.PENDING: + build_hdf = _build_to_hdf(self.env, req, build) + req.hdf['%s.%s' % (prefix, platform.id)] = build_hdf - idx += 1 - if idx >= page * 12: - more = True - break + idx += 1 + if idx >= page * builds_per_page: + more = True + break - if page > 1: - if page == 2: - prev_href = self.env.href.build(config.name) - else: - prev_href = self.env.href.build(config.name, page=page - 1) - add_link(req, 'prev', prev_href, 'Previous Page') - if more: - next_href = self.env.href.build(config.name, page=page + 1) - add_link(req, 'next', next_href, 'Next Page') - - except TracError: - self.log.error('Error accessing repository info', exc_info=True) + if page > 1: + if page == 2: + prev_href = self.env.href.build(config.name) + else: + prev_href = self.env.href.build(config.name, page=page - 1) + add_link(req, 'prev', prev_href, 'Previous Page') + if more: + next_href = self.env.href.build(config.name, page=page + 1) + add_link(req, 'next', next_href, 'Next Page') def _render_config_confirm(self, req, config_name): req.perm.assert_permission('BUILD_DELETE')