Execution API Reference

Execution units extracted from MerlinProcessor (PML-305).

This module hosts the focused components that own remote chunk execution:

  • BatchChunker — splits Merlin-level batches into microbatch chunks, runs them with bounded concurrency, and stitches the outputs back together.

  • RemoteJobRunner — executes one remote chunk end to end: builds a fresh remote processor, prepares sampler iterations, submits a job, polls it, and maps the raw results into a tensor.

Both units receive their dependencies (fresh-processor factory, backend capabilities, result mapping, job tracking) as injected callables so they are independently testable with no-cloud fakes. MerlinProcessor remains the public entry point and coordinates these units without owning the execution details itself.

The threading semantics (daemon chunk threads, cooperative cancellation, deadline checks, polling backoff) are intentionally identical to the pre-extraction MerlinProcessor implementation.

class merlin.core.execution.BatchChunker(*, run_chunk, get_chunk_concurrency, cancel_all)

Bases: object

Split input batches into chunks and run them with bounded concurrency.

Parameters:

  • run_chunk (Callable) – Callable executing one chunk with the signature (layer, config, input_chunk, nsample, state, deadline, job_base_label) and returning a torch.Tensor. Injected so chunk orchestration is testable with fakes and so monkeypatched processor methods stay observable.

  • get_chunk_concurrency (Callable[[], int]) – Returns the maximum number of chunk jobs allowed in flight at once. Read once per scheduling pass (rather than captured at construction) so a mid-call change to the processor’s chunk_concurrency takes effect, matching the pre-extraction loop.

  • cancel_all (Callable[[], None]) – Invoked when the deadline elapses so in-flight remote jobs are cancelled best-effort before raising TimeoutError.

run_chunks(layer, config, input_tensor, chunks, nsample, state, deadline)

Submit chunk jobs with limited concurrency and stitch results.

Parameters:

  • layer (MerlinModule) – Quantum leaf whose backend execution produces each chunk output.

  • config (ValidatedLayerConfig) – Validated layer configuration (circuit, input state, param order).

  • input_tensor (torch.Tensor) – Full input batch to split across chunks.

  • chunks (list[tuple[int, int]]) – [start, end) row ranges produced by split_batch().

  • nsample (int | None) – Requested sample count, or None/<= 0 for exact probabilities.

  • state (CallState) – Per-call state used for chunk counters and cooperative cancellation.

  • deadline (float | None) – Absolute time.time() deadline, or None for no timeout.

Returns:

The per-chunk outputs concatenated along the batch dimension.

Return type:

torch.Tensor

Raises:

  • TimeoutError – If deadline elapses before all chunks finish; in-flight remote jobs are cancelled best-effort first.

  • BaseException – The first error raised by any chunk, re-raised once the remaining in-flight chunks have settled.

static split_batch(batch_size, microbatch_size)

Split batch_size rows into [start, end) microbatch chunks.

Parameters:

  • batch_size (int) – Total number of rows to split.

  • microbatch_size (int) – Maximum number of rows per chunk. Must be strictly positive.

Returns:

[start, end) half-open index ranges covering all batch_size rows in order.

Return type:

list[tuple[int, int]]

Raises:

ValueError – If microbatch_size is not strictly positive. A non-positive size would never advance the split and loop forever.

class merlin.core.execution.RemoteJobRunner(*, create_processor, get_available_commands, extract_input_params, effective_sample_count, get_max_shots_per_call, default_shots_per_call, map_results, register_job, unregister_job, get_microbatch_limit, max_retries=3, job_name_max=50, default_sampling_command=None)

Bases: object

Execute a single remote chunk: fresh processor, submit, poll, map.

Parameters:

  • create_processor (Callable[[], RemoteProcessor]) – Factory returning a fresh, independent RemoteProcessor per attempt.

  • get_available_commands (Callable[[], tuple[str, ...]]) – Returns the backend command snapshot driving probs-vs-sampling.

  • extract_input_params (Callable[[ValidatedLayerConfig], list[str]]) – Returns the ordered circuit-parameter names that receive model inputs. Injected (rather than reading config.input_param_order directly) so the remote and local paths share the single param-routing seam.

  • effective_sample_count (Callable[[int | None], int]) – Maps a requested nsample to the capped shot count to submit.

  • get_max_shots_per_call (Callable[[], int | None]) – Returns the current hard cap on shots per sampler call.

  • default_shots_per_call (int) – Fallback shots value used when the cap is unset.

  • map_results (Callable) – Maps a raw results dict to a tensor with the signature (raw_results, batch_size, layer, nsample, is_probability).

  • register_job (Callable[[RemoteJob], None]) – Records a submitted job for cancellation tracking and history.

  • unregister_job (Callable[[RemoteJob], None]) – Removes a job from active-cancellation tracking.

  • get_microbatch_limit (Callable[[], int | None]) – Returns the per-chunk size guard, or None when chunk sizes are not bounded (session backends).

  • max_retries (int) – Number of submission attempts per chunk.

  • job_name_max (int) – Maximum length of remote job names.

