Skip to content

[WIP] Parallel linclust - #1124

Draft
bbuschkaemper wants to merge 45 commits into
soedinglab:masterfrom
bbuschkaemper:parallel-linclust
Draft

[WIP] Parallel linclust#1124
bbuschkaemper wants to merge 45 commits into
soedinglab:masterfrom
bbuschkaemper:parallel-linclust

Conversation

@bbuschkaemper

@bbuschkaemper bbuschkaemper commented Aug 3, 2026

Copy link
Copy Markdown
Contributor

Shared-filesystem parallel linclust

Work in progress. Runs linclust across many Slurm workers on a shared filesystem, with no MPI and
no node-to-node communication, to allow clustering 1e11-1e12 sequences.

Why

We assume availability of up to 2TB memory nodes and large amounts of shared filesystem storage ("scratch").
Several structures in linclust are sized by key space or by the whole database, so they cannot
exist at 1e12 on a 2 TB node:

structure at 1e12
seqkey_to_len, countTable, repSequence (kmermatcher.cpp) 2-4 TB each
resident DBReader::Index[] ~24 TB
assignedCluster[dbSize] (Align2clust.cpp:445) 8 TB
std::list<size_t>[N] (mergeclusters.cpp:28) 24 TB of empty headers

There is also a time cost. When the k-mer array does not fit memory, kmermatcher splits and
re-extracts every k-mer per split, roughly 68 times over at 1e12.

Approach

Partition k-mer space rather than sequence space. The partition of a k-mer is the low bits of the
hashUInt64 score kmermatcher already computes, so every occurrence of a k-mer lands in the same
partition and a partition can be grouped on its own. The division is lossless. Sequence-space
sharding is not.

Keys are dense and length-ranked: createdbparallel writes sequences longest first, so a key is
its global length rank. That removes SORT_BY_LENGTH and its side arrays, lets an entry be
addressed by key with no resident index, and makes stock's longest-first greedy the same thing as
ascending key order. Clustering becomes one left-to-right sweep needing 2 bits per key instead of
8 bytes, so 25 GB at 1e11 rather than 800 GB.

Coordination is files only. Every worker of a stage runs the same command line and takes its
identity from a fetch_add on a counter file, so a stage maps onto a Slurm array job and workers
can join late, die, or restart. Items are claimed under a lease and re-claimed if a worker dies.
Locking is fcntl whole-file locks, not flock, which is node-local on GPFS and Lustre.

Nine commands, driven by data/workflow/linclustparallel.sh:

createdbparallel        FASTA -> dense, length-ranked sequence DB
kmermatcherparallel     extract k-mers into P partition buckets, one wave at a time
kmerreduceparallel      per partition: group, emit candidate edges bucketed by representative key
alignparallel           per bucket: merge duplicate copies of a pair, align once
greedycluster           single node, one key-ordered sweep
createrepdb / translatecluster / mergeclusterparallel / translatekeys

Two decisions that were measured, not assumed

Alignment is keyed by representative range, not by k-mer partition. We built the fused version
first. A k-mer partition's pairs are spread over the whole key space, so one partition needed
1.61 GB of sequences and read 83.9 GB, a 52x amplification. Bucketing edges by representative key
gives 1.15x, removes cross-partition duplicates before aligning rather than paying for them, and
reproduces stock's global per-(pair, diagonal) accumulation exactly.

Bucket records are packed. K-mer records went from 24 to ~14 bytes, candidate edges from 17 to
~7. --raw-records writes the old fixed-width form, so the two can be run against each other;
they produce identical output. Peak scratch is 0.67x what the first version of this branch used.

Results

Verified identical results on a 1M subsample of MGnify sequences, current master branch ("stock") and this branch on the same database:

0 of 1,000,000 sequences in a differing cluster
seq -> rep identical for 1,000,000 of 1,000,000

Output is byte-identical across worker counts, wave counts, thread counts, record encodings, reduce slice counts, 32- and 64-bit builds, and after kill-and-resume.

Comparing against stock on a differently keyed database is not meaningful: stock against itself
on input-order versus length-ranked keys moves 3.83% of sequences.

Speed on one machine, 128 cores total, FASTA to TSV:

stock this branch, 8 workers
1M 21 s 50 s
10M 137 s 121 s
100M 1334 s 1078 s

Peak memory at 100M is 61.9 GB against stock's 91.6 GB. On a 64-bit-id build, which a real run
needs, the gap widens: 5.7 GB against 11.2 GB at 10M, because stock's key-space-sized structures
scale with key width and these do not.

