Backends

Where the counting happens. See Running in parallel for the guide.

Selecting one

thresher.backends.get_backend

get_backend(backend: Any) -> Backend

Resolve the backend option to something that can do the counting.

Parameters:

Name	Type	Description	Default
backend	Any	the name of a backend, or an object already implementing the protocol - which is how a caller passes a pre-configured RayBackend(num_shards=...).	required

Returns:

Type	Description
Backend	The backend to use.

Raises:

Type	Description
UnknownBackendError	if the name is not recognised. It is a ValueError.
BackendDependencyError	if 'ray' was asked for and Ray is not installed. It is an ImportError, and the message says how to install it.

Note

The names build default instances. To configure one - MultiprocessingBackend( num_workers=4), RayBackend(num_shards=...) - construct it and pass the object as the backend option instead of the name.

Source code in src/thresher/backends/__init__.py

def get_backend(backend: Any) -> Backend:
    """Resolve the `backend` option to something that can do the counting.

    Args:
        backend: the name of a backend, or an object already implementing the protocol -
            which is how a caller passes a pre-configured `RayBackend(num_shards=...)`.

    Returns:
        The backend to use.

    Raises:
        UnknownBackendError: if the name is not recognised. It is a `ValueError`.
        BackendDependencyError: if `'ray'` was asked for and Ray is not installed. It is
            an `ImportError`, and the message says how to install it.

    Note:
        The names build default instances. To configure one - `MultiprocessingBackend(
        num_workers=4)`, `RayBackend(num_shards=...)` - construct it and pass the object
        as the `backend` option instead of the name.
    """
    if not isinstance(backend, str):
        # Already a backend instance; trust the protocol.
        return backend  # type: ignore[no-any-return]

    name = backend.lower()
    if name == "local":
        return LocalBackend()
    if name == "mp":
        return MultiprocessingBackend()
    if name == "ray":
        from thresher.backends.ray_backend import RayBackend

        return RayBackend()

    raise UnknownBackendError(backend, AVAILABLE_BACKENDS)

The contract

thresher.backends.base

The execution-backend contract, and the pure map/reduce steps behind it.

A backend decides where the counting happens, never what the answer is. Every backend must return bit-identical results for the same input; only the distribution of the work changes. That is why the map and reduce steps live here as plain functions rather than inside any one backend - they are shared verbatim, and can be tested without a cluster.

Two primitives cover every algorithm that can be parallelised:

tally_candidates Score a fixed list of candidate thresholds against the data. Linear search and grid search are both "score these candidates, keep the best", so both reduce to this.

class_counts_by_score Count the classes at each distinct score. The exact sweep needs only these counts, not the samples themselves, so the per-record work distributes and the driver is left with one pass over the distinct scores.

Backend

Bases: Protocol

Where the counting happens.

Implementations must not change the answer - see the module docstring.

tally_candidates

tally_candidates(
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]

Count correct predictions for each candidate threshold, over all the data.

Source code in src/thresher/backends/base.py

def tally_candidates(
    self,
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]:
    """Count correct predictions for each candidate threshold, over all the data."""
    ...

class_counts_by_score

class_counts_by_score(
    scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]

Count negatives and positives at each distinct score, over all the data.

Source code in src/thresher/backends/base.py

def class_counts_by_score(
    self, scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]:
    """Count negatives and positives at each distinct score, over all the data."""
    ...

tally_chunk

tally_chunk(
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]

Count correct predictions per candidate, for one shard of the data.

This is the map step of tally_candidates, and it is deliberately a free function: the local backend calls it directly and the Ray backend ships it to workers, so both run exactly the same code.

Parameters:

Name	Type	Description	Default
candidates	Sequence[float]	the thresholds to score.	required
scores	Sequence[float]	this shard's scores.	required
actual_classes	Sequence[int]	this shard's classes, as -1 and 1.	required

Returns:

Type	Description
list[int]	One count per candidate, in the same order: how many of this shard's samples
list[int]	that candidate classifies correctly.

