perf(server-ng): reach single-node latency parity with legacy server - #3746
perf(server-ng): reach single-node latency parity with legacy server#3746hubcio wants to merge 1 commit into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #3746 +/- ##
=============================================
- Coverage 74.54% 61.29% -13.26%
Complexity 969 969
=============================================
Files 1306 1305 -1
Lines 150028 139529 -10499
Branches 125462 115037 -10425
=============================================
- Hits 111840 85521 -26319
- Misses 34686 50371 +15685
- Partials 3502 3637 +135
🚀 New features to boost your workflow:
|
server-ng trailed the legacy server on single-node benchmarks: the default cpu allocation collapsed all work onto one shard, produce hashed each batch three times (convert, stamp, flush revalidation), and a lone consumer crossing a sealed segment degraded to full-segment scans through unbounded read handles. Shard allocation now defaults to numa:auto. Produce computes the batch checksum once and marks locally originated batches trusted; replicated blobs keep a per-message receive gate on followers, so transit integrity still holds end to end. Sealed segment read handles are capped by a per-partition LRU. The batch checksum is redefined to cover the six header meta fields plus each message's stored checksum field instead of the whole blob, binding bodies transitively through the per-message checksums that every validating decode re-verifies. Stamping a batch now hashes N*8 bytes instead of the full blob, removing the last full-blob pass from the produce path. Message data written by earlier server-ng builds fails checksum validation after this change; wipe local_data when upgrading. Per-message checksums are computed in a single oneshot pass over the encoded record, pinned by a streaming-vs-oneshot reference test.
7bf57b2 to
3a944be
Compare
| fn calculate_batch_checksum(header: &SendMessages2Header, blob: &[u8]) -> u64 { | ||
| let mut hasher = XxHash3_64::new(); | ||
| hasher.write(header_tail); | ||
| hasher.write(payload); | ||
| hasher.write(user_headers); | ||
| write_batch_header_fields(&mut hasher, header); | ||
| let batch = SendMessages2Ref { | ||
| header: *header, | ||
| blob, | ||
| }; | ||
| for framed in batch.iter_with_offsets() { | ||
| hasher.write(&blob[framed.start..framed.start + 8]); | ||
| } |
There was a problem hiding this comment.
Trailing bytes past batch_length now reach disk and permanently desync the segment walk
Changed lines: core/server_common/src/send_messages2.rs:923-932 and core/partitions/src/iggy_partition.rs:1965. Both gates that used to catch this were removed by this PR; the unchanged lines they protected are send_messages2.rs:807 and :677.
Two pre-existing behaviours make an oversized request representable. decode_batch_slice:677 tests body.len() < header.total_size(), a lower bound rather than an equality, and then clamps the blob to blob_len, so any suffix past COMMAND_HEADER_SIZE + blob_len is left uncovered. stamp_prepare_for_persistence:807 slices bytes[PREPARE_SPLIT_POINT..total_size] off PrepareHeader.size rather than off batch_length. Neither line is touched by this PR.
Reachability: core/server-ng/src/dispatch.rs:404 sets total_size = header_size + body.len(), deriving size from the bytes actually received. A client that appends K junk bytes after a well-formed batch therefore produces size = 512 + blob_len + K with a correct batch_length, and nothing cross-checks the two.
What changed is the outcome. Before this PR, calculate_batch_checksum was hasher.write(blob), covering every byte of the slice handed to it, so the stamp hashed blob plus K while the validating flush walk at iggy_partition.rs:1965 recomputed over blob[..blob_len] without K. The mismatch fired, the batch was never appended, and the error was logged. The corruption was caught before disk.
This PR removes both gates. :923-932 replaces the bulk write with a frame walk that stops after the last decodable frame, so the stamp and the read-back recompute now produce an identical value and K becomes invisible. :1965 downgrades the flush walk to decode_prepare_slice_trusted, which performs neither the checksum comparison nor the tiling check, so nothing rejects prepare.size - 512 != blob_len.
The consequence is durable. The flush writes entry.slice(256..) (core/partitions/src/iggy_partition.rs:2568), which is 256 + blob_len + K bytes, while flush accounting advances by batch.header.total_size() (:1999), which is 256 + blob_len. walk_disk_chunk then advances cursor += total_size (core/partitions/src/poll_plan.rs:863), lands inside the junk, fails to decode, and every later batch in that segment becomes unreadable. Backups write the same bytes, because the receive gate truncates the blob to blob_len before verifying.
Scope: the silent case is junk that fails frame decode, which covers any K below 48 bytes, and any K at or above 48 whose bytes do not satisfy reserved == 0 plus self-consistent length fields. Junk that does decode as a frame is absorbed into the stamp and then mismatches on read-back, so it still fails loudly. Reaching any of this requires a non-conforming client; the official SDKs will not emit it.
The fix is one line: bound the stamp blob by command.blob_len()?, or tighten :677 to an equality.
| /// # Errors | ||
| /// | ||
| /// Returns an error if the file metadata or bytes cannot be read. | ||
| pub async fn load_all(&self) -> Result<IggyIndexCache, IggyError> { |
There was a problem hiding this comment.
load_all() reads and materializes an entire .index file on the poll hot path with no bound
core/partitions/src/iggy_index_reader.rs:126
Index density is one sparse entry per flush, governed by messages_required_to_save and size_of_messages_required_to_save (core/server-ng/config.toml:515,520). At defaults a 1 GiB segment yields about 1024 entries, a 24 KiB read, and the "index files are tiny" comment holds. Set messages_required_to_save = 1, which is what in-tree repair_config() uses, with 100-byte messages, and the same segment yields 7.25M entries: a single 174 MB read_exact_at issued from inside a poll, which compio will punt to io-wq. That is the exact stall this PR set out to remove, now reachable by supported config. There is no cap, no chunking, and no fallback.
This also reintroduces up to 12 resident index caches per partition, which is the thing core/partitions/src/iggy_partition.rs:2731-2736 deliberately drops at rotation, with a comment naming retained hundreds of MB as the reason.
Since IggyIndexCache is sorted and only ever binary-searched (core/partitions/src/iggy_index.rs:91,105), the bounded alternative is roughly 20 24-byte read_entry_at preads above a byte budget; the reader already has that primitive at :67.
| return self.open_segment_with_retry(path).await; | ||
| }; | ||
| // Borrow only to clone the `Option<File>` out, never across the await. | ||
| if let Some(cached) = handle.fd.borrow().clone() { |
There was a problem hiding this comment.
A cached fd can serve messages from an explicitly purged topic. core/partitions/src/poll_plan.rs:557. A poll plan clones a sealed handle whose fd slot is already populated; the pump then processes PurgeTopic into purge() (core/partitions/src/iggy_partition.rs:2985), which retires every segment, unlinks the files, and recreates 00000000000000000000.log. The in-flight Rc keeps the old inode alive, and resolve_segment_file returns the cached fd without touching the filesystem, so the walk serves purged messages. The window widens on a multi-segment walk suspended in read_chunk_with_retry. The handle doc at :31-33 blesses reading the unlinked inode, which is correct for time and size retention but not for a user-issued purge. Before this PR the re-open hit the freshly created empty file and failed closed.
|
@krishvishal tyvm for review, i think i will split this PR into 2 smaller PRs to make it more bearable for review. |
2 participants
server-ng trailed the legacy server on single-node benchmarks:
the default cpu allocation collapsed all work onto one shard,
produce hashed each batch three times (convert, stamp, flush
revalidation), and a lone consumer crossing a sealed segment
degraded to full-segment scans through unbounded read handles.
Shard allocation now defaults to numa:auto. Produce computes
the batch checksum once and marks locally originated batches
trusted; replicated blobs keep a per-message receive gate on
followers, so transit integrity still holds end to end. Sealed
segment read handles are capped by a per-partition LRU.
The batch checksum is redefined to cover the six header meta
fields plus each message's stored checksum field instead of the
whole blob, binding bodies transitively through the per-message
checksums that every validating decode re-verifies. Stamping a
batch now hashes N*8 bytes instead of the full blob, removing
the last full-blob pass from the produce path. Message data
written by earlier server-ng builds fails checksum validation
after this change; wipe local_data when upgrading. Per-message
checksums are computed in a single oneshot pass over the encoded
record, pinned by a streaming-vs-oneshot reference test.