Splitting the same cores further has diminishing returns: 1 to 4 workers is worth 2.7x, 4 to 8
only 1.08x.

Not finished

  • Multi-node, real slurm was verified on smaller subsamples (1-10M) only, at least a 1B stock vs. multi-node parallel run should be done.
  • At 1e12 the map has ~1e6 work items, each costing two fsyncs through one global lock. This needs batched claims and sharded queue files.
  • No checks up front that linclust's pass-2 will fit the scratch budget, so a long run can still fail late.

Limitations

  • --cov-mode 1 or 2 only. Symmetric coverage modes make linclust select SET_COVER plus the count-table rounds; neither is implemented, so the command refuses them.
  • Protein only. alignparallel rejects nucleotides.
  • Output is representative<TAB>member in accessions, not a cluster DB. A per-key index is state no single node can hold at this scale.
  • Sequences are capped at 65,535 residues and rejected above it.
  • Stock is touched in 6 files (+150/−77): two defaulted-NULL parameters and guarded branches in
    kmermatcher.{cpp,h}, and parsePrecisionLib de-duplicated into Matcher.cpp. Both were
    checked behaviour-preserving against a reference binary built from the base commit.
    Parameters.cpp also relaxes checkIfDatabaseIsValid so a mkdir race is tolerated when the
    directory already exists, which fixes a real Slurm-array race.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
…es).

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
…location during translate keys.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Foundations for the packed bucket formats. The length-rank table recovers a
sequence length from its key, which the length-ranked key assignment already
fixes, so the k-mer record no longer has to carry seqLen.
K-mer records 24 -> ~14 B, candidate edges 17 -> ~7 B, framed with a magic,
record count, length and checksum so a torn tail is recognisable. Adds
--raw-records, which writes the old fixed-width form as an exactness control,
and --write-header-db, since nothing between createdb and the final TSV reads
the header database. createrepdb now writes the pass-2 sub-database's
length-rank table.

Peak scratch falls to 0.67x the previous implementation at 10M and 100M.
reserve() allocates exactly what is asked, so reserving per block reallocated
and copied the whole accumulated bucket every time. Worth 2.1-2.5x at 100M.
A partition exceeding the worker's memory budget is now grouped one k-mer slice
at a time instead of failing to allocate. Exact: a slice is a pure function of
the k-mer, so a group is never split. --reduce-slices forces the count. The
reduce also reports partition and group sizes.
The heartbeat thread slept in one-second granules, so join() waited up to a
second after every work item. Worth 4.6x on the map at 1M.
TestEdgeCodec covers round-trip, raw/packed equivalence, and rejection of
truncated, over-long and corrupt blocks.
The k-mer partition and edge bucket counts came from --split-memory-limit
alone, so a generous per-node limit made *fewer* work units and left most
workers nothing to claim: kmerreduceparallel and alignparallel ran on one
worker at every scale from 1e7 to 1e9, in both the 4- and 8-worker configs. At
1e9 that was alignparallel on a single node for half a 6h32m run.

The memory figure is now a ceiling, edge buckets target 1 GiB of sequences, and
--workers raises both counts for the allocation. Bucketing is a partitioning,
not a semantic choice, so the clustering does not move: byte-identical output
at 1e6.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
deriveAlignBucketCount's first argument is the sequence database size, not the
edge volume, and the comments called it the edge set throughout; corrected,
along with a worked example that was off by 2x.

