[WIP] Parallel linclust - #1124
Draft
bbuschkaemper wants to merge 45 commits into
Draft
Conversation
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>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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:
seqkey_to_len,countTable,repSequence(kmermatcher.cpp)DBReader::Index[]assignedCluster[dbSize](Align2clust.cpp:445)std::list<size_t>[N](mergeclusters.cpp:28)There is also a time cost. When the k-mer array does not fit memory,
kmermatchersplits andre-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
hashUInt64score kmermatcher already computes, so every occurrence of a k-mer lands in the samepartition and a partition can be grouped on its own. The division is lossless. Sequence-space
sharding is not.
Keys are dense and length-ranked:
createdbparallelwrites sequences longest first, so a key isits global length rank. That removes
SORT_BY_LENGTHand its side arrays, lets an entry beaddressed 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_addon a counter file, so a stage maps onto a Slurm array job and workerscan join late, die, or restart. Items are claimed under a lease and re-claimed if a worker dies.
Locking is
fcntlwhole-file locks, notflock, which is node-local on GPFS and Lustre.Nine commands, driven by
data/workflow/linclustparallel.sh: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-recordswrites 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:
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:
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
Limitations
--cov-mode 1or2only. Symmetric coverage modes make linclust selectSET_COVERplus the count-table rounds; neither is implemented, so the command refuses them.alignparallelrejects nucleotides.representative<TAB>memberin accessions, not a cluster DB. A per-key index is state no single node can hold at this scale.NULLparameters and guarded branches inkmermatcher.{cpp,h}, andparsePrecisionLibde-duplicated intoMatcher.cpp. Both werechecked behaviour-preserving against a reference binary built from the base commit.
Parameters.cppalso relaxescheckIfDatabaseIsValidso amkdirrace is tolerated when thedirectory already exists, which fixes a real Slurm-array race.