Class tigrbl.runtime.kernel.Kernel

tigrbl.runtime.kernel.Kernel

Kernel(atoms=None)

SSoT for runtime scheduling. One Kernel per App (not per API). Auto-primed under the hood. Downstream users never touch this.

Source code in tigrbl/runtime/kernel.py

def __init__(self, atoms: Optional[Sequence[_DiscoveredAtom]] = None):
    self._atoms_cache: Optional[list[_DiscoveredAtom]] = (
        list(atoms) if atoms else None
    )
    self._specs_cache = _SpecsOnceCache()
    self._opviews: _WeakMaybeDict[Any, Dict[Tuple[type, str], OpView]] = (
        _WeakMaybeDict()
    )
    self._kernelz_payload: _WeakMaybeDict[Any, Dict[str, Dict[str, List[str]]]] = (
        _WeakMaybeDict()
    )
    self._primed: _WeakMaybeDict[Any, bool] = _WeakMaybeDict()
    self._lock = threading.Lock()

get_specs

get_specs(model)

Source code in tigrbl/runtime/kernel.py

def get_specs(self, model: type) -> Mapping[str, Any]:
    return self._specs_cache.get(model)

prime_specs

prime_specs(models)

Source code in tigrbl/runtime/kernel.py

def prime_specs(self, models: Sequence[type]) -> None:
    self._specs_cache.prime(models)

invalidate_specs

invalidate_specs(model=None)

Source code in tigrbl/runtime/kernel.py

def invalidate_specs(self, model: Optional[type] = None) -> None:
    self._specs_cache.invalidate(model)

build

build(model, alias)

Source code in tigrbl/runtime/kernel.py

def build(self, model: type, alias: str) -> Dict[str, List[StepFn]]:
    chains = _hook_phase_chains(model, alias)
    specs = getattr(getattr(model, "ops", SimpleNamespace()), "by_alias", {})
    sp_list = specs.get(alias) or ()
    sp = sp_list[0] if sp_list else None
    target = (getattr(sp, "target", alias) or "").lower()
    persist_policy = getattr(sp, "persist", "default")
    persistent = (
        persist_policy != "skip" and target not in {"read", "list"}
    ) or _is_persistent(chains)
    try:
        _inject_atoms(chains, self._atoms() or (), persistent=persistent)
    except Exception:
        logger.exception(
            "kernel: atom injection failed for %s.%s",
            getattr(model, "__name__", model),
            alias,
        )
    if persistent:
        try:
            start_anchor, start_run = _sys.get("txn", "begin")
            end_anchor, end_run = _sys.get("txn", "commit")
            chains.setdefault(start_anchor, []).append(
                _wrap_atom(start_run, anchor=start_anchor)
            )
            chains.setdefault(end_anchor, []).append(
                _wrap_atom(end_run, anchor=end_anchor)
            )
        except Exception:
            logger.exception(
                "kernel: failed to inject txn system steps for %s.%s",
                getattr(model, "__name__", model),
                alias,
            )
    for ph in PHASES:
        chains.setdefault(ph, [])
    return chains

plan_labels

plan_labels(model, alias)

Source code in tigrbl/runtime/kernel.py

def plan_labels(self, model: type, alias: str) -> list[str]:
    labels: list[str] = []
    chains = self.build(model, alias)
    ordered_anchors = _ev.all_events_ordered()
    phase_for = {a: _ev.get_anchor_info(a).phase for a in ordered_anchors}
    for anchor in ordered_anchors:
        phase = phase_for[anchor]
        for step in chains.get(phase, []) or []:
            lbl = getattr(step, "__tigrbl_label", None)
            if isinstance(lbl, str) and lbl.endswith(f"@{anchor}"):
                labels.append(lbl)
    return labels

invoke async

invoke(*, model, alias, db, request=None, ctx=None)

Execute an operation for model.alias using the executor.

Source code in tigrbl/runtime/kernel.py

