Repository navigation
Conversation
…a shared pane store - Ingestion: N workers (`--ingestion-workers`, default one per core). Remote Write series are sharded by a hash of their label set; each worker owns its shard's open panes. The global watermark is the minimum over the workers that have seen a sample; when it passes a pane, every worker builds its part, and the last part completes the pane: parts are joined and the states of a group several workers saw are merged with the family's merge. A request is queued to every worker or refused whole. - Queries: a pool of `--query-threads` threads, one query (and every step of a range query) per thread. The executor's compiled plans hold `Rc`s, so each thread reads plan.json and compiles its own. - SDS: `RwLock<HashMap<(definition, pane end), Arc<pane>>>`. Panes are inserted once and never rewritten; readers clone the `Arc`. Kept query-time results use the same store, shared by the query threads. A stored pane binds to a root as a shared source, copied once per run (the executor's sources emit owned batches) instead of twice. - jemalloc is the data plane's global allocator.
…he watermark it announces Two races of the sharded engine, both found by a test that writes a pane's samples and queries it at once: - A planned query ran on the query pool while the ingestion workers could still be ingesting the Remote Write requests answered before it, so the newest pane was missing and the query was forwarded. A query now waits until every worker has run what was queued before it (`sync`), as the single engine thread's queue ordered it. - A worker closed its panes at the watermark it read before another worker moved it, then announced the newer watermark to the other workers but not to itself, so its part of the newest pane never arrived. It now closes and announces the same watermark.
zzylol
added
enhancement
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.
Stack 2/4. Base: #828.
Why
One engine thread ran ingestion, pane builds and every query, because the Planner's DAG types hold
Rcs. A long query held up ingestion and every other query behind it. Each evaluation also deep-copied the stored panes it read. User decisions (design doc, Stage D and the threading section): a multi-threaded precompute engine as in ASAPQuery, one query per thread, a pane store of immutableArcpanes under a read-write lock, and jemalloc as the default allocator.What
--ingestion-workers, default one per core). Remote Write series are sharded by a hash of their label set. Each worker owns its shard's open panes.--query-threads, default one per core). A query, with every step of a range query, runs on one pool thread. Before reading, a query waits until the workers have run every write answered before it, so it sees those writes, as the single queue guaranteed.RwLock<HashMap<(definition, pane end), Arc<pane>>>. A pane is inserted once and never rewritten, and readers clone theArcunder the read lock. Eviction takes the write lock. Kept query-time results use the same store, shared by all query threads, and are inserted only once their pane is closed.Arc. It is copied once per run, because the executor's sources emit owned batches; before, it was copied twice.tikv-jemallocator) is the data plane's global allocator.How
CompiledPhysicalDAG,RawSampleSelector) are notSend:CompiledExpressionholdsRc<OperatorNode>. So each worker and query thread reads plan.json and compiles its own.BatchandValueareSend + Sync(their summary states areArc<dyn AggregateCore + Send + Sync>), so panes cross threads asArc<Batch>.precompute::Sharedholds what threads share: the store, per-worker watermarks, the partial panes and the counters.ShardWorkeris one worker's compiled state.Precomputeruns all workers on the caller's thread, for tests.Before this PR
One
enginethread. A planned query waited behind ingestion and behind every other query. Each evaluation cloned every stored pane it read. glibc malloc.After this PR
Ingestion runs on N workers and queries on M threads; queries read panes concurrently by reference.
Verification
cargo test --workspacepasses, includinge2e_examples(9),range,kept,answers.sds::store_shares_panes_and_never_rewrites_themandsds::store_is_read_concurrently;precompute::a_pane_closes_once_every_shard_has_passed_it(minimum watermark over 2 shards);precompute::sharded_panes_answer_like_one_shard(4 shards against 1: Example 4b's per-series KLL; Example 1's rates and sums);answers::answers_concurrent_queries_like_one_at_a_time(32 concurrent queries);answers::a_query_reads_the_panes_of_the_writes_before_it. It caught two races in this PR's first commit, fixed in its second: a query racing the workers, and a worker announcing a newer watermark than it had closed. It fails 3 of 3 runs without the fix and 0 of 8 with it.Benefit harness
Same setup as #828. "Prometheus alone" is the baseline of the pr1 run. The harness issues one query at a time, so it measures per-query cost and ingestion contention, not concurrent query throughput.
s10
s10-noraw
s50
s50-noraw
jemalloc vs glibc. "pr2-glibc" is this PR built without the global allocator line (a local build, not committed); everything else is identical.
s10
s10-noraw
s50
s50-noraw
jemalloc lowers data-plane CPU in every config (s50: 174.0 vs 247.1 core-s with raw data, 145.3 vs 179.2 without) and peak RSS (s50: 1.73 vs 2.71 GiB). The quantile queries are faster with jemalloc in three of four configs (p99[60m] p50 at s50: 3853 vs 5192 ms). The one regression is p99[5m] at s50 without raw data: 308.2 vs 292.4 ms p50, 389.7 vs 347.2 ms p99. The earlier in-process comparison's +9 ms on the quantile queries was single-threaded; here the allocator is shared by 8 engine threads.
Planner gaps found
Send.CompiledExpressionholdsRc<OperatorNode>throughScalarExproperator references, soCompiledPhysicalDAGandRawSampleSelectorare neitherSendnorSync, and every thread compiles its own. Smallest fix: compile expressions to anRc-free form (operator references resolved to node ids), which makes compiled plans shareable.Operator::sourcetakes ownedBatches and clones each on emit, and output streams yield ownedBatch(Vec<Vec<Value>>). So a stored pane is copied into every evaluation that reads it. Smallest fix: a source overArc<Batch>yielding the shared value, orBatchrows behind anArc.