mstar.worker.batch_builder

mstar.worker.batch_builder#

Batch composition: which ready rows of one (node, walk) run this step.

The MicroScheduler picks the (node, walk) and owns the mechanics (popping ready queues, the backlog, holds, TP follow); a BaseBatchBuilder decides which of the popped rows go into the step, which stay parked in the backlog and which go back to their ready queues.

Functions

make_batch_builder(kind)

off_group_rids(batch, capture_group[, ...])

The rids of batch outside anchor's capture group, or outside the first eligible rid's when no anchor is given.

Classes

BaseBatchBuilder()

BatchBuildRequest(node_name, graph_walk, ...)

BatchBuildResult(scheduled, backlog, returned)

BatchBuilderType(*values)

FIFOBatchBuilder()

Backlog first, then fresh rows, in arrival order, up to the cap; one capture group per step.
class mstar.worker.batch_builder.BaseBatchBuilder[source]#

Bases: ABC

abstractmethod build_batch(request)[source]#

One step’s rows out of request’s backlog and entries.

Parameters:

request (BatchBuildRequest)

Return type:

BatchBuildResult

chain_must_yield(backlog, live_rids, anchor, capture_group)[source]#

Whether a speculation chain anchored at anchor has to stop so backlog’s live rows can run: the chain’s merge (a build with capture_group_of=anchor) would never take them.

Parameters:
Return type:

bool

room_for_continuing(cap, backlog)[source]#

How many of a speculation chain’s own rids fit in a step capped at cap when backlog waits under the same (node, walk).

The chain only ever continues its own rids, so at the cap a backlogged chunk would never be reached. Giving the backlog first claim costs the displaced rids one step — their nodes go ready again when the in-flight batch lands. None for an uncapped node: nothing is displaced.

Parameters:

  • cap (int | None)

  • backlog (ScheduledBatch | None)

Return type:

int | None

class mstar.worker.batch_builder.BatchBuildRequest(node_name: str, graph_walk: str, backlog: 'ScheduledBatch | None' = None, fresh: 'ScheduledBatch | None' = None, blocked_rids: set[int] = <factory>, capture_group_of: int | None = None, walk_caps: dict[str, int | None] = <factory>, walk_row_caps: dict[str, int] = <factory>, pre_existing_batch_size: int = 0, pre_existing_walk: str | None = None, capture_group: collections.abc.Callable[[int], typing.Any] = <function BatchBuildRequest.<lambda> at 0x7f230d952f20>)[source]#

Bases: object

Parameters:

  • node_name (str)

  • graph_walk (str)

  • backlog (ScheduledBatch | None)

  • fresh (ScheduledBatch | None)

  • blocked_rids (set[int])

  • capture_group_of (int | None)

  • walk_caps (dict[str, int | None])

  • walk_row_caps (dict[str, int])

  • pre_existing_batch_size (int)

  • pre_existing_walk (str | None)

  • capture_group (Callable[[int], Any])

backlog: ScheduledBatch | None = None#
blocked_rids: set[int]#
capture_group()#
capture_group_of: int | None = None#
fresh: ScheduledBatch | None = None#
graph_walk: str#
node_name: str#
pre_existing_batch_size: int = 0#
pre_existing_walk: str | None = None#
walk_caps: dict[str, int | None]#
walk_row_caps: dict[str, int]#
class mstar.worker.batch_builder.BatchBuildResult(scheduled, backlog, returned)[source]#

Bases: NamedTuple

Parameters:

  • scheduled (ScheduledBatch | None)

  • backlog (ScheduledBatch | None)

  • returned (dict[int, int] | None)

backlog: ScheduledBatch | None#

Alias for field number 1

returned: dict[int, int] | None#

Alias for field number 2

scheduled: ScheduledBatch | None#

Alias for field number 0

class mstar.worker.batch_builder.BatchBuilderType(*values)[source]#

Bases: Enum

FIFO = 'fifo'#
class mstar.worker.batch_builder.FIFOBatchBuilder[source]#

Bases: BaseBatchBuilder

Backlog first, then fresh rows, in arrival order, up to the cap; one capture group per step.

Under a combined walk a step of one walk takes that walk’s cap and a mixed step the combined walk’s own, or with none the smallest of the caps of the walks present in it; a walk with no row in the step does not cap it. Rows join in order; one whose walk would cap the step at or below the rows already in it is left out, and is first in line next step.

Where this differs from the scheduler it replaces, and why. main had two paths: with a backlog it drained that alone, never touching the ready queues, so the backlog strictly shrank; with none it popped a fresh set, capped it and parked the remainder. This builder always does both, which is better – a backlog one row short of the cap gets topped up instead of running short – but it means a step that drains a backlog also pops fresh rows, and parking their overflow would refill the backlog from the queues on every pass. It would then never drain, and since a key holding a backlog is served before any other node, that node starves the rest: on wan22.yaml with dit capped at 1 a fifth request’s text_encoder waited 48.7 s for its first step against 0.2 s on main.

So a fresh row the cap pushed out goes back to its ready queue only on a step that served a backlog, which is the case main never created. With no backlog it parks, exactly as main did, and the remainder keeps its arrival order instead of being re-popped and reordered. A row held back for any other reason – OOM backoff, pending remove, a foreign capture group, a walk the combined cap left out – always parks: it cannot run this step whoever holds it.

build_batch(request)[source]#

One step’s rows out of request’s backlog and entries.

Parameters:

request (BatchBuildRequest)

Return type:

BatchBuildResult

mstar.worker.batch_builder.make_batch_builder(kind)[source]#
Parameters:

kind (BatchBuilderType)

Return type:

BaseBatchBuilder

mstar.worker.batch_builder.off_group_rids(batch, capture_group, exclude_rids=frozenset(), anchor=None)[source]#

The rids of batch outside anchor’s capture group, or outside the first eligible rid’s when no anchor is given.

Parameters:
Return type:

set[int]