Source code in src/thresher/backends/base.py

def tally_chunk(
    candidates: Sequence[float], scores: Sequence[float], actual_classes: Sequence[int]
) -> list[int]:
    """Count correct predictions per candidate, for one shard of the data.

    This is the map step of `tally_candidates`, and it is deliberately a free function:
    the local backend calls it directly and the Ray backend ships it to workers, so both
    run exactly the same code.

    Args:
        candidates: the thresholds to score.
        scores: this shard's scores.
        actual_classes: this shard's classes, as -1 and 1.

    Returns:
        One count per candidate, in the same order: how many of *this shard's* samples
        that candidate classifies correctly.
    """
    tallies = [0] * len(candidates)
    for index, candidate in enumerate(candidates):
        correct = 0
        for score, actual in zip(scores, actual_classes, strict=False):
            if (1 if score > candidate else -1) == actual:
                correct += 1
        tallies[index] = correct
    return tallies

merge_tallies

merge_tallies(
    partials: Iterable[Sequence[int]],
) -> list[int]

Add per-shard tallies together elementwise.

This is the reduce step of tally_candidates. Addition is associative and commutative, so the order shards arrive in cannot affect the result - which is what lets the answer be identical across backends.

Parameters:

Name	Type	Description	Default
partials	Iterable[Sequence[int]]	one tally list per shard, all the same length.	required

Returns:

Type	Description
list[int]	The summed tallies.

Raises:

Type	Description
ShardMergeError	if no partials were given, or they disagree on length. It is a ValueError.

Source code in src/thresher/backends/base.py

def merge_tallies(partials: Iterable[Sequence[int]]) -> list[int]:
    """Add per-shard tallies together elementwise.

    This is the reduce step of `tally_candidates`. Addition is associative and
    commutative, so the order shards arrive in cannot affect the result - which is what
    lets the answer be identical across backends.

    Args:
        partials: one tally list per shard, all the same length.

    Returns:
        The summed tallies.

    Raises:
        ShardMergeError: if no partials were given, or they disagree on length. It is a
            `ValueError`.
    """
    merged: list[int] | None = None
    for partial in partials:
        if merged is None:
            merged = list(partial)
            continue
        if len(partial) != len(merged):
            raise ShardMergeError(f"shard tallies disagree on length: {len(partial)} vs {len(merged)}")
        for index, value in enumerate(partial):
            merged[index] += value

    if merged is None:
        raise ShardMergeError("no shard tallies to merge")
    return merged

count_chunk

count_chunk(
    scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]

Count negatives and positives at each distinct score, for one shard.

The map step of class_counts_by_score.

Parameters:

Name	Type	Description	Default
scores	Sequence[float]	this shard's scores.	required
actual_classes	Sequence[int]	this shard's classes, as -1 and 1.	required

Returns:

Type	Description
dict[float, ClassCounts]	A mapping of score to (negatives, positives) seen at it in this shard.

Source code in src/thresher/backends/base.py

def count_chunk(scores: Sequence[float], actual_classes: Sequence[int]) -> dict[float, ClassCounts]:
    """Count negatives and positives at each distinct score, for one shard.

    The map step of `class_counts_by_score`.

    Args:
        scores: this shard's scores.
        actual_classes: this shard's classes, as -1 and 1.

    Returns:
        A mapping of score to `(negatives, positives)` seen at it in this shard.
    """
    counts: dict[float, ClassCounts] = {}
    for score, actual in zip(scores, actual_classes, strict=False):
        negatives, positives = counts.get(score, (0, 0))
        if actual == 1:
            counts[score] = (negatives, positives + 1)
        else:
            counts[score] = (negatives + 1, positives)
    return counts

merge_counts

merge_counts(
    partials: Iterable[Mapping[float, ClassCounts]],
) -> dict[float, ClassCounts]

Merge per-shard score counts by summing them.

The reduce step of class_counts_by_score, and order-independent for the same reason merge_tallies is.

Parameters:

Name	Type	Description	Default
partials	Iterable[Mapping[float, ClassCounts]]	one score-to-counts mapping per shard.	required