poll_job(job, state, deadline, batch_size, layer, nsample, is_probability=False)

Poll a submitted job until complete/failed/timeout and return results.

Continuously polls the job status, updating call state and handling timeouts, cancellation, and failures. Upon completion, maps results to a tensor through the injected map_results dependency.

Parameters:

  • job (perceval.runtime.RemoteJob) – Submitted job to poll.

  • state (CallState) – Per-call state updated with status and job ids, and observed for cooperative cancellation.

  • deadline (float | None) – Absolute time.time() deadline, or None for no timeout.

  • batch_size (int) – Number of input rows in this chunk, forwarded to map_results.

  • layer (MerlinModule) – Quantum leaf forwarded to map_results for output extraction.

  • nsample (int | None) – Original sample-count request, forwarded to map_results.

  • is_probability (bool) – Whether the job runs in exact-probability mode. Default value is False.

Returns:

The mapped [batch_size, ...] output tensor.

Return type:

torch.Tensor

Raises:
run_chunk(layer, config, input_chunk, nsample, state, deadline, job_base_label=None)

Submit a single chunk job with retries and return the mapped tensor.

Builds a fresh remote processor and sampler on each attempt (so a corrupted processor cannot poison retries), submits the job, polls it to completion, and maps the raw results into a tensor. Cancellation and deadline are checked cooperatively before every attempt.

Parameters:

  • layer (MerlinModule) – Quantum leaf whose backend execution produces the chunk output.

  • config (ValidatedLayerConfig) – Validated layer configuration (circuit, input state, param order).

  • input_chunk (torch.Tensor) – Rows of the batch assigned to this chunk.

  • nsample (int | None) – Requested sample count, or None/<= 0 for exact probabilities.

  • state (CallState) – Per-call state observed for cooperative cancellation and job ids.

  • deadline (float | None) – Absolute time.time() deadline, or None for no timeout.

  • job_base_label (str | None) – Base label for the remote job name, or None to leave it unset.

Returns:

The mapped [chunk_size, ...] output tensor for this chunk.

Return type:

torch.Tensor

Raises:
submit_job(sampler, nsample, job_base_label)

Submit a job to the sampler, selecting command based on backend capabilities.

Command Selection Strategy

  1. Exact Probabilities ("probs" command): - Used if backend exposes "probs" AND (nsample is None or nsample <= 0). - Returns normalized probability distribution; nsample is ignored.

  2. Sampling ("sample_count" or "samples" commands): - Used if exact probabilities are not available or nsample > 0. - Uses "sample_count" first, otherwise "samples". - Number of samples = effective_sample_count(nsample).

Job names are sanitized and capped through _capped_name.

Parameters:

  • sampler (perceval.algorithm.Sampler) – Perceval Sampler instance configured with circuit and iterations.

  • nsample (int | None) – Number of samples requested. If None or <= 0, triggers exact probability computation (if available).

  • job_base_label (str | None) – Base label for the remote job name, or None to leave it unset.

Returns:

The submitted job handle and the is_probability execution flag.

Return type:

tuple[RemoteJob, bool]

merlin.core.execution.build_iteration_parameters(input_chunk, input_parameter_names)

Map input rows to circuit parameters after validating their shape.

Parameters:

  • input_chunk (torch.Tensor) – Two-dimensional input tensor with one column per circuit parameter.

  • input_parameter_names (Sequence[str]) – Circuit parameter names in the expected column order.

Returns:

Parameter mappings for each input row.

Return type:

list[dict[str, float]]

Raises:

ValueError – If the input tensor column count does not match the parameter count.

merlin.core.execution.select_sampling_command(available_commands, *, default_command=None)

Select the supported sampling command for a backend.

Parameters:

  • available_commands (Sequence[str]) – Commands advertised by the backend.

  • default_command (str | None) – Explicit command to use when capability metadata is unavailable. Defaults to None.

Returns:

"sample_count" when available, otherwise "samples".

Return type:

str

Raises:

RuntimeError – If the backend advertises neither sampling command and no default was provided.