The fix was also opt-in and its absence silent, so both stages now warn when a
worker's id is past the work-unit count. An absurd --workers is clamped rather
than fatal, and no longer derives partitions holding almost nothing. The bucket
monotonicity test swept a range where the count never moves; it now sweeps
where it does, and asserts every key maps into a bucket that exists.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The workflow sets alignmentMode to SCORE_COV_SEQID, as linclust does, but
alignparallel neither registered --alignment-mode nor was passed it, so the
stage used its own FAST_AUTO default. At --min-seq-id 0 that degrades to
SCORE_COV and at -c 0 to SCORE_ONLY, which reports the score-per-column
estimate as the sequence identity. Both are silent and both cluster
differently from stock.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
A directory at $OUT passes the -f test, and the final mv -f then files the
clustering inside it and reports success.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The singleton pass cleared each thread's buffer from inside the parallel
region but flushed all --threads slots afterwards. num_threads() is a request,
not a guarantee, so a smaller team left the remaining slots holding the
previous key block's text, which was written again verbatim. The block loop
runs more than once for any database past 64M entries. The singleton counter
counts real work, so it stayed correct and agreed with the corrupt output.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The slicing budget read par.splitMemoryLimit raw while the other eight call
sites wrap it in Util::computeMemory(). The workflow defaults the limit to 0,
so the limit was 0, the slicing never triggered however skewed a partition
was, and a partition that did not fit killed every worker that claimed it in
turn.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The destination array was only bounds-checked when slicing was off, yet the
sliced path is the one whose capacity comes from a separate counting pass over
shards a lapsed map worker may still be appending to. A disagreement between
the two passes wrote past the end of the array and the run continued.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The chunk size scaled with the total input to hit a target chunk count, so
per-worker memory grew with the database -- the opposite of the bound the file
documents. At 1e12 sequences it derived 3.5 GB chunks, around 515 GB per node
at 64 threads, and re-derived the same size on every restart.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Two full sequential passes over the lookup ran before any bucket could start:
one counting lines for the key space, one recording each bucket's first
offset. The lookup is hundreds of gigabytes at 1e10.

Keys ascend, so the key space is the last line's key plus one and each
bucket's offset can be found by binary search. Taking the highest key rather
than the line count is also correct for a lookup with gaps, where the count is
too small and the database rejects its own keys.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The idle wait was in whole seconds, which is a rounding error for an item that
runs for minutes and the dominant cost for a phase that finishes in under a
second. The default is unchanged in effect and every existing caller used it.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Both stages were single-node, together 0.8 h of a 2.7 h serial floor at 1e10
that no node count reduces, and every allocated node is billed through all of
it. Neither was single-node for an algorithmic reason: both are bucketed joins
whose per-bucket work is already independent.

They now claim work from a WorkQueue like the map, reduce and align stages,
gain a scatter over input byte ranges, and end by pwriting their parts into the
output at prefix-sum offsets rather than one worker copying the result.

A shard is named by (bucket, work item, worker) and lives in a per-bucket
directory. The item fixes the order it is read back in; the worker keeps two
attempts at one item off a single path, since the writers append and a second
attempt truncating the first mid-flight leaves a file that is still a whole
number of records, still in range, and silently short. Which attempt counts is
taken from the queue's completion record rather than guessed.

Work-unit counts are raised for the allocation but bounded. Buckets and chunks
both rise with the worker count, and a scatter writes one file per (bucket,
chunk), so unbounded they made the file count rise with the square of the
allocation -- 1.3e8 files per side at 1e11 on 4096 workers. Buckets are bounded
by memory and cannot give way, so the chunk count does. The remap's work item
is a band of source buckets for the same reason.

Buckets are sized against the rows a bucket loads as well as the keys it spans.
Sizing on keys alone bounded the remap array and left both sides of the join
unbounded, so the stages overran the limit they were told to honour.

--workers is derived from the allocation when the caller does not say, and each
stage now says so when a worker cannot get any work.

Output is byte-identical to a single-worker run, verified end to end on the
1M, 10M and 100M MGnify subsamples and across kills mid-stage.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Three of the nine stages divide across the allocation and the rest are one
process on one node, so a single N-node job bills N nodes through the serial
ones. Measured on JURECA at 1e8 those are 17.6% of an 8-node run; extrapolated
to 1e10 they are ~1.9 h, 39% of a 32-node allocation.

PHASES selects a subset of "p1 s1 p2 s2 p3 s3", so the groups can be submitted
as a dependency chain with the multi-worker ones on N nodes and the single-node
ones on one. Every phase was already idempotent -- the multi-worker stages
resume from their work queues, the single-node ones are guarded on their
output -- so the split needed no new state, only the two dropIntermediate calls
moved behind the greedy sweep that consumes what they delete.

Verified byte-identical to an unsplit run on 1M MGnify sequences.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
The driver is invoked from a batch script that inherits the whole environment,
which for Slurm means --export=ALL. PHASES is a name a user may plausibly
already have set for something else, and the driver carries a comment about
WORKER_RUNNER for exactly this reason: it had to be renamed out of a collision
with RUNNER. LINCLUST_PHASES cannot collide the same way.

An unknown value is still refused rather than ignored, so a collision that did
happen would stop the run instead of silently skipping stages.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Both clu1.tsv scans derived their byte range from omp_get_thread_num() inside a
bare `omp parallel` with no `omp for`, so every range belonging to a thread the
team never created was never read. num_threads() is a request: OMP_THREAD_LIMIT,
OMP_DYNAMIC and a Slurm cpuset all give a smaller team, and all three are
routine on the machines this runs on.