Returns:

Type	Description
dict[float, ClassCounts]	The combined mapping.

Source code in src/thresher/backends/base.py

def merge_counts(partials: Iterable[Mapping[float, ClassCounts]]) -> dict[float, ClassCounts]:
    """Merge per-shard score counts by summing them.

    The reduce step of `class_counts_by_score`, and order-independent for the same reason
    `merge_tallies` is.

    Args:
        partials: one score-to-counts mapping per shard.

    Returns:
        The combined mapping.
    """
    merged: dict[float, ClassCounts] = {}
    for partial in partials:
        for score, (negatives, positives) in partial.items():
            running_negatives, running_positives = merged.get(score, (0, 0))
            merged[score] = (running_negatives + negatives, running_positives + positives)
    return merged

plan_shards

plan_shards(
    total: int, workers: int, min_rows: int
) -> list[tuple[int, int]]

Work out the shard boundaries for a dataset.

Kept separate from any backend so the arithmetic can be tested on its own, including on machines where Ray cannot be installed.

Parameters:

Name	Type	Description	Default
total	int	number of samples.	required
workers	int	how many shards are wanted at most, normally the cluster's CPU count.	required
min_rows	int	smallest worthwhile shard. Below this the coordination costs more than the work saved, so fewer, larger shards are produced instead.	required

Returns:

Type	Description
list[tuple[int, int]]	A list of (start, stop) index pairs covering range(total) exactly once, in
list[tuple[int, int]]	order and without gaps. Empty when total is 0.

Source code in src/thresher/backends/base.py

