perf(pipeline): proportional + auto goroutine budget, parallel ProcessList forward write - #9762
Open
rahst12 wants to merge 17 commits into
Conversation
…sList forward write Builds on the per-predicate mutation pipeline (dgraph-io#9467). Three additive, flag-gated changes that make a single hot/dominant predicate's apply actually parallelize: 1. Proportional goroutine budget across predicates (allocateWorkers): mutations-pipeline-goroutines=N distributes N workers by edge count (largest-remainder), so a dominant predicate gets >1 worker. 2. Auto mode (=-1): budget = min(GOMAXPROCS*fraction, edges/minEdgesPerWorker), derived at runtime. Tunables mutations-pipeline-goroutines-fraction (1.0) and mutations-pipeline-min-edges-per-worker (256). 3. Parallel forward data-write in ProcessList (the merge-light <pred,uid> pass), matching the existing ProcessSingle split, so [uid]/@reverse predicates' forward write parallelizes. ProcessReverse stays SERIAL. Default is off (0) => byte-identical to the current one-goroutine-per- predicate path. Byte-identical equivalence + conflict-key-set tests pass under -race for scalar, [uid] @reverse, and AUTO==fixed. Benchmarks (8-core dev box, ~20k-edge batch) included. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…chemas Confirms the intra-predicate goroutine budget (proportional, auto, and the ProcessList parallel forward write) produces byte-identical committed state and identical conflict-key sets vs. the legacy one-goroutine-per-predicate path, across string (exact/hash/term/fulltext/trigram), int(+count), float, dateTime, bool, geo, [uid]@reverse(+count), uid@reverse, [string], @upsert, @noconflict, and @lang predicates. Generic: scans every committed Badger key (data/index/reverse/count) per predicate and compares dumps. 36 subtests (12 schemas x {fixed8, fixed32, auto}); race-clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
matthewmcneely
self-requested a review
June 23, 2026 21:55
Author
|
TODO: The flags need consolidated to something more manageable. |
… grant ProcessReverse built reverseredMap[targetUid] then wrote each target serially via AddDelta, which holds the GLOBAL txn.cache.Lock() across proto.Marshal. With many predicates applying concurrently at L1 they all convoy on that one mutex. A live goroutine dump (25 [uid] @reverse predicates, 20k triples/txn) showed 61% of pipeline goroutines parked in ProcessReverse and 86% of those blocked on that lock. Two changes: 1. Split "use the lock-free store" from "split across k goroutines". Both decisions were tied to workers > 1, but allocateWorkers hands out mostly 1-worker grants when the budget is below ~2x the predicate count (the shipped default of 30 with 25 predicates yields {1:20, 2:5}), so nearly every predicate fell back to the global lock. Now any enabled budget uses concStore; workers == 0 keeps the legacy locked AddDelta so "budget off == byte-identical legacy" still holds. 2. Above reverseParallelMinTargets (256) distinct targets, partition the target-uid keyspace across the predicate's worker grant. reverseredMap is keyed BY target, so each <~pred,targetUid> key has exactly one writer (I1) — the same disjointness that makes ProcessList's forward write safe. Below the threshold the map holds a few hot targets with large lists; sharding that shape measured as a net loss, so it stays serial. concStore reproduces AddDelta(..., true, true) exactly: prior-delta prepend (a *Txn spans proposals, so dropping it silently loses an earlier proposal's reverse postings) and unconditional sort/dedup, reading only the sharded delta map — never (*Deltas).Get, which also touches indexMap under cache.Lock. Per-worker key buffers are required: addConflictKeyWithUid fingerprints the full buffer including the 8 uid bytes. The reverseredMap BUILD stays serial (many-to-one, shared edge struct); parallelising it needs a merge stage that costs more than it saves. The info.count path returns to ProcessCount before any of this, unchanged. Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn, 64 threads, badger_write_bytes_user: budget=30 (default): 9.63 -> 10.58 MB/s (+9.9%) budget=-1 (auto): 9.68 -> 11.29 MB/s (+16.6%) Reverse-write parking drops 61% -> 37%; global cache.Lock blocking drops 86% -> 41%. Tests: byte-identical committed state and conflict-key sets across budgets {0,8,32,auto} for 13 schemas, incl. two new high-cardinality rows covering the ProcessList and ProcessSingle legs of the parallel path, plus a cross-proposal regression test for the addToList prepend (verified to fail when the prepend is removed). Full posting package green under -race. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
ProcessList and ProcessSingle tied two independent decisions to the same
`workers > 1` test: whether to use the lock-free store, and whether to
split the uid space across goroutines. allocateWorkers hands out mostly
one-worker grants whenever the budget is below ~2x the predicate count —
the shipped default of 30 with 25 predicates yields {1:20, 2:5} — so 20 of
25 predicates fell back to AddDelta, which holds the GLOBAL
txn.cache.Lock() across proto.Marshal. With every predicate goroutine at
L1 doing that, they convoy on one mutex and the parallel forward write
added in 37c834a never engages for most of the batch.
Split the two decisions, mirroring 4a02d8e's fix for ProcessReverse:
workers == 0 keeps the legacy locked AddDelta so "budget off ==
byte-identical legacy" holds; any enabled budget uses the lock-free store
(concStore for ProcessList, which must reproduce AddDelta's addToList
prepend and info.isUid sort/dedup; plain AddDeltaConcurrent for
ProcessSingle, whose serial call already passes addToList=false,
doSortAndDedup=false). The k-way split still requires workers > 1.
Single-writer safety is unchanged: <pred,uid> forward keys are one-to-one,
and x.generateKey encodes attrLen plus a per-type discriminator byte, so
two predicates cannot produce the same key string.
Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse
predicates, 20k triples/txn, 64 threads, badger_write_bytes_user, 2 runs
per arm:
stock 9.63 MB/s
+ reverse (4a02d8e) 10.58 MB/s (+9.9%)
+ this commit 11.44 MB/s (+18.8%)
Budget stops mattering once the lock-free store is reachable at any grant:
auto measures 11.38 MB/s, within noise of the default's 11.44.
The global cache lock disappears from the blocking profile entirely —
goroutines parked on it go 86.4% (stock) -> 41-52% (reverse only) -> 0%,
and runnable goroutines rise to 57.9%. addConflictKeyWithUid's single
global txn.Lock() is now 71.2% of parked pipeline goroutines and is the
next bottleneck. CPU remains at 8% of 64 vCPU: processApplyCh applies one
Raft entry at a time, so a single Alpha cannot saturate a large box
regardless of intra-transaction parallelism.
Full posting package green under -race.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
After 4a02d8e and 612ae8c removed the txn.cache.Lock() convoy, Txn.conflicts became the only remaining mutex contention in the apply path: on a 32-core ingest run 25% of pipeline goroutines were blocked on it and 46% were running inside addConflictKeyWithUid. The map insert was never the problem. A line-level profile attributed 80% of that function's 6.80% of total CPU to acquiring and releasing the mutex (8.75s + 5.33s of 17.86s cum) and only 15% to the map writes (2.71s), with every worker in every predicate contending one lock once per key. So batch instead of shard. Each Process* accumulates its conflict keys in a conflictBuf and flushes them under a single txn.Lock(); parallel workers get their own buffers, merged after wg.Wait(), and never touch txn.Mutex at all. Sharding the container was considered and rejected: it only attacks the 2.71s of inserts (~1% of total CPU) while changing Txn.conflicts' type, nine test sites, FillContext, and NumShards — which is shared with the delta and index maps. Keys are expanded EAGERLY into uint64. Every call site reuses one scratch key buffer across iterations, so a buffer retaining the []byte and fingerprinting at flush time would hash the last uid's bytes for every entry — wrong keys, no panic, nothing for -race to catch. In InsertTokenizerIndexes the flush defer is registered ABOVE cache.Lock() so LIFO runs it after cache.Unlock(); registering it below compiles, passes every test, and silently keeps the txn.Lock()-inside- cache.Lock() nesting this removes. Workers break rather than return on error so already-buffered keys still flush. The pre-existing error path never rolled back emitted keys, so dropping them would change the set — and since Zero's hasConflict is a pure existential, a subset risks a lost update while a superset only costs a spurious abort. Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn, 64 threads, budget=30, 2 runs per arm: badger_write_bytes_user 11.44 -> 12.45 MB/s (+8.8%) triples/s 113001 -> 123732 (+9.5%) Cumulative over stock across the three commits: 9.63 -> 12.45 MB/s (+29%). The conflict-key path disappears from the goroutine profile entirely: share of pipeline goroutines 68.5% -> 0.0%, blocked on its mutex 23.0% -> 0.0%, and total pipeline goroutine observations fall 518 -> 216. The new top blocker is Deltas.AddToDeltas (57.6% of a much smaller blocked population) — shard contention on the delta map, where NumShards = 30 is thin for 64 vCPU. New TestFillContextKeysGolden pins the exact ctx.Keys sent to Zero against a golden captured at 612ae8c, and asserts they are startTs-independent. The six existing byte-identical tests only compare budget 0 vs N, so they cannot catch a regression that changes the serial and parallel paths the same way; nothing asserted FillContext's output before. Full posting package green under -race. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…nquad worker.MaxLeaseId takes a global RWMutex read lock on the membership state to return a single uint64. ExtractBlankUIDs called verifyUid — and thus MaxLeaseId — for EVERY subject and every object uid, including blank nodes, whose uid is 0 and always passes. A 20k-nquad mutation of uid edges therefore took roughly 40k acquisitions of shared cluster state to compare against a number that barely moves. Take one snapshot per ExtractBlankUIDs call and only fall through to verifyUid for uids above it. The lease is monotonically non-decreasing, so a uid at or below the snapshot was already leased and cannot become unleased; a uid above it still goes through the live re-read and the existing wait-for-lease loop, so a uid racing an in-flight lease extension is accepted exactly as before. A stale (lower) snapshot is conservative — it can only route a uid down the slow path, never wrongly accept one. Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn with EXPLICIT uids, 64 threads, 2 runs per arm: ExtractBlankUIDs 3.67% -> absent from the profile verifyUid 3.52% -> absent MaxLeaseId 3.51% -> absent CPU utilisation 8.85% -> 8.6% of 64 vCPU Throughput is UNCHANGED (12.53 MB/s both arms). That is expected and worth recording: at ~9% CPU on 64 vCPU this Alpha is not CPU-bound, because processApplyCh applies one Raft entry at a time. This removes real work and real contention on shared membership state, which matters for a CPU-bound deployment, for co-located Alphas, and once cross-transaction concurrency raises utilisation — but it does not move throughput today. Only relevant when mutations carry explicit uids, which is the common case for the high-ingest workload this series targets. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Measured results for the four perf commits on an i4i.16xlarge (32 physical cores / 64 vCPU, local NVMe), 25 [uid] @reverse predicates, 20k triples per transaction, 64 writer threads, 300s runs, using badger_write_bytes_user — the same metric as the Preview1 vs Preview1-Hybrid comparison in dgraph-io#9727. Records the throughput ladder (9.63 -> 12.45 MB/s), the goroutine-parking progression that shows the bottleneck migrating from the global cache lock to the conflict-key lock and then away entirely, and the CPU utilisation that stayed under 10% of 64 vCPU throughout. Also documents the runs that were DISCARDED and why, so they are not re-run by accident: a stale Alpha holding :8080 that made both arms profile the same binary, a load generator whose file striding caused 1.47M transaction aborts, and one throughput outlier. Each of these produced plausible-looking numbers that were wrong. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Full write-up for 5b69ac2, kept because the interesting part is the decision NOT to shard. Sharding Txn.conflicts was the obvious fix and was rejected on arithmetic: a line-level profile showed ~80% of addConflictKeyWithUid's cost was lock acquire/release and only ~15% the map insert, so batching to one acquisition per Process* captured the win while sharding would have chased a ~1% residual — at the cost of changing the container type, nine test sites, FillContext, and NumShards, which is shared with the delta and index maps. Records the explicit revisit trigger so the question stays closed unless the numbers change. Also captures the three implementation traps that produce silent corruption rather than a failure: eager key expansion (call sites reuse one scratch key buffer, so a deferred fingerprint hashes the wrong uid), defer ordering in InsertTokenizerIndexes (registering the flush below cache.Lock compiles and passes every test while keeping the nested lock), and break-not-return on worker error (a shrunken conflict set risks a lost update, since Zero's hasConflict is a pure existential). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Forward-looking companion to the validation record. Three sections worth
flagging for review:
Flag simplification. The pipeline now carries four superflag knobs, two of
which are inert unless mutations-pipeline-goroutines=-1 (not the default),
and the shipped default of 30 sits exactly on the allocateWorkers cliff:
with 25 predicates it yields a {1:20, 2:5} grant. Before 612ae8c that
silently disabled the lock-free path for 20 of 25 predicates. Also notes
that WorkerOptions.String() hand-formats the struct and omits every
MutationsPipeline* field, so the startup log cannot confirm the effective
configuration — this cost real time during live debugging.
Where the CPU actually goes now. After the series the profile is no longer
dominated by posting/: RWMutex read traffic (~11%), worker.(*groupi).Tablet
(~7-9%, a read-mostly membership map behind SafeMutex hit once per
proposal), and allocation pressure (~10%). MaxLeaseId/verifyUid was in this
list until 016a03a removed it.
Harness lessons, recorded so they are not re-learned: Dgraph returns HTTP
200 with an errors body on rejection, explicit uids need a zero lease
first, cross-predicate lock contention requires multiple predicates per
TRANSACTION rather than multiple client threads, and sequential A/B on a
thermally-throttling laptop produced a spurious 43% regression on an
unchanged code path.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Deltas.AddToDeltas was the top blocker left inside the mutation pipeline — 57.6% of blocked pipeline goroutines. It is one Set into a LockedShardedMap[string, []byte] that every predicate worker writes to, and the map was fixed at NumShards = 30. That constant is sized for types.ShardedMap, an unrelated single-threaded map used by the query path, and is far too thin for 64 vCPU. Two coupled changes: 1. Shard count is now per-instance, defaulting to 4x GOMAXPROCS clamped to [64, 1024] and rounded to a power of two. The struct already held shards and locks as slices, so only the constructor was fixed at NumShards — but getShardIndex ALSO indexed off the global constant, which would over-run the slice or strand every shard above index 30 once the count varied. It now masks against the map's own size. types.ShardedMap and the shared NumShards constant are deliberately untouched. 2. String keys hashed with farm.Fingerprint64([]byte(k)), whose conversion escapes to the heap on every Get/Set — previously ~13% of all allocations in a 20k-edge batch, since every delta write hashes its key. maphash.String is allocation-free. Nothing persisted depends on shard assignment, so the hash may change freely. uint64 keys are multiplied by a 64-bit constant before masking, because dense monotonic uids would stripe poorly against a power-of-two mask. Merge now falls back to a key-by-key path when two maps have different shard counts; positional merging would silently drop keys. It has no callers today, but per-instance sizing made that a live hazard. Micro-benchmark, interleaved, median of 5 (Set with N goroutines): workers=1 499 -> 434 ns/op 3 -> 2 allocs workers=32 229 -> 171 ns/op 2 -> 1 allocs (1.34x) workers=64 212 -> 165 ns/op 2 -> 1 allocs (1.28x) End to end on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn, 64 threads, budget=30, 2 runs per arm: badger_write_bytes_user 12.44 -> 13.17 MB/s (+5.9%) triples/s 123795 -> 131449 (+6.2%) blocked on mutex 25.0% -> 10.5% runnable 48.8% -> 75.6% pipeline goroutines 160 -> 86 Cumulative over stock: 9.63 -> 13.17 MB/s (+36.8%). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…rule CLAUDE.md is gitignored in this repo, so the same summary goes here to survive in git history and reach anyone reading the branch. Records the six-commit ladder (9.63 -> 13.17 MB/s, +36.8%) and the observed bottleneck migration — global cache lock 86% -> conflict lock 71% -> AddToDeltas 58% -> nothing above ~10% — which is why the commits had to land in that order, each fix exposing the next. The load-bearing part is the rule for future work: a single Alpha sits at ~10% of 64 vCPU because processApplyCh applies one Raft entry at a time, so a change that only reduces CPU or allocations will NOT move throughput. 016a03a removed ~3.5% of CPU for exactly zero gain. Candidates must be ranked by share of BLOCKED goroutines, not CPU. Without that rule the remaining backlog items look attractive and are worthless — which is why they were skipped deliberately rather than left undone by accident. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Records the strongest gate run so far: a mixed-workload corpus (~408k triples, 17 predicate types) ingested into four fresh clusters and fingerprinted with 72 DQL queries. stock@30, new@0, new@30 and new@auto all produce IDENTICAL results with zero mutation errors, covering every apply-path branch the pipeline work touched — ProcessSingle/List/Reverse/ Count, ten index tokenizers, @upsert, @noconflict, @lang, lists, DEL and star-delete — rather than only the [uid] @reverse shape used for benchmarking. Also records, with evidence, that the worker package's -race failures are pre-existing and structural rather than caused by this work: worker.Init overwrites a package-global limiter and starts a bleed() goroutine that is never stopped, so every test's Init races the previous test's goroutine and the detector blames whichever test is running. Confirmed by running the same suite on ab0a49e — both trees fail, with fluctuating sets, and TestLimiterDeadlock fails 5/5 on both. posting passes cleanly on both. Plus the harness trap that invalidated the first attempt: /admin export is asynchronous, so sleeping and reading the export dir compares partial flushes, not databases. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Full t/ runner suite against a docker image verified by symbol inspection to contain the changes. 39 packages ok with no test or package failures, including systest/mutations-and-queries, systest/vector (HNSW), and worker — which passes cleanly here despite the pre-existing worker.Init race that makes its -race unit run flaky. Records the setup a clean box needs (Docker, gotestsum, a git root, make local-image; ack is unavailable on RHEL 9 but the runner does not need it), and two things that look like failures and are not: the four panic: entries come from the deliberate panic_catcher test, and TestRebuildTokIndex only appears to stall if badger's INFO output is truncated by grep | head — it passes in 0.06s. One honest gap: the first full-suite invocation exited 1 with no failing test and a log ending mid-posting; rerunning that package alone gave exit 0. Not explained. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Re-ran the A/B against the ext-pointer-tech-collab three-stage generator, which replicates the NiFi/dgraph4j production pipeline: 30 @reverse and 20 @upsert predicates, 1.67M seeded nodes, Zipf hot-node reverse fan-out, @upsert re-asserts and update waves, 532 transaction-sized files. Both arms restore the same seeded snapshot so only the binary differs. stock 1.475 -> shard 1.735 MB/s (+17.6%), 44,699 -> 52,357 nquads/s (+17.1%), 129.4s -> 110.5s wall clock, reproducible to ~1% across 2 runs per arm. Abort counts are identical (~1,860 both arms), which is the strongest correctness signal yet for the conflict-key batching: conflict semantics are unchanged under real @upsert contention. Records two things the synthetic corpus hid: absolute throughput is ~1.5 MB/s rather than 13 because this workload is abort-dominated (only ~283 of 532 files commit; the ceiling is Zero's conflict arbitration, not the apply path), and the relative gain is half the synthetic figure. 17.6% is the number to quote for production, not 36.8%. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Swept client threads {2,4,8,16} on the production-replica corpus. Two
findings.
Throughput more than TRIPLES as threads drop from 16 to 2 (stock 1.48 ->
5.30 MB/s, 44.7k -> 143k nquads/s); aborts fall 1,860 -> 380 and committed
files rise 283 -> 502 of 532. On this workload the Alpha is not the
constraint — Zero's conflict arbitration over @upsert re-asserts is, and
each extra client thread buys more aborted work than committed work. This
quantifies what the production client-side node-UID lock cache is worth,
and suggests an ingest tuned for high thread counts without it is losing
throughput rather than gaining it.
The apply-path gain holds at every thread count (1.12x at 2 threads through
1.18x at 16) and is therefore not an artifact of abort noise: at 2 threads,
where 502/532 files commit and aborts are minimal, it is still +12%.
Abort counts track each other across all four thread counts (380/379,
945/931, 1465/1479, 1855/1857), confirming conflict semantics are unchanged.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…tion-* Four superflag keys become three, renamed to say what they scope. All of this parallelizes a single mutation; processApplyCh is still serial across transactions, so the names now make that visible rather than leaving an operator to infer it. mutations-pipeline-threshold -> intra-mutation-min-edges mutations-pipeline-goroutines -> intra-mutation-parallelism mutations-pipeline-goroutines-fraction ^ (merged: off|auto|N|Fx) mutations-pipeline-min-edges-per-worker -> intra-mutation-edges-per-worker The two merged keys were the same axis in two notations -- a worker count and a multiple of GOMAXPROCS -- with the first doubling as the tag selecting which was read. The -1 AUTO sentinel is gone; auto is 1x. Two of the four keys used to be inert unless goroutines == -1. That is fixed by deleting the conditional, not the flags: edges-per-worker now caps every sizing mode, because "do not spin N workers for a handful of edges" is as true of a fixed count as a derived one. It stays a flag deliberately -- 256 was adopted by analogy to DivideAndRule and never measured, and it is the binding term on a large box, where a 20k-edge mutation caps at 78 workers however many cores exist. Decouple the lock-free store from the worker grant. allocateWorkers now floors every predicate at 1 instead of returning nil, so a grant of 1 means "no fan-out", not "take the legacy locked store". Tying those together made the disabled setting also give up the lock-free store -- about two thirds of this branch's measured gain -- and made the flag non-monotonic, since a one-worker grant dropped index tokenization from 10 goroutines to 1. numGo is now max(10, grant). intra-mutation-min-edges=0 remains the single kill switch. Also adds the observability that was missing: the three fields now appear in WorkerOptions.String() (the startup log could not previously confirm whether the budget was 30 or 0), a V(2) line reports the resolved grant and which term bound it, and intra_mutation_no_fanout_total counts the silent-degradation case. Measured on an i4i.16xlarge against the reverse-heavy corpus this branch was optimized for: 132,730 vs 133,026 triples/s at a fixed budget (-0.22%) and 132,664 vs 132,513 at auto (+0.11%) -- both within run-to-run noise. TestSchemaMatrixByteIdentical now compares against the REAL legacy path (a serial runMutation loop) rather than the pipeline with its budget off, which the store decoupling removed. That is strictly stronger, and it surfaced three pipeline-vs-legacy divergences that predate this work; they are measured on the pre-change tree and recorded in pipeline-todo.md rather than pinned here. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
pipeline-todo.md §1 is closed, with the reasoning that mattered: the fix for the inert flags was removing the conditional rather than the flags, and edges-per-worker stayed tunable because it is unmeasured and is what actually binds on a large box. Two findings the trace produced that were not in the original write-up: a worker count of 1 was silently identical to 0 while 2 gave every predicate a lock-free grant, and the AUTO fraction saturated -- making the help text's "raise it to oversubscribe to 2-3x cores" advice unfollowable at the batch size it was written for. Both are now pinned by tests. New §1c records three pipeline-vs-legacy divergences surfaced by rebasing the byte-identity matrix onto the real legacy path. All three reproduce on the pre-change tree, so they are unrelated to this work. The @lang case deserves its own investigation: the pipeline drops 600 conflict keys the legacy path emits, which is the direction that risks a lost update, and the pipeline has been the production default all along. ec2-validation-results.md keeps its tables verbatim -- they describe runs that actually happened under the old keys -- with a translation table for reproducing them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.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
1 participant
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.
What
Three small, additive, flag-gated changes on top of this PR's per-predicate mutation pipeline, so a single hot/dominant predicate's apply actually parallelizes (today the pipeline fans one goroutine per predicate, so a single dominant predicate gets ~no intra-predicate speedup).
mutations-pipeline-goroutines=Ndistributes N workers across the predicates of a batch by edge count (largest-remainder), so a dominant predicate gets >1 worker.-1) — budget derived at runtime:min(GOMAXPROCS * fraction, totalEdges / minEdgesPerWorker). Tunables:mutations-pipeline-goroutines-fraction(default 1.0) andmutations-pipeline-min-edges-per-worker(default 256).ProcessList— the merge-light one-to-one<pred,uid>pass, mirroring the existingProcessSinglesplit, so[uid]/@reversepredicates' forward write parallelizes.ProcessReversestays serial (many-to-one, hot-target-prone — parallelizing it is a proven net loss).Why
Profiling a reverse-heavy ingest workload (~50–70%
@reverse [uid]) showed the per-predicate pipeline gives a dominant scalar predicate a good speedup but leaves@reverse [uid]predicates at 1.00× — they route toProcessList, whose forward write was serial. Change #3 closes that gap; #1/#2 make the worker count adaptive instead of one-per-predicate.How to configure
All keys live under the existing
--feature-flagssuperflag (semicolon-separatedkey=valuepairs). Two keys work together — the new intra-predicate parallelism only kicks in when both are set:mutations-pipeline-threshold1>0andlen(edges) >= threshold.0= legacy path entirely (the budget below is then irrelevant).mutations-pipeline-goroutines300or1= one goroutine per predicate (no intra-predicate split).N>1= fixed budget.-1= auto (derive from the two keys below).mutations-pipeline-goroutines-fraction1.0round(GOMAXPROCS * fraction), capped by the next key.mutations-pipeline-min-edges-per-worker256totalEdges / thisso small batches don't over-spawn.Example: alpha with a fixed 64-goroutine budget
dgraph alpha \ --my=localhost:7080 \ --zero=localhost:5080 \ --feature-flags "mutations-pipeline-threshold=1; mutations-pipeline-goroutines=64"Example: alpha with the runtime auto budget (recommended on high-core hardware)
dgraph alpha \ --my=localhost:7080 \ --zero=localhost:5080 \ --feature-flags "mutations-pipeline-threshold=1; mutations-pipeline-goroutines=-1; mutations-pipeline-goroutines-fraction=1.0; mutations-pipeline-min-edges-per-worker=256"Auto sizes the budget to the host at runtime: for a host with
Hcores and a ~20k-edge batch it grantsmin(round(H * fraction), 20000/256)workers, distributed across the batch's predicates by edge count.Disable (legacy one-goroutine-per-predicate)
dgraph alpha --my=localhost:7080 --zero=localhost:5080 \ --feature-flags "mutations-pipeline-goroutines=0"Correctness
mutations-pipeline-goroutines=0(or any value<2) ⇒ byte-identical to the current one-goroutine-per-predicate path.-race) for: scalar predicates,[uid] @reversepredicates, and AUTO-budget == fixed-budget.ProcessReverse,allocateWorkers, and the auto derivation are pure/serial where it matters; per-worker scratch key buffers; lock-free store only for provably-disjoint<pred,uid>keys; all workers join before reverse/index/count passes (MVCC order preserved).TestAllocateWorkers,TestAutoBudget,TestPipelineBudgetByteIdentical,TestProcessListBudgetByteIdentical,TestPipelineBudgetAutoByteIdentical.Benchmarks (8-core dev box — a floor; the win scales with cores)
~20k-edge batch, 5 hot reverse targets (modeling 2–5 hot nodes).
edges/s, speedup vsbudget=0:[uid] @reversedominant@indexdominant (control)The
@reversecase was 1.00× before change #3. The still-serialProcessReversecaps it below the scalar ceiling, as expected. Numbers are a lower bound (8 cores);BenchmarkReverseDominant/BenchmarkReverseFiftyFiftyare included so this can be re-swept on production-class hardware to tunefraction.Notes
fractionlikely differs —autocurrently trails the best fixed budget on a small box. Recommend a sweep on real hardware before defaulting toauto.🤖 Generated with Claude Code