async def invoke(
    self,
    *,
    model: type,
    alias: str,
    db: Any,
    request: Any | None = None,
    ctx: Optional[Mapping[str, Any]] = None,
) -> Any:
    """Execute an operation for ``model.alias`` using the executor."""
    phases = self.build(model, alias)
    base_ctx = _Ctx.ensure(request=request, db=db, seed=ctx)
    base_ctx.model = model
    base_ctx.op = alias
    specs = self.get_specs(model)
    base_ctx.opview = self._compile_opview_from_specs(
        specs, SimpleNamespace(alias=alias)
    )
    return await _invoke(request=request, db=db, phases=phases, ctx=base_ctx)

ensure_primed

ensure_primed(app)

Autoprime once per App: specs → OpViews → /kernelz payload.

Source code in tigrbl/runtime/kernel.py

def ensure_primed(self, app: Any) -> None:
    """Autoprime once per App: specs → OpViews → /kernelz payload."""
    with self._lock:
        if self._primed.get(app):
            return
        from ..system.diagnostics.utils import (
            model_iter as _model_iter,
            opspecs as _opspecs,
        )

        models = list(_model_iter(app))

        # 1) per-model specs once
        for m in models:
            self._specs_cache.get(m)

        # 2) compile OpViews per (model, alias)
        ov_map: Dict[Tuple[type, str], OpView] = {}
        for m in models:
            specs = self._specs_cache.get(m)
            for sp in _opspecs(m):
                ov_map[(m, sp.alias)] = self._compile_opview_from_specs(specs, sp)
        self._opviews[app] = ov_map

        # 3) build /kernelz payload once (dedup wire hooks)
        payload = self._build_kernelz_payload_internal(app)
        self._kernelz_payload[app] = payload
        self._primed[app] = True

get_opview

get_opview(app, model, alias)

Return OpView for (model, alias); compile on-demand if missing.

Source code in tigrbl/runtime/kernel.py

def get_opview(self, app: Any, model: type, alias: str) -> OpView:
    """Return OpView for (model, alias); compile on-demand if missing."""
    self.ensure_primed(app)

    ov_map: Dict[Tuple[type, str], OpView] = self._opviews.setdefault(app, {})
    ov = ov_map.get((model, alias))
    if ov is not None:
        return ov

    try:
        specs = self._specs_cache.get(model)
        from types import SimpleNamespace
        from ..system.diagnostics.utils import opspecs as _opspecs

        found = False
        for sp in _opspecs(model):
            ov_map.setdefault(
                (model, sp.alias), self._compile_opview_from_specs(specs, sp)
            )
            if sp.alias == alias:
                found = True

        if not found:
            temp_sp = SimpleNamespace(alias=alias)
            ov_map[(model, alias)] = self._compile_opview_from_specs(specs, temp_sp)

        return ov_map[(model, alias)]
    except Exception:
        raise RuntimeError(
            f"opview_missing: app={app!r} model={getattr(model, '__name__', model)!r} alias={alias!r}"
        )

kernelz_payload

kernelz_payload(app)

Thin accessor for endpoint: guarantees primed, returns cached payload.

Source code in tigrbl/runtime/kernel.py

def kernelz_payload(self, app: Any) -> Dict[str, Dict[str, List[str]]]:
    """Thin accessor for endpoint: guarantees primed, returns cached payload."""
    self.ensure_primed(app)
    return self._kernelz_payload[app]

invalidate_kernelz_payload

invalidate_kernelz_payload(app=None)

Source code in tigrbl/runtime/kernel.py

def invalidate_kernelz_payload(self, app: Optional[Any] = None) -> None:
    with self._lock:
        if app is None:
            self._kernelz_payload = _WeakMaybeDict()
            self._opviews = _WeakMaybeDict()
            self._primed = _WeakMaybeDict()
        else:
            self._kernelz_payload.pop(app, None)
            self._opviews.pop(app, None)
            self._primed.pop(app, None)