tokenization: stop starting a process pool that makes it slower

Asking where the counting loop sits in a full run turned up something
larger next door. Tokenization was gated on a fixed
PARALLEL_MIN_CHARS = 2_000_000 -- but the measurement table sitting
directly above that constant records the pool losing at EVERY size it
covers, up to 163M characters, and at exactly 2M it records serial
0.075s against pool 3.870s. The conclusion was never drawn, so every
corpus over a couple of megabytes paid ~3s of spawn startup to tokenize
more slowly.

Re-measured with the v2 tokenizer on synthetic Zipfian text:

     6.6M chars   serial 0.42s   pool 2.91s
    26.2M chars   serial 1.80s   pool 3.52s
    52.5M chars   serial 3.72s   pool 5.07s

The gap narrows but does not close; extrapolation puts break-even beyond
100M characters, which for a package aimed at small domain corpora is
never. The cost is not the tokenizing, it is shipping every string to a
worker and every token list back.

The decision is now measured -- sample, extrapolate, require a margin --
mirroring pair_counts._pool_is_worth_starting, which is already accurate
in practice (estimated 2.81s against an actual 3.09s). The probe strides
across the input instead of taking its head, because corpora are usually
ordered and a head is not a fair sample. The optimism of the
serial/workers model is documented where it lives rather than hidden.

Measured effect on a 1.5M-token run: corpus construction 4.05s -> 1.73s,
end to end 12.41s -> 10.04s (19%).

Scheduling never affected the result: map_pool preserves order and the
tokenizer is pure, so serial and pooled output are identical and no score
moves. The two tests that forced the pool through the character threshold
now force the measured verdict instead, and the case that was actually
broken -- clearing the character gate must NOT be enough on its own -- is
now a test of its own.

Author: jfilter

CHANGELOG.md

@@ -7,6 +7,21 @@ Everything below is user-visible; several items change numeric results.
 
 ### Added
 
+-   **Tokenization no longer starts a process pool that makes it slower.** The
+    pool was gated on a fixed `PARALLEL_MIN_CHARS = 2_000_000`, but the
+    measurements recorded next to that constant show the pool losing at *every*
+    size tested, up to 163M characters -- so every corpus above a couple of
+    megabytes paid ~3s of spawn startup to tokenize more slowly. Re-measured with
+    the v2 tokenizer: 6.6M chars 0.42s serial vs 2.91s pooled; 52.5M chars 3.72s
+    vs 5.07s. The decision is now *measured* (sample, extrapolate, require a
+    margin) the same way `count_pairs` already decides, so it self-calibrates
+    instead of being quietly wrong. On a 1.5M-token corpus this cuts corpus
+    construction from 4.05s to 1.73s and a full corpus->count->PMI->SVD->evaluate
+    run from 12.41s to 10.04s (**19% end to end**).
+
+    Scheduling never affected the result -- `map_pool` preserves order and the
+    tokenizer is pure -- so no score or matrix changes.
+
 -   **A dead worker now explains itself.** A script without an
     `if __name__ == "__main__":` guard makes every spawned worker re-run the
     script from the top — including the `hyperhyper` call that started the pool —

hyperhyper/preprocessing.py

@@ -6,11 +6,12 @@
 import multiprocessing
 import re
 import sys
+import time
 import unicodedata
 
 from tqdm import tqdm
 
-from .utils import map_pool
+from .utils import _default_workers, map_pool
 
 logger = logging.getLogger(__name__)
 
@@ -74,6 +75,110 @@ def _preprocess_string(s, filters):
 # tokenization.
 PARALLEL_MIN_CHARS = 2_000_000
 
