mixle.enumeration.quantization.parallel module

Parallel quantization and distributed enumeration for the count-semiring index.

Two capabilities, both built on the existing resource/chunking layer (mixle.utils.parallel.planner Resources and _split_range):

  • Parallel quantization (ConvolutionExecutor): the count-DP’s cost is dominated by big-integer convolutions of count histograms (sequence power chains, composite suffix convolutions). Those are pure-data and embarrassingly parallel over the output bucket range, so a worker pool computes disjoint output slices and the result is concatenated. Output out[k] = sum_i a[i]*b[k-i] is independent of the chunk boundaries, so the parallel result equals the serial one exactly. Attach an executor to a Quantizer and the DP routes its convolutions through it (see mixle.enumeration.quantization.Quantizer.convolve()).

  • Distributed unranking (distributed_unrank()): unranking a contiguous rank range is embarrassingly parallel. The structural unrankers are closures (not picklable), so each worker rebuilds the index from the (picklable) distribution and unranks only its assigned rank sub-range; the per-worker rebuild is duplicated but the unranking is parallelized. Works on a local process pool and on a Spark context.

Use parallel quantization when histograms are large (deep budgets / high oversample); for small problems the serial path is used automatically to avoid pickling overhead.

resolve_workers(num_workers=None)[source]

Resolve a worker count from an explicit request, Resources, or the CPU count.

An explicit num_workers must be an exact positive integer: non-positive or fractional values are rejected rather than silently clamped to 1 or truncated (MXR-080-0209). None defers to auto-detection from Resources or the CPU count, which are not caller-supplied values and so are still floored at 1 the way they always were (a sandboxed os.cpu_count() can return 0/None).

Parameters:

num_workers (int | None)

Return type:

int

class ConvolutionExecutor(num_workers=None, min_parallel_width=2048, start_method=None)[source]

Bases: object

Process-pool executor for big-integer histogram convolutions (parallel quantization).

A context manager holding a reusable pool. convolve(a, b, max_fine_bucket) returns a mixle.enumeration.quantization.CountHistogram equal to the serial convolution. Falls back to serial when the output is small or only one worker is available.

start_method selects the pool’s multiprocessing start method (MXR-080-0210): None (the default) resolves to the safe 'spawn' default rather than forking this process, which is not safe when other threads may be active (a NumPy/BLAS/logging/accelerator thread pool, in the notebook/ML hosts this class targets) – see the _DEFAULT_START_METHOD module comment for the full hazard. Pass 'fork' or 'forkserver' explicitly to opt into cheaper pool startup in an environment you have confirmed is fork-safe.

Parameters:

  • num_workers (int | None)

  • min_parallel_width (int)

  • start_method (str | None)

close()[source]

Close the worker pool, if one is active.

Return type:

None

convolve(a, b, max_fine_bucket=None)[source]

Convolve two histograms, optionally using worker processes.

Parameters:

max_fine_bucket (int | None)

distributed_unrank(dist, budget_bits, start=0, count=None, bin_width_bits=1.0, oversample=8, num_workers=None, backend='local', spark_context=None, start_method=None)[source]

Unrank the rank range [start, start+count) in parallel, returning items in rank order.

Each worker rebuilds the index from dist (picklable) and unranks its assigned sub-range. count=None unranks to the end of the index. backend must be exactly 'local' (process pool) or 'spark' (requires spark_context); any other spelling – including a near-miss typo like 'Spark', 'spark ', or 'pyspark' – is rejected rather than silently falling back to local execution, which could otherwise run cluster-sized work on the caller’s own machine with no indication the requested backend was ignored (MXR-080-0209). start and an explicit count must be exact non-negative integers, validated before they reach the range splitter. The result equals the serial enumeration order.

For backend='local', start_method selects the worker pool’s multiprocessing start method (MXR-080-0210): None (the default) resolves to the safe 'spawn' default rather than forking this process, which is not safe when other threads may be active – see the _DEFAULT_START_METHOD module comment in this file for the full hazard. Validated eagerly even for backend='spark', where it has no effect (no local pool is started), so a typo’d start_method is rejected rather than silently ignored.

Parameters:

  • budget_bits (float)

  • start (int)

  • count (int | None)

  • bin_width_bits (float)

  • oversample (int)

  • num_workers (int | None)

  • backend (str)

  • start_method (str | None)

Return type:

list[tuple[Any, float]]