def plan_shards(total: int, workers: int, min_rows: int) -> list[tuple[int, int]]:
    """Work out the shard boundaries for a dataset.

    Kept separate from any backend so the arithmetic can be tested on its own, including
    on machines where Ray cannot be installed.

    Args:
        total: number of samples.
        workers: how many shards are wanted at most, normally the cluster's CPU count.
        min_rows: smallest worthwhile shard. Below this the coordination costs more than
            the work saved, so fewer, larger shards are produced instead.

    Returns:
        A list of `(start, stop)` index pairs covering `range(total)` exactly once, in
        order and without gaps. Empty when `total` is 0.
    """
    if total <= 0:
        return []

    usable = max(1, min(workers, total // max(1, min_rows)))
    size, remainder = divmod(total, usable)

    boundaries: list[tuple[int, int]] = []
    start = 0
    for index in range(usable):
        # Spread the remainder over the first shards, so sizes differ by at most one.
        stop = start + size + (1 if index < remainder else 0)
        if start < stop:
            boundaries.append((start, stop))
        start = stop
    return boundaries

Local

thresher.backends.local.LocalBackend

Do the work here, in this process.

This is what every version before 0.4.2 did, and what still happens unless a different backend is asked for.

tally_candidates

tally_candidates(
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]

Count correct predictions for each candidate threshold.

Parameters:

Name	Type	Description	Default
candidates	Sequence[float]	the thresholds to score.	required
scores	Sequence[float]	the values being split.	required
actual_classes	Sequence[int]	the matching classes, as -1 and 1.	required

Returns:

Type	Description
list[int]	One count per candidate, in the same order.

Source code in src/thresher/backends/local.py

def tally_candidates(
    self,
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]:
    """Count correct predictions for each candidate threshold.

    Args:
        candidates: the thresholds to score.
        scores: the values being split.
        actual_classes: the matching classes, as -1 and 1.

    Returns:
        One count per candidate, in the same order.
    """
    return tally_chunk(candidates, scores, actual_classes)

class_counts_by_score

class_counts_by_score(
    scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]

Count negatives and positives at each distinct score.

Parameters:

Name	Type	Description	Default
scores	Sequence[float]	the values being split.	required
actual_classes	Sequence[int]	the matching classes, as -1 and 1.	required

Returns:

Type	Description
dict[float, ClassCounts]	A mapping of score to (negatives, positives).

Source code in src/thresher/backends/local.py

def class_counts_by_score(
    self, scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]:
    """Count negatives and positives at each distinct score.

    Args:
        scores: the values being split.
        actual_classes: the matching classes, as -1 and 1.

    Returns:
        A mapping of score to `(negatives, positives)`.
    """
    return count_chunk(scores, actual_classes)

Multiprocessing

thresher.backends.mp_backend.MultiprocessingBackend

MultiprocessingBackend(
    num_workers: int | None = None,
    min_rows_per_shard: int = DEFAULT_MIN_ROWS_PER_SHARD,
)

Count in parallel across this machine's CPU cores.

Example

from thresher import Thresher Thresher(backend="mp").optimize_threshold(scores, actual_classes) # doctest: +SKIP

Note

Because the workers are separate processes, any script that builds one of these at module level must sit behind an if __name__ == "__main__": guard. Without it the workers re-import the script and build their own pools; see the module docstring. That mistake is reported rather than left to hang.

Configure how the data is divided, and over how many processes.

Parameters:

Name	Type	Description	Default
num_workers	int | None	how many worker processes to use. Defaults to one per processor; -1 means every processor bar one.	None
min_rows_per_shard	int	do not produce shards smaller than this. Below it the work is done in this process instead, since a fork would cost more than it saves.	DEFAULT_MIN_ROWS_PER_SHARD

Raises:

Type	Description
ConfigurationError	if num_workers is 0 or below -1. Checked here rather than on first use, so the failure lands when the backend is asked for. It is a ValueError.

Source code in src/thresher/backends/mp_backend.py

def __init__(
    self, num_workers: int | None = None, min_rows_per_shard: int = DEFAULT_MIN_ROWS_PER_SHARD
) -> None:
    """Configure how the data is divided, and over how many processes.

    Args:
        num_workers: how many worker processes to use. Defaults to one per processor;
            `-1` means every processor bar one.
        min_rows_per_shard: do not produce shards smaller than this. Below it the work
            is done in this process instead, since a fork would cost more than it saves.

    Raises:
        ConfigurationError: if `num_workers` is 0 or below -1. Checked here rather than
            on first use, so the failure lands when the backend is asked for. It is a
            `ValueError`.
    """
    self._num_workers = resolve_worker_count(num_workers)
    self._min_rows_per_shard = min_rows_per_shard

tally_candidates

tally_candidates(
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]

Count correct predictions per candidate, sharded across processes.

Parameters:

Name	Type	Description	Default
candidates	Sequence[float]	the thresholds to score.	required
scores	Sequence[float]	the values being split.	required
actual_classes	Sequence[int]	the matching classes, as -1 and 1.	required

Returns:

Type	Description
list[int]	One count per candidate, identical to what the local backend returns.

Raises:

Type	Description
ParallelBootstrapError	if the worker processes could not start, which on a re-importing start method means a missing __main__ guard.

Source code in src/thresher/backends/mp_backend.py

def tally_candidates(
    self,
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]:
    """Count correct predictions per candidate, sharded across processes.

    Args:
        candidates: the thresholds to score.
        scores: the values being split.
        actual_classes: the matching classes, as -1 and 1.

    Returns:
        One count per candidate, identical to what the local backend returns.

    Raises:
        ParallelBootstrapError: if the worker processes could not start, which on a
            re-importing start method means a missing `__main__` guard.
    """
    boundaries = self._shards(len(scores))
    if not candidates:
        return [0] * len(candidates)
    if len(boundaries) <= 1:
        # One shard is just the local backend with extra steps.
        return tally_chunk(candidates, scores, actual_classes)

    shared = list(candidates)
    return merge_tallies(
        self._map(
            tally_chunk,
            [
                (shared, list(scores[start:stop]), list(actual_classes[start:stop]))
                for start, stop in boundaries
            ],
        )
    )

class_counts_by_score

class_counts_by_score(
    scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]

Count classes per distinct score, sharded across processes.

This is the step that makes the exact sweep parallel: each worker returns one count per distinct score it saw, never the samples.

Parameters:

Name	Type	Description	Default
scores	Sequence[float]	the values being split.	required
actual_classes	Sequence[int]	the matching classes, as -1 and 1.	required

Returns:

Type	Description
dict[float, ClassCounts]	A mapping of score to (negatives, positives), identical to the local result.

Raises:

Type	Description
ParallelBootstrapError	if the worker processes could not start.

Source code in src/thresher/backends/mp_backend.py

def class_counts_by_score(
    self, scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]:
    """Count classes per distinct score, sharded across processes.

    This is the step that makes the exact sweep parallel: each worker returns one count
    per distinct score it saw, never the samples.

    Args:
        scores: the values being split.
        actual_classes: the matching classes, as -1 and 1.

    Returns:
        A mapping of score to `(negatives, positives)`, identical to the local result.

    Raises:
        ParallelBootstrapError: if the worker processes could not start.
    """
    boundaries = self._shards(len(scores))
    if not boundaries:
        return {}
    if len(boundaries) == 1:
        return count_chunk(scores, actual_classes)

    return merge_counts(
        self._map(
            count_chunk,
            [(list(scores[start:stop]), list(actual_classes[start:stop])) for start, stop in boundaries],
        )
    )

thresher.backends.mp_backend.resolve_worker_count

resolve_worker_count(num_workers: int | None) -> int

Turn a requested worker count into a usable number of processes.

Shared with linear search's n_jobs, which names the same quantity, so the two cannot disagree about what -1 means or about which values are refusable.

Parameters:

Name	Type	Description	Default
num_workers	int | None	how many processes to ask for. None means one per processor, -1 every processor bar one - the historical meaning of n_jobs=-1.	required

Returns:

Type	Description
int	A process count of at least 1, never more than the machine has processors.
int	Over-asking is clamped rather than refused: how many cores are available is a
int	property of the machine, not a mistake in the caller's code.

Raises:

Type	Description
ConfigurationError	for 0, or anything below -1, which name no sensible number of processes. It is a ValueError.

Source code in src/thresher/backends/mp_backend.py

def resolve_worker_count(num_workers: int | None) -> int:
    """Turn a requested worker count into a usable number of processes.

    Shared with linear search's `n_jobs`, which names the same quantity, so the two cannot
    disagree about what `-1` means or about which values are refusable.

    Args:
        num_workers: how many processes to ask for. `None` means one per processor, `-1`
            every processor bar one - the historical meaning of `n_jobs=-1`.

    Returns:
        A process count of at least 1, never more than the machine has processors.
        Over-asking is clamped rather than refused: how many cores are available is a
        property of the machine, not a mistake in the caller's code.

    Raises:
        ConfigurationError: for 0, or anything below -1, which name no sensible number of
            processes. It is a `ValueError`.
    """
    available = multiprocessing.cpu_count()

    if num_workers is None:
        return available
    if (
        not isinstance(num_workers, int)
        or isinstance(num_workers, bool)
        or num_workers == 0
        or num_workers < -1
    ):
        raise ConfigurationError(INVALID_WORKERS.format(got=num_workers))
    if num_workers == -1:
        # Leave a processor for the caller's own machine, as this has always meant.
        return max(1, available - 1)
    return min(num_workers, available)

Ray

thresher.backends.ray_backend.RayBackend

RayBackend(
    num_shards: int | None = None,
    min_rows_per_shard: int = DEFAULT_MIN_ROWS_PER_SHARD,
)

Count in parallel across a Ray cluster.

Connects to whatever cluster Ray is already attached to. If Ray has not been initialised, it is started locally with default settings - so a caller who has already called ray.init(address=...) keeps their cluster, and one who has not gets a working local cluster without ceremony.

Configure how the data is divided.

Parameters:

Name	Type	Description	Default
num_shards	int | None	how many shards to split into. Defaults to the cluster's CPU count, which is the useful maximum since each shard occupies one worker.	None
min_rows_per_shard	int	do not produce shards smaller than this. Sharding a small dataset costs more in scheduling than it saves in computation.	DEFAULT_MIN_ROWS_PER_SHARD

Raises:

Type	Description
BackendDependencyError	if Ray is not installed, an ImportError. Checked here rather than on first use, so the failure lands when the backend is asked for instead of partway through a long run.

Source code in src/thresher/backends/ray_backend.py

def __init__(
    self, num_shards: int | None = None, min_rows_per_shard: int = DEFAULT_MIN_ROWS_PER_SHARD
) -> None:
    """Configure how the data is divided.

    Args:
        num_shards: how many shards to split into. Defaults to the cluster's CPU
            count, which is the useful maximum since each shard occupies one worker.
        min_rows_per_shard: do not produce shards smaller than this. Sharding a small
            dataset costs more in scheduling than it saves in computation.

    Raises:
        BackendDependencyError: if Ray is not installed, an `ImportError`. Checked
            here rather than on first use,
            so the failure lands when the backend is asked for instead of partway
            through a long run.
    """
    _require_ray()
    self._num_shards = num_shards
    self._min_rows_per_shard = min_rows_per_shard

tally_candidates

tally_candidates(
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]

Count correct predictions per candidate, sharded across the cluster.

The candidate list is put into the object store once and shared by reference, so it is not re-serialised per shard.

Parameters:

Name	Type	Description	Default
candidates	Sequence[float]	the thresholds to score.	required
scores	Sequence[float]	the values being split.	required
actual_classes	Sequence[int]	the matching classes, as -1 and 1.	required

Returns:

Type	Description
list[int]	One count per candidate, identical to what the local backend returns.

Source code in src/thresher/backends/ray_backend.py

def tally_candidates(
    self,
    candidates: Sequence[float],
    scores: Sequence[float],
    actual_classes: Sequence[int],
) -> list[int]:
    """Count correct predictions per candidate, sharded across the cluster.

    The candidate list is put into the object store once and shared by reference, so
    it is not re-serialised per shard.

    Args:
        candidates: the thresholds to score.
        scores: the values being split.
        actual_classes: the matching classes, as -1 and 1.

    Returns:
        One count per candidate, identical to what the local backend returns.
    """
    ray = self._ray()
    boundaries = self._shards(ray, len(scores))
    if not boundaries or not candidates:
        return [0] * len(candidates)

    remote_tally = ray.remote(tally_chunk)
    candidate_ref = ray.put(list(candidates))

    futures = [
        remote_tally.remote(candidate_ref, list(scores[start:stop]), list(actual_classes[start:stop]))
        for start, stop in boundaries
    ]
    return merge_tallies(ray.get(futures))

class_counts_by_score

class_counts_by_score(
    scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]

Count classes per distinct score, sharded across the cluster.

This is the step that makes the exact sweep distributable: the driver never sees the samples, only one count per distinct score.

Parameters:

Name	Type	Description	Default
scores	Sequence[float]	the values being split.	required
actual_classes	Sequence[int]	the matching classes, as -1 and 1.	required

Returns:

Type	Description
dict[float, ClassCounts]	A mapping of score to (negatives, positives), identical to the local result.

Source code in src/thresher/backends/ray_backend.py

def class_counts_by_score(
    self, scores: Sequence[float], actual_classes: Sequence[int]
) -> dict[float, ClassCounts]:
    """Count classes per distinct score, sharded across the cluster.

    This is the step that makes the exact sweep distributable: the driver never sees
    the samples, only one count per distinct score.

    Args:
        scores: the values being split.
        actual_classes: the matching classes, as -1 and 1.

    Returns:
        A mapping of score to `(negatives, positives)`, identical to the local result.
    """
    ray = self._ray()
    boundaries = self._shards(ray, len(scores))
    if not boundaries:
        return {}

    remote_count = ray.remote(count_chunk)

    futures = [
        remote_count.remote(list(scores[start:stop]), list(actual_classes[start:stop]))
        for start, stop in boundaries
    ]
    return merge_counts(ray.get(futures))