Nothing caught it. repCount is counted by the same loop, so it shrank with the
representatives it missed and planCopy's `kept != keptCount` guard still passed.
The stage exited 0 on a partial representative database, and pass 2 is built
entirely from that.

Measured at 173,313 sequences: 165,481 representatives with a full team of 8,
39,519 under OMP_THREAD_LIMIT=2. After this, twelve combinations of
OMP_THREAD_LIMIT 1..16 and --threads 1..32 all agree on 165,481.

This is the defect greedycluster.cpp:379-387 documents in its singleton pass.
These two scans were missed by that fix; a sweep of the remaining
omp_get_thread_num sites found no others -- translatekeys.cpp:709 has an `omp
for` and createdbparallel.cpp:565 only forms a work-queue id.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
KmerShardReader treated every fopen failure as the benign case and a zero-byte
fread as a clean end of shard, with no ferror() check. The reduce then read the
absent k-mers as "this k-mer had no partner" and wrote a smaller clustering.

Only ENOENT is benign, and it is the race the reduce is built to survive: it
unlinks consumed partitions while a lapsed worker may still be reading them.
EMFILE, EIO and ESTALE are not that. EMFILE is not hypothetical here -- this
pipeline already hit the 8192 descriptor limit at exactly this scale, as
translatekeys.cpp:226-230 records.

Measured on a 4-partition shuffle with one 12.8 MB shard unreadable: 107,050
candidate edges against 142,394, three partitions grouped instead of four, no
warning, exit 0. Now it stops with the errno.

EdgeBucketReader::readShard already makes the equivalent condition fatal, with a
comment saying missing edges are indistinguishable from pairs that never
matched. The same is true of k-mers; the two now agree.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
assemblePart pwrote into the ftruncate'd output and closed without fsync, while
WorkQueue::completeLocked fsyncs the DONE record it writes immediately
afterwards. The two were the wrong way round: a node lost in that window comes
back with the queue saying assembly finished and the bytes never written.

There is no recovery from that. The resume takes the queue.allDone() branch,
skips the whole assemble block, and cleanup unlinks the parts -- and because
createAssembled sizes the output with ftruncate, the missing range reads as NUL
bytes. Simulated by punching the hole a lost writeback leaves: the resume exits
0, logs nothing, and publishes 72,594 NUL bytes of 290,376 with the parts gone.

One fsync per bucket, not per write, so a few thousand at 1e10 against a stage
that moves terabytes. createdbparallel.cpp:789-806 already orders these
correctly and says why; this stage was the one that did not.

Prevention, not repair: damage already on disk is still not detected.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Pass 2 removed each bucket's spill unconditionally, but KeyBuckets::flush creates
a file only for a bucket that received a row and FileUtil::remove is fatal on
ENOENT. A clustering sparse enough to leave one key range empty therefore
aborted the stage that writes the final result: reproduced with a 2-row
clustering against a 173,313-key lookup at --split-memory-limit 1K, which exits
1 with "Could not delete ...bymember.1!" and writes no output. Pass 3 already
guards its own removal this way.

KeyBuckets::read also returned a torn spill as empty, dropping every row in that
key range from clusters.tsv while still exiting 0. translatecluster.cpp:136-146
and mergeclusterparallel.cpp:264-273 both make this fatal, the first saying the
empty return "silently dropped every assignment in that key range" and that the
two now agree -- there are three sites and this was the one missed. Not
reachable today, since removeSpillFiles clears an earlier attempt before pass 1
and every write checks fopen, fwrite and fclose; the guard is for when one of
those stops holding.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
At pos == fileSize - 1 the window is one byte, the pair scan cannot run because
it needs two, the short-read retry does not fire because got == want == 1, and
`pos += got - 1` advances by zero. The loop spun forever at full CPU with
nothing logged, and every worker that re-claimed the chunk hung in the same
place, so the stage neither finished nor failed.

Reached whenever no "\n>" exists at or after a chunk boundary, i.e. the file's
last record starts before it. Reproduced with a 3 MB single-record FASTA at
--chunk-size 1M: no output, no error, killed at a 45 s timeout. It now plans the
one sequence and returns.

Checking against the end of the file before the advance is what makes the loop
terminate: it either breaks at EOF, retries a genuinely short read, or moves
forward by at least one byte.

Signed-off-by: Björn Buschkämper <bjoern.buschkaemper@gmail.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant