[Fix] Multi-threaded pipe bottleneck

[noel-improv]

Package

lexical-graph

Problem statement

extract() fans documents out to N workers, then pulls everything back through a single generator chain in the parent (lexical_graph_index.py:498 and :503):

nodes | extraction_pipeline | Pipe(handler.accept) | build_pipeline | sink

Pipe is plain generator composition with no threads, so extraction scales with workers but every document afterwards passes through one stage, one at a time. Eight times the workers returns 1.58x, converging near 33 docs/min.

At 1M documents that works out to roughly 21 days no matter how large the fleet, against a Bedrock quota floor of about 1.5 days.

Evidence

Two runs sit behind this. The 33 docs/min ceiling comes from the worker sweep: 2,000 documents, 2/4/8/16 workers at 16 threads each, on-demand, ml.m5.12xlarge. The rows below come from the single-worker runs: 500 documents, one worker, on-demand, ml.m5.4xlarge. Each run splits into a model phase and a drain phase, and the drain costs about 1.4 s per document.

Variable changed	Drain	docs/min
16 / 32 / 64 threads	13.10 / 11.43 / 11.72 min	38 / 44 / 43
local disk instead of S3	14.1 min	35.5
Nova 2 Lite instead of Sonnet 4.6	12.7 min	39.4

Removing S3 made it slower, which rules out upload cost. And the drain hardly moves between two models whose extraction phases differ by 2.8x.

Four candidates sit in that stage, none of them timed separately: extraction_monitor (:496, present only when a progress monitor is passed), Pipe(handler.accept), build_pipeline (created with num_workers=1 hardcoded at :490 and running only a NullBuilder), and sink.

Proposed solution

Time the four stages first, which is about ten lines. Two earlier attempts to fix this by reasoning from the code targeted the wrong component. One raised the uploader's effective thread count from the default of 4 to 64 and measured 1.15x end to end, which is real but leaves the drain untouched.

Then, in increasing order of effort:

1. Raise num_workers on the build pipeline (:490). One line, resting on the untested assumption that it's free.
2. Give the handler stage a bounded reorder window. It has to yield in order and only once writes are durable.
3. Remove the funnel entirely so workers write their own output rather than returning documents to the parent. This is the only option that scales with worker count.

Single run per configuration. The 13.10 / 11.43 / 11.72 thread row is a 13% spread with no repeat, so read it as "does not improve" rather than as a flat line. The local-disk and cross-model rows are the stronger evidence: both change the drain by less than the upstream phase they alter.

[mykola-pereyma]

Took a look at this from the code. Short version: there are two independent bottlenecks, and fixing only the barrier won't move the ceiling.

1. The barrier — run_pipeline does an ordered p.map gather inside a with ProcessPoolExecutor(...) per batch, so the parent blocks on the slowest bucket and a fresh (spawn) pool is created/destroyed every batch. This hurts scaling efficiency (why 8× workers → ~1.58×).

2. The single-threaded serial tail — after extraction, results flow through one lazy generator chain in the parent: extraction | [monitor] | [handler] | build_pipeline | sink. The per-document write (handle_output_doc) — and, on extract_and_build, the downstream build stage — run one document at a time. A single-threaded terminal stage caps throughput at 1 / (per-doc cost) regardless of worker count. If the drain really is ~1.4 s/doc, that's a hard ceiling around ~43 docs/min no matter how many workers you add.

Implication: removing the barrier alone gets you ~33 → ~43 docs/min, not the quota ceiling. To scale past that, the write has to come off the parent's critical path — ideally workers write their own output (issue's option 3), so throughput becomes extraction-bound and scales with workers. That's the lever.

Two things to pin down before building anything:

• Which path/entrypoint produced the ~33 docs/min number? The synchronous path (_extract_fixed_batch) has this bottleneck; the Bedrock batch path (_extract_auto_tuned → BatchExtractorBase, S3 + async batch jobs) is a different model entirely and wouldn't be helped by this. Also extract() vs extract_and_build() matters — on extract() the build is a no-op NullBuilder(num_workers=1), so the tail is just the extraction write.
• What write concurrency the backend (S3 / graph store) actually sustains — that sets the real ceiling once the drain is parallelised, independent of the pipeline.