+# The threshold above turned out to be far too low, and the table above says why
+# without having drawn the conclusion: the pool is a *net loss at every size that
+# was measured*, including 163M characters -- yet 2M lets it run, and at exactly
+# 2M the same table records serial 0.075s against pool 3.870s. Every corpus
+# larger than a couple of megabytes therefore paid ~3s of spawn startup to make
+# tokenization slower.
+#
+# Re-measured with `tokenize_string_v2` on synthetic Zipfian text (M1 Pro, 10
+# workers, Python 3.12):
+#
+#      6.6M chars   serial 0.42s   pool 2.91s
+#     26.2M chars   serial 1.80s   pool 3.52s
+#     52.5M chars   serial 3.72s   pool 5.07s
+#
+# The gap narrows but does not close: extrapolating the two slopes puts
+# break-even beyond 100M characters, which for a package aimed at *small,
+# domain-specific* corpora is never. The cost is not the tokenizing, it is
+# shipping every string to a worker and every token list back.
+#
+# So the decision is no longer a fixed character count. It is measured, the same
+# way `pair_counts._pool_is_worth_starting` measures it: tokenize a sample,
+# extrapolate, and only start a pool if it beats serial by a margin. That
+# self-calibrates on machines and text this table never saw, and it cannot go on
+# being quietly wrong.
+#
+# Scheduling has no effect on the *result*: `map_pool` preserves order and the
+# tokenizer is a pure function, so serial and pooled output are identical.
+
+# What a pool costs before it tokenizes anything -- dominated by each spawned
+# child importing the package. Same measurement as `pair_counts`.
+POOL_STARTUP_SECONDS = 3.0
+
+# How much faster the pool must look before it is worth starting.
+POOL_SPEEDUP_MARGIN = 1.3
+
+# Texts tokenized in-process to estimate the per-text cost: enough to be above
+# timer noise, few enough to be a rounding error on any corpus where the answer
+# is not already obvious.
+PROBE_TEXTS = 2000
+
+
+def _pool_is_worth_starting(texts, tokenizer):
+    """
+    Whether tokenizing `texts` across a process pool beats doing it here.
+
+    Measured rather than derived from a character count, because the per-character
+    cost is not a property of the corpus: it swings with text shape (many short
+    lines cost more per character than few long ones) and with the tokenizer.
+    """
+    workers = _default_workers()
+    if len(texts) < 2 or workers < 2:
+        return False
+
+    # Sample with a stride rather than taking the first N: a corpus is very
+    # often ordered (by document, by date, by source), so its head is not a fair
+    # sample of it. Probing `texts[:N]` of a corpus whose short headlines come
+    # first would underestimate the work and refuse a pool that was worth
+    # starting -- and the reverse for a corpus that opens with long documents.
+    step = max(1, len(texts) // PROBE_TEXTS)
+    sample = texts[::step][:PROBE_TEXTS]
+    start = time.perf_counter()
+    for t in sample:
+        tokenizer(t)
+    elapsed = time.perf_counter() - start
+    if elapsed == 0:
+        return False
+
+    serial = elapsed / len(sample) * len(texts)
+    # `serial / workers` is the *optimistic* bound: it counts the tokenizing but
+    # not the cost of shipping every string to a worker and every token list
+    # back, which the measurements above show to be the dominant term for this
+    # workload. The estimate is therefore biased towards the pool, and
+    # `POOL_SPEEDUP_MARGIN` is what keeps that bias from deciding marginal cases.
+    parallel = POOL_STARTUP_SECONDS + serial / min(workers, len(texts))
+    logger.debug(
+        "tokenizing %d texts: serial ~%.2fs, pool ~%.2fs", len(texts), serial, parallel
+    )
+    return parallel * POOL_SPEEDUP_MARGIN < serial
+
+
+def _should_pool(texts, tokenizer):
+    """
+    The single place that decides serial vs pool for tokenization.
+
+    Three gates, cheapest first:
+
+    1. **Inside a worker** -- never. `Corpus.from_text_files` already runs the
+       tokenizer in a pool of `workers` processes, so nesting would ask for
+       `workers ** 2`.
+    2. **Below `PARALLEL_MIN_CHARS`** -- never, without measuring. This keeps
+       the common small case (the evaluation preprocesses one dataset column at
+       a time) from paying even for the probe.
+    3. **Otherwise, measured** -- see `_pool_is_worth_starting`.
+
+    All three are scheduling only: `map_pool` preserves order and the tokenizer
+    is pure, so the tokens are identical whichever way this goes.
+    """
+    if multiprocessing.parent_process() is not None:
+        return False
+    if sum(len(t) for t in texts) < PARALLEL_MIN_CHARS:
+        return False
+    return _pool_is_worth_starting(texts, tokenizer)
+
+
 # spaCy is imported lazily, not at module import time: it costs ~2.2s, it is
 # only ever needed by `texts_to_sents`, and every process-pool child that
 # imports this module used to pay for it. `_UNSET` distinguishes "not tried
@@ -154,10 +259,7 @@ def tokenize_texts_parallel(texts):
     if not hasattr(texts, "__len__"):
         texts = list(texts)
 
-    if multiprocessing.parent_process() is not None:
-        return tokenize_texts(texts)
-
-    if sum(len(t) for t in texts) < PARALLEL_MIN_CHARS:
+    if not _should_pool(texts, tokenize_string):
         return tokenize_texts(texts)
 
     return map_pool(texts, tokenize_string)
@@ -263,10 +365,7 @@ def tokenize_texts_parallel_v2(texts):
     if not hasattr(texts, "__len__"):
         texts = list(texts)
 
-    if multiprocessing.parent_process() is not None:
-        return tokenize_texts_v2(texts)
-
-    if sum(len(t) for t in texts) < PARALLEL_MIN_CHARS:
+    if not _should_pool(texts, tokenize_string_v2):
         return tokenize_texts_v2(texts)
 
     return map_pool(texts, tokenize_string_v2)

tests/test_evaluation.py

@@ -885,6 +885,7 @@ def test_parallel_preprocessing_does_not_change_any_score(
     assert count_pools == []
 
     monkeypatch.setattr(preprocessing, "PARALLEL_MIN_CHARS", 0)
+    monkeypatch.setattr(preprocessing, "_pool_is_worth_starting", lambda *_: True)
     pooled = evaluation.eval_similarity(
         toy_embedding, TOKEN2ID, tokenize_texts_parallel, lang="en"
     )

tests/test_preprocessing.py

@@ -13,6 +13,7 @@
 
 import subprocess
 import sys
+import time
 import unicodedata
 from concurrent import futures
 
@@ -167,11 +168,12 @@ def test_small_input_does_not_spawn_a_pool(pool_calls):
 
 def test_large_input_still_uses_the_pool(monkeypatch, pool_calls):
     """
-    The threshold must not quietly delete the parallel path that large corpora
-    rely on. Lowering it is equivalent to handing in a huge input, without
-    building one.
+    The decision must not quietly delete the parallel path that large corpora
+    rely on. Forcing the verdict is equivalent to handing in an input big enough
+    for the pool to pay, without building one.
     """
     monkeypatch.setattr(preprocessing, "PARALLEL_MIN_CHARS", 1)
+    monkeypatch.setattr(preprocessing, "_pool_is_worth_starting", lambda *_: True)
 
     result = tokenize_texts_parallel(TEXTS)
 
@@ -180,16 +182,62 @@ def test_large_input_still_uses_the_pool(monkeypatch, pool_calls):
     assert result == tokenize_texts(TEXTS)
 
 
-def test_threshold_counts_characters_not_items(monkeypatch, pool_calls):
+def test_character_threshold_short_circuits_before_measuring(monkeypatch, pool_calls):
+    """
+    Below `PARALLEL_MIN_CHARS` the decision is made without measuring at all --
+    the probe would otherwise cost more than tokenizing the whole input, which
+    is the small-input case this gate exists for. The threshold counts
+    characters, not items.
+    """
+    measured = []
+    monkeypatch.setattr(
+        preprocessing,
+        "_pool_is_worth_starting",
+        lambda *_: measured.append(1) or True,
+    )
     monkeypatch.setattr(preprocessing, "PARALLEL_MIN_CHARS", 100)
 
     tokenize_texts_parallel(["a b c"] * 5)  # 25 chars, way under
     assert pool_calls == []
+    assert measured == []  # not even probed
 
-    tokenize_texts_parallel(["x" * 60] * 2)  # 120 chars, over
+    tokenize_texts_parallel(["x" * 60] * 2)  # 120 chars, over -> now measured
+    assert measured == [1]
     assert len(pool_calls) == 1
 
 
+def test_pool_is_refused_when_it_would_not_pay(monkeypatch, pool_calls):
+    """
+    The bug this replaced: `PARALLEL_MIN_CHARS` alone let every corpus over a
+    couple of megabytes into a pool that was measured to be a NET LOSS at every
+    size up to 163M characters -- paying ~3s of spawn startup to make
+    tokenization slower. Clearing the character gate must not be enough; the
+    work has to actually be worth distributing.
+    """
+    monkeypatch.setattr(preprocessing, "PARALLEL_MIN_CHARS", 1)
+
+    result = tokenize_texts_parallel(TEXTS)
+
+    assert pool_calls == []  # a handful of short strings: nowhere near worth it
+    assert result == tokenize_texts(TEXTS)
+
+
+def test_pool_verdict_scales_with_the_measured_cost(monkeypatch):
+    """
+    The verdict is measured, not derived from a size: a tokenizer slow enough
+    that distributing it beats a ~3s pool startup is accepted, the same input
+    with a fast one is refused.
+    """
+    texts = ["a b c"] * 5000
+
+    def slow(text):
+        time.sleep(0.001)
+        return text.split()
+
+    assert preprocessing._pool_is_worth_starting(texts, slow) is True
+    assert preprocessing._pool_is_worth_starting(texts, str.split) is False
+
+
 def test_accepts_an_iterator(pool_calls):
     """
     `eval_similarity` hands in the columns of a dataset, which are tuples, and
@@ -238,3 +286,29 @@ def test_a_disabled_spacy_is_not_re_imported(monkeypatch):
 
     assert preprocessing._import_spacy() is None
     assert preprocessing.spacy is None
+
+
+def test_pool_probe_samples_across_the_input(monkeypatch):
+    """
+    A corpus is usually ordered -- by document, by date, by source -- so its head
+    is not a fair sample of it. Probing only `texts[:N]` of a corpus that opens
+    with short lines and ends with long ones would underestimate the work and
+    refuse a pool worth starting.
+    """
+    seen = []
+
+    def record(text):
+        seen.append(text)
+        return text.split()
+
+    monkeypatch.setattr(preprocessing, "PROBE_TEXTS", 10)
+    texts = [f"t{i}" for i in range(1000)]
+    preprocessing._pool_is_worth_starting(texts, record)
+
+    assert len(seen) == 10
+    # spread across the input rather than taken from its head: a stride of
+    # `len // PROBE_TEXTS` reaches the last decile, where `texts[:10]` would
+    # never leave the first percent
+    assert seen[0] == "t0"
+    assert seen[-1] == "t900"
+    assert seen == [f"t{i}" for i in range(0, 1000, 100)]