computer-playbook/roles/web-app-docs/files/python/infinito_docs/build_queue.py

"""The build queue, the cross-replica builder lock and the daemon loop.

``Queue`` is a mixin of :class:`infinito_docs.library.Library`. A marker is a
file named ``version`` or ``version:code``; the background lane holds the
builds nobody is waiting on - the daemon's own prefetch and every language no
visitor has asked for.
"""

from __future__ import annotations

import fcntl
import os
import shutil
import subprocess
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor

from infinito_docs.builder import DEPLOYED, LATEST, QUEUE_SEPARATOR

POLL_SECONDS = 2
BACKGROUND_LANE = "background"
KEPT_TAGS = 3


class Queue:
    def request(self, version, code=None, background=False):
        """Queue a build of ``version``, or of one of its translated sites.

        Args:
            version: ``latest`` or a release tag.
            code: ISO 639-1 code to build instead of the version's own site.
            background: queue behind everything a visitor is waiting on.
        """
        head, _ = self.refs()
        if not head and version != DEPLOYED:
            return
        if code is None:
            if self._current(version, head):
                return
            marker = version
        elif not self.translatable(version, code) or self._translation_current(
            version, code, head
        ):
            return
        else:
            marker = f"{version}{QUEUE_SEPARATOR}{code}"
        failed_ref = self._failed_at(marker)
        if background and failed_ref and failed_ref == self._wanted_ref(version, head):
            return
        lane = self.queue / BACKGROUND_LANE if background else self.queue
        lane.mkdir(parents=True, exist_ok=True)
        if background and (self.queue / marker).exists():
            return
        queued = lane / marker
        if queued.exists():
            return
        queued.touch()

    def failed(self, version, code=None):
        """Whether this target's last build failed at the ref it would build now.

        Args:
            version: version directory name.
            code: language code of a translation, or None for the site itself.
        """
        head, _ = self.refs()
        marker = version if code is None else f"{version}{QUEUE_SEPARATOR}{code}"
        failed_ref = self._failed_at(marker)
        return bool(failed_ref) and failed_ref == self._wanted_ref(version, head)

    def is_queued(self, marker):
        """Whether ``marker`` waits in either lane.

        Args:
            marker: queue file name, ``version`` or ``version:code``.
        """
        return (self.queue / marker).exists() or (
            self.queue / BACKGROUND_LANE / marker
        ).exists()

    def _dequeue(self, marker):
        """Drop ``marker`` from both lanes.

        Args:
            marker: queue file name, ``version`` or ``version:code``.
        """
        (self.queue / marker).unlink(missing_ok=True)
        (self.queue / BACKGROUND_LANE / marker).unlink(missing_ok=True)

    def next_queued(self, skip=()):
        """Return the oldest marker a visitor is waiting on, else the oldest background one.

        Args:
            skip: markers to pass over, normally the ones already building.
        """
        for lane in (self.queue, self.queue / BACKGROUND_LANE):
            if not lane.is_dir():
                continue
            waiting = sorted(
                (
                    marker
                    for marker in lane.iterdir()
                    if marker.is_file() and marker.name not in skip
                ),
                key=lambda marker: marker.stat().st_mtime,
            )
            if waiting:
                return waiting[0].name
        return None

    def _buildable(self, marker, head):
        """Whether ``marker`` can be built now rather than bounced.

        A translated site needs its version current. Deciding that here and
        not in ``build_language`` keeps a free slot from claiming the marker,
        bouncing it back into the lane and claiming it again for as long as
        the version takes to build.

        Args:
            marker: queue file name, ``version`` or ``version:code``.
            head: the mirror's current commit.
        """
        version, separator, _code = marker.partition(QUEUE_SEPARATOR)
        return not separator or self._current(version, head)

    def next_free(self):
        """Claim the oldest marker a slot can build now, or None.

        The build functions dequeue their marker only when they finish, so
        without a claim the same marker would be handed to every free slot.
        """
        head, _ = self.refs()
        with self._lock:
            skip = set(self._in_flight)
            while True:
                marker = self.next_queued(skip=frozenset(skip))
                if marker is None:
                    return None
                if self._buildable(marker, head):
                    self._in_flight.add(marker)
                    return marker
                skip.add(marker)

    def _slot_free(self, slots):
        """Whether fewer than ``slots`` markers are claimed now.

        A claim taken while every slot is busy waits in the pool's own FIFO,
        so a background build claimed early would run ahead of a visitor's
        marker that became buildable after it.

        Args:
            slots: builds the pool runs at once.
        """
        with self._lock:
            return len(self._in_flight) < slots

    def _claim(self, marker):
        """Claim ``marker`` for a slot when it waits and no slot holds it yet.

        Args:
            marker: queue file name, ``version`` or ``version:code``.

        Returns:
            ``True`` when the caller owns the build of ``marker`` now.
        """
        with self._lock:
            if marker in self._in_flight or not self.is_queued(marker):
                return False
            self._in_flight.add(marker)
            return True

    def acquire_builder(self):
        """Try to become the builder of all replicas without blocking.

        Returns:
            ``True`` while this process holds the builder lock.
        """
        if self._lock_handle is not None:
            return True
        self.lock_file.parent.mkdir(parents=True, exist_ok=True)
        handle = self.lock_file.open("a+")
        try:
            fcntl.lockf(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            handle.close()
            return False
        self._lock_handle = handle
        return True

    def _tags_on_disk(self):
        """Return every release tag that has a site, translation or prepared tree."""
        names = set()
        for parent in (self.sites, self.translations):
            if parent.is_dir():
                names.update(entry.name for entry in parent.iterdir())
        if self.prepared.is_dir():
            names.update(
                entry.name.rpartition(".")[0] for entry in self.prepared.iterdir()
            )
        names -= {LATEST, DEPLOYED}
        return {name for name in names if name and not name.startswith(".")}

    def _opened_at(self, tag):
        """Return when a visitor last opened ``tag``, 0 when nobody has.

        Args:
            tag: release tag.
        """
        mark = self.opened / tag
        return mark.stat().st_mtime if mark.is_file() else 0.0

    def _prune_tags(self):
        """Drop every release tag on disk beyond the ``KEPT_TAGS`` opened last.

        The trees move out under the lock ``next_free`` claims with, so no slot
        starts cloning a prepared tree that is on its way out, and are deleted
        after it. A tag stays while one of its markers builds or its own marker
        is queued. A queued language alone keeps nothing: every finished tag
        queues its preloaded languages, so it would keep every tag a crawler
        opens, and no slot claims a language of a tag whose site is gone.
        """
        trash = self.scratch / f".pruned.{os.getpid()}.{threading.get_ident()}"
        with self._lock:
            ranked = sorted(self._tags_on_disk(), key=self._opened_at, reverse=True)
            building = {
                marker.partition(QUEUE_SEPARATOR)[0] for marker in self._in_flight
            }
            for tag in ranked[KEPT_TAGS:]:
                if tag not in building and not self.is_queued(tag):
                    self._discard_tag(tag, trash)
        shutil.rmtree(trash, ignore_errors=True)

    def _discard_tag(self, tag, trash):
        """Move the trees of ``tag`` into ``trash``, drop its states and languages.

        Args:
            tag: release tag to drop.
            trash: directory on the same volume, deleted by the caller.
        """
        trash.mkdir(parents=True, exist_ok=True)
        trees = (
            self.sites / tag,
            self.translations / tag,
            *self.prepared.glob(f"{tag}.*"),
        )
        for tree in trees:
            if tree.exists():
                tree.rename(trash / f"{tree.parent.name}.{tree.name}")
        (self.states / f"{tag}.json").unlink(missing_ok=True)
        for state in self.states.glob(f"{tag}{QUEUE_SEPARATOR}*.json"):
            state.unlink(missing_ok=True)
        for lane in (self.queue, self.queue / BACKGROUND_LANE):
            for language in lane.glob(f"{tag}{QUEUE_SEPARATOR}*"):
                language.unlink(missing_ok=True)

    def _build_marker(self, marker):
        """Build what ``marker`` names, prune the tags, then release its slot.

        Args:
            marker: queue file name, ``version`` or ``version:code``.
        """
        try:
            version, separator, code = marker.partition(QUEUE_SEPARATOR)
            if separator:
                self.build_language(version, code)
            else:
                self.build(version)
            self._prune_tags()
        except Exception as exc:
            print(f"build {marker} crashed: {exc!r}", file=sys.stderr, flush=True)
            self._dequeue(marker)
        finally:
            with self._lock:
                self._in_flight.discard(marker)

    def run_builder(self, interval, parallel=1):
        """Serve the queue until the process ends.

        Only the lock holder writes the scratch, so what a killed holder left
        there, a build or a pruned tag halfway deleted, would stay forever
        unless the next holder clears it before its first build.

        Args:
            interval: seconds between fetches of the mirror.
            parallel: builds to run at once. A translated site waits for its
                version to be current, so the first pair still serialises; the
                slots pay off on the languages queued behind a built version.
        """
        while not self.acquire_builder():
            time.sleep(POLL_SECONDS)
        shutil.rmtree(self.scratch, ignore_errors=True)
        next_fetch = 0.0
        slots = max(1, int(parallel))
        with ThreadPoolExecutor(max_workers=slots) as pool:
            if self.snapshot.is_dir():
                self.request(DEPLOYED)
                if self._claim(DEPLOYED):
                    pool.submit(self._build_marker, DEPLOYED)
            while True:
                if time.monotonic() >= next_fetch:
                    try:
                        self.fetch()
                    except (OSError, subprocess.SubprocessError) as exc:
                        print(
                            f"fetch of {self.repository} failed: {exc}", file=sys.stderr
                        )
                    next_fetch = time.monotonic() + interval
                marker = self.next_free() if self._slot_free(slots) else None
                if marker is None:
                    time.sleep(POLL_SECONDS)
                    continue
                pool.submit(self._build_marker, marker)