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)

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

Parameters:

num_workers (int | None)

Return type:

int

class ConvolutionExecutor(num_workers=None, min_parallel_width=2048)

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.

Parameters:

• num_workers (int | None)

• min_parallel_width (int)

close()

Return type:

None

convolve(a, b, max_fine_bucket=None)

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)

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 is 'local' (process pool) or 'spark' (requires spark_context). The result equals the serial enumeration order.

Parameters:

• budget_bits (float)

• start (int)

• count (int | None)

• bin_width_bits (float)

• oversample (int)

• num_workers (int | None)

• backend (str)

Return type:

list[tuple[Any, float]]