This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/server-ng-sealed-read-lru in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b9b41fb57e28b9c0141190bf97beac1f0cdef891 Author: Hubert Gruszecki <[email protected]> AuthorDate: Mon Aug 3 10:52:40 2026 +0200 perf(server-ng): cap sealed-segment read handles with per-partition LRU A lone consumer crossing a sealed segment degraded to full-segment scans: rotation drops the writer and the resident index, so every poll re-opened the file and scanned from byte 0. Sealed segments now share a per-partition read-state handle caching the read fd and the sparse index reloaded from the .index file, capped by an LRU so descriptors stay bounded. Shard allocation defaults back to numa:auto now that multi-shard server-ng is stable; the previous single-shard default collapsed all work onto one core. --- core/partitions/src/iggy_index_reader.rs | 35 +++- core/partitions/src/iggy_partition.rs | 285 +++++++++++++++++++++++++++-- core/partitions/src/iggy_partitions.rs | 15 +- core/partitions/src/log.rs | 295 ++++++++++++++++++++++++++++++- core/partitions/src/poll_plan.rs | 175 +++++++++++++++++- core/server-ng/config.toml | 3 +- core/server-ng/src/dispatch.rs | 8 +- 7 files changed, 775 insertions(+), 41 deletions(-) diff --git a/core/partitions/src/iggy_index_reader.rs b/core/partitions/src/iggy_index_reader.rs index a105a6982..cd618c19f 100644 --- a/core/partitions/src/iggy_index_reader.rs +++ b/core/partitions/src/iggy_index_reader.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndex}; +use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndex, IggyIndexCache}; use bytes::Buf; use compio::fs::{File, OpenOptions}; use compio::io::AsyncReadAtExt; @@ -114,4 +114,37 @@ impl IggyIndexReader { .await?, )) } + + /// Load every whole entry into an [`IggyIndexCache`] for offset / timestamp + /// lower-bound lookups. Reads the whole file in one pass (index files are + /// tiny: one sparse entry per flushed chunk). A trailing partial entry + /// (torn write) is ignored (see [`Self::entry_count`]). + /// + /// # Errors + /// + /// Returns an error if the file metadata or bytes cannot be read. + pub async fn load_all(&self) -> Result<IggyIndexCache, IggyError> { + let count = usize::try_from(self.entry_count().await?) + .map_err(|_| IggyError::CannotReadFileMetadata)?; + if count == 0 { + return Ok(IggyIndexCache::empty()); + } + + // `with_capacity` (len 0): `read_exact_at` fills the spare capacity in + // place and advances the length (see `read_entry_at`). + let buffer = Vec::with_capacity(count * IGGY_INDEX_SIZE); + let (result, buffer): (std::io::Result<()>, Vec<u8>) = + self.file.read_exact_at(buffer, 0).await.into(); + result.map_err(|_| IggyError::CannotReadFile)?; + + let mut cache = IggyIndexCache::with_capacity(count); + let mut view = buffer.as_slice(); + for _ in 0..count { + let offset = view.get_u64_le(); + let timestamp = view.get_u64_le(); + let position = view.get_u64_le(); + cache.insert(offset, timestamp, position); + } + Ok(cache) + } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index eb48863c5..eab240191 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -723,8 +723,9 @@ where /// can run the disk read + offset persist off the partition borrow. The /// in-memory journal tier is read here directly (mem reads never yield); /// the disk tier is captured as owned descriptors in [`DiskReadPlan`]. + #[allow(clippy::too_many_lines)] pub(crate) fn build_poll_plan( - &self, + &mut self, consumer: PollingConsumer, args: &PollingArgs, ) -> PollPlan { @@ -815,13 +816,22 @@ where } let (start_segment, start_position) = self.disk_poll_start(&query); + // Cap resident sealed read handles: touch this poll's start segment so + // the LRU keeps the hot set and drops the least-recently-used fd + + // index (a no-op for the active segment, whose handle never caches). + self.log.touch_sealed_read_state(start_segment); // Snapshot only the segments the disk walk visits (`start_segment..`), - // so `start_position` applies to the first snapshotted segment. + // so `start_position` applies to the first snapshotted segment. A sealed + // segment carries its shared read-state handle (fd + sparse index) so + // the off-borrow read reuses (or fills) it; the active segment opens + // fresh and resolves from its resident index. let segments = self.log.segments()[start_segment..] .iter() - .map(|segment| DiskSegment { + .zip(self.log.sealed_read_state()[start_segment..].iter()) + .map(|(segment, read_state)| DiskSegment { start_offset: segment.start_offset, persisted: segment.size.as_bytes_u64(), + read_state: segment.sealed.then(|| Rc::clone(read_state)), }) .collect(); let disk = DiskReadPlan { @@ -2808,12 +2818,10 @@ where let mut deleted_messages = 0u64; for _ in 0..removable { // The removable run is always a prefix (oldest first), so the next - // victim is index 0 once the previous one is gone. - let segment = self.log.segments_mut().remove(0); - let mut storage = self.log.storages_mut().remove(0); - self.log.indexes_mut().remove(0); - self.log.messages_writers_mut().remove(0); - self.log.index_writers_mut().remove(0); + // victim is the front once the previous one is gone. + let Some((segment, mut storage)) = self.log.retire_front() else { + break; + }; let (messages_path, index_path) = storage.segment_and_index_paths(); let _ = storage.shutdown(); @@ -2967,11 +2975,9 @@ where // Drain every segment (including the active one) and unlink its files. let segment_count = self.log.segments().len(); for _ in 0..segment_count { - self.log.segments_mut().remove(0); - let mut storage = self.log.storages_mut().remove(0); - self.log.indexes_mut().remove(0); - self.log.messages_writers_mut().remove(0); - self.log.index_writers_mut().remove(0); + let Some((_, mut storage)) = self.log.retire_front() else { + break; + }; let (messages_path, index_path) = storage.segment_and_index_paths(); let _ = storage.shutdown(); @@ -3453,7 +3459,7 @@ fn nth_oldest_sealed_end(segments: &[Segment], count: u32) -> Option<u64> { #[cfg(test)] mod tests { use super::*; - use crate::poll_plan::DiskReadOutcome; + use crate::poll_plan::{DiskReadOutcome, SealedSegmentHandle}; use bytes::Bytes; use compio::io::AsyncWriteAtExt; use consensus::LocalPipeline; @@ -3909,10 +3915,12 @@ mod tests { DiskSegment { start_offset: 0, persisted: 512, + read_state: None, }, DiskSegment { start_offset: 5, persisted: later_len, + read_state: None, }, ], start_position: 0, @@ -3992,10 +4000,12 @@ mod tests { DiskSegment { start_offset: 0, persisted: corrupt_len, + read_state: None, }, DiskSegment { start_offset: 5, persisted: later_len, + read_state: None, }, ], start_position: 0, @@ -4028,6 +4038,7 @@ mod tests { segments: vec![DiskSegment { start_offset: 0, persisted: 512, + read_state: None, }], start_position: 0, namespace_raw: IggyNamespace::new(1, 1, 0).inner(), @@ -4057,6 +4068,7 @@ mod tests { segments: vec![DiskSegment { start_offset: 0, persisted: 512, + read_state: None, }], start_position: 0, namespace_raw: IggyNamespace::new(1, 1, 0).inner(), @@ -4076,6 +4088,249 @@ mod tests { ); } + /// A sealed-segment poll opens the file once and caches the read fd; a later + /// poll of the same segment reuses the cached descriptor. Proven by + /// unlinking the file after the first read: a fresh open-by-path would now + /// fail, so a successful second read can only come from the cached fd (which + /// reads the still-open, unlinked inode). + #[compio::test] + async fn read_disk_caches_and_reuses_sealed_segment_fd() { + let namespace = IggyNamespace::new(1, 1, 0); + + let dir = std::env::temp_dir().join(format!( + "iggy-read-disk-fdcache-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + let partition_dir = dir.to_string_lossy().into_owned(); + + let record = build_segment_record(namespace, 0); + let record_len = record.len() as u64; + let path = format!("{partition_dir}/{:0>20}.log", 0u64); + { + let mut file = compio::fs::File::create(&path) + .await + .expect("create segment file"); + let (written, _) = file.write_all_at(record, 0).await.into(); + written.expect("write segment record"); + file.sync_all().await.expect("flush segment file"); + } + + let handle = SealedSegmentHandle::default(); + // The pump touches the poll's start segment before cloning its handle + // into the plan, so a cache-eligible handle is always tracked. + handle.tracked.set(true); + assert!(handle.fd.borrow().is_none(), "fd cache slot starts empty"); + + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: record_len, + read_state: Some(Rc::clone(&handle)), + }], + start_position: 0, + namespace_raw: namespace.inner(), + }; + let first = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(first, DiskReadOutcome::Matched { .. }), + "first sealed poll must match the batch", + ); + assert!( + handle.fd.borrow().is_some(), + "first sealed poll must populate the read-fd cache slot", + ); + + // Unlink the file: a fresh open-by-path would fail now, so the second + // read succeeding proves the cached fd was reused. + std::fs::remove_file(&path).expect("unlink segment file"); + + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: record_len, + read_state: Some(Rc::clone(&handle)), + }], + start_position: 0, + namespace_raw: namespace.inner(), + }; + let second = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(second, DiskReadOutcome::Matched { .. }), + "cached fd must serve the read after the segment path is unlinked", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// An untracked handle (a sealed segment the walk crosses without being the + /// poll's start segment, or a slot evicted mid-poll) opens its file + /// transiently: the read succeeds but no fd is retained, so the sealed LRU + /// cap stays a true bound on resident descriptors. + #[compio::test] + async fn read_disk_does_not_retain_fd_for_untracked_handle() { + let namespace = IggyNamespace::new(1, 1, 0); + + let dir = std::env::temp_dir().join(format!( + "iggy-read-disk-untracked-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + let partition_dir = dir.to_string_lossy().into_owned(); + + let record = build_segment_record(namespace, 0); + let record_len = record.len() as u64; + let path = format!("{partition_dir}/{:0>20}.log", 0u64); + { + let mut file = compio::fs::File::create(&path) + .await + .expect("create segment file"); + let (written, _) = file.write_all_at(record, 0).await.into(); + written.expect("write segment record"); + file.sync_all().await.expect("flush segment file"); + } + + let handle = SealedSegmentHandle::default(); + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: record_len, + read_state: Some(Rc::clone(&handle)), + }], + start_position: 0, + namespace_raw: namespace.inner(), + }; + let outcome = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(outcome, DiskReadOutcome::Matched { .. }), + "the transient open must still serve the read", + ); + assert!( + handle.fd.borrow().is_none(), + "an untracked handle must not retain the fd", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A sealed-segment poll reloads the dropped sparse index from the `.index` + /// file and resolves the start byte from it, skipping the full-segment scan. + /// Proven by prefixing the `.log` with bytes a scan from position 0 would + /// fault on: only an index that jumps straight to the batch reads it. + #[compio::test] + async fn read_disk_reloads_sealed_index_to_skip_scan() { + let namespace = IggyNamespace::new(1, 1, 0); + + let dir = std::env::temp_dir().join(format!( + "iggy-read-disk-idxreload-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + let partition_dir = dir.to_string_lossy().into_owned(); + + // `.log`: an undecodable prefix (a scan from byte 0 faults on it) then a + // valid batch at offset 5. `.index`: one sparse entry mapping offset 5 + // to the batch's byte position, so the poll jumps past the prefix. + let prefix = vec![0xABu8; 512]; + let prefix_len = prefix.len() as u64; + let batch = build_segment_record(namespace, 5); + let mut log_bytes = prefix; + log_bytes.extend_from_slice(&batch); + let log_len = log_bytes.len() as u64; + let log_path = format!("{partition_dir}/{:0>20}.log", 0u64); + { + let mut file = compio::fs::File::create(&log_path) + .await + .expect("create segment log"); + let (written, _) = file.write_all_at(log_bytes, 0).await.into(); + written.expect("write segment log"); + file.sync_all().await.expect("flush segment log"); + } + + let index_bytes = crate::iggy_index::IggyIndexCache::serialize( + &crate::iggy_index::IggyIndex::new(5, 0, prefix_len), + ); + let index_path = format!("{partition_dir}/{:0>20}.index", 0u64); + { + let mut file = compio::fs::File::create(&index_path) + .await + .expect("create segment index"); + let (written, _) = file.write_all_at(index_bytes, 0).await.into(); + written.expect("write segment index"); + file.sync_all().await.expect("flush segment index"); + } + + let handle = SealedSegmentHandle::default(); + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: log_len, + read_state: Some(Rc::clone(&handle)), + }], + // Byte 0, exactly what disk_poll_start returns for a sealed segment + // whose resident index was dropped. + start_position: 0, + namespace_raw: namespace.inner(), + }; + let outcome = plan + .read_disk(MessageLookup::Offset { + offset: 5, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(outcome, DiskReadOutcome::Matched { .. }), + "the reloaded sparse index must skip the prefix; a scan from byte 0 would fault", + ); + assert!( + handle.index.borrow().is_some(), + "the sealed poll must cache the reloaded sparse index", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + fn repair_config() -> PartitionsConfig { PartitionsConfig { messages_required_to_save: 1, diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index 202799a60..c16b14843 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -391,9 +391,10 @@ where } /// Build an owned [`PollPlan`] for a partition poll synchronously, under a - /// single [`Self::with_partition`] borrow (the in-memory journal tier + the - /// resident-tail straddle snapshot are read here; mem reads never yield). - /// Returns `None` for a missing or tombstoned namespace. + /// single pump-only `&mut` borrow (the in-memory journal tier + the + /// resident-tail straddle snapshot are read here, and the sealed-read-handle + /// LRU is touched; mem reads never yield). Returns `None` for a missing or + /// tombstoned namespace. /// /// Pairs with [`PollPlan::execute`], which runs the disk read + /// offset persist/apply off the borrow on the owned plan. Splitting the @@ -406,9 +407,11 @@ where consumer: PollingConsumer, args: &PollingArgs, ) -> Option<PollPlan> { - self.with_partition(namespace, |partition| { - partition.build_poll_plan(consumer, args) - }) + // `build_poll_plan` touches the partition's sealed-read-handle LRU, so it + // needs `&mut`. Sound on the pump: it is fully synchronous (no `.await` + // inside), so no sibling task can realloc the partitions vec under it. + let partition = self.get_mut_by_ns(namespace)?; + Some(partition.build_poll_plan(consumer, args)) } /// Read a consumer's stored offset + the partition commit offset. Fully diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs index fe53992d2..baef25a11 100644 --- a/core/partitions/src/log.rs +++ b/core/partitions/src/log.rs @@ -18,11 +18,13 @@ use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndexCache}; use crate::iggy_index_writer::IggyIndexWriter; use crate::messages_writer::MessagesWriter; +use crate::poll_plan::SealedSegmentHandle; use crate::segment::Segment; use iggy_common::{IggyByteSize, IggyMessagesBatch}; use journal::{Journal, Storage}; use ringbuffer::AllocRingBuffer; use server_common::{IggyMessagesBatchSetInFlight, SegmentStorage}; +use std::collections::VecDeque; use std::fmt::Debug; use std::rc::Rc; @@ -30,6 +32,14 @@ const SEGMENTS_CAPACITY: usize = 1024; const ACCESS_MAP_CAPACITY: usize = 8; const SIZE_16MB: usize = 16 * 1024 * 1024; +/// Max sealed segments per partition that keep a resident read handle (fd + +/// sparse index). Without a cap every sealed segment a reader ever touched pins +/// one fd for the partition's lifetime; the server-wide budget is this cap times +/// the partition count, so keep it small. 12 covers a lagging consumer's working +/// set (the recent sealed segments it re-reads) with room for a few concurrent +/// readers before an LRU eviction forces a re-open. +const SEALED_READ_STATE_CAP: usize = 12; + /// Tracking metadata for the journal's current state. /// /// Replaces the server journal's `Inner` struct — lives in the `SegmentedLog` @@ -136,6 +146,16 @@ where storage: Vec<SegmentStorage>, messages_writers: Vec<Option<Rc<MessagesWriter>>>, index_writers: Vec<Option<Rc<IggyIndexWriter>>>, + // Parallel to `segments`: a shared read-state handle (fd + sparse index) + // per segment, filled lazily on the first sealed-segment poll and cloned + // into the off-borrow poll plan. Maintained in lockstep with `segments` + // (push/remove together). + sealed_read_state: Vec<SealedSegmentHandle>, + // LRU of sealed-segment `start_offset`s (most-recently-used at the front) + // bounding how many `sealed_read_state` handles stay resident, capped at + // `SEALED_READ_STATE_CAP`. Keyed by offset (stable), not slot index (which + // shifts on retire). See `touch_sealed_read_state`. + sealed_lru: VecDeque<u64>, in_flight: IggyMessagesBatchSetInFlight, } @@ -155,6 +175,8 @@ where indexes: Vec::with_capacity(SEGMENTS_CAPACITY), messages_writers: Vec::with_capacity(SEGMENTS_CAPACITY), index_writers: Vec::with_capacity(SEGMENTS_CAPACITY), + sealed_read_state: Vec::with_capacity(SEGMENTS_CAPACITY), + sealed_lru: VecDeque::with_capacity(SEALED_READ_STATE_CAP + 1), in_flight: IggyMessagesBatchSetInFlight::default(), } } @@ -173,11 +195,99 @@ where &self.segments } - pub const fn segments_mut(&mut self) -> &mut Vec<Segment> { + /// Mutable segment views. Length mutation lives in + /// [`Self::add_persisted_segment`] / [`Self::retire_front`] only, so the + /// parallel vecs cannot desync from the outside. + pub fn segments_mut(&mut self) -> &mut [Segment] { &mut self.segments } - pub const fn storages_mut(&mut self) -> &mut Vec<SegmentStorage> { + /// Shared read-state handles, parallel to [`Self::segments`]. Cloned into + /// the poll plan for sealed segments (see [`SealedSegmentHandle`]). + pub fn sealed_read_state(&self) -> &[SealedSegmentHandle] { + &self.sealed_read_state + } + + /// Record a sealed-segment access and enforce [`SEALED_READ_STATE_CAP`] + /// (LRU). `slot` indexes [`Self::segments`]; an out-of-range slot or an + /// unsealed (active) segment is a no-op, so the poll path passes its start + /// segment unconditionally. The LRU is keyed by `start_offset` - stable + /// across retire, unlike the slot index. The touched segment moves to the + /// most-recently-used front and its handle is marked tracked (eligible to + /// cache a read fd, see `SealedSegmentReadState::tracked`); once more than + /// the cap distinct sealed segments are tracked, the least-recently-used + /// one's handle is untracked and dropped (replaced with a fresh empty + /// handle) so its fd + sparse index free. An in-flight poll holding a clone + /// of the dropped handle keeps it alive until it finishes (see + /// [`SealedSegmentHandle`]). + pub fn touch_sealed_read_state(&mut self, slot: usize) { + let Some(touched) = self.segments.get(slot) else { + return; + }; + if !touched.sealed { + return; + } + let start_offset = touched.start_offset; + self.sealed_read_state[slot].tracked.set(true); + if let Some(pos) = self + .sealed_lru + .iter() + .position(|&offset| offset == start_offset) + { + self.sealed_lru.remove(pos); + } + self.sealed_lru.push_front(start_offset); + if self.sealed_lru.len() > SEALED_READ_STATE_CAP { + let Some(evicted) = self.sealed_lru.pop_back() else { + return; + }; + if let Some(evicted_slot) = self + .segments + .iter() + .position(|segment| segment.start_offset == evicted) + { + // Untrack before orphaning so an in-flight poll holding the old + // handle stops caching fds into it. + self.sealed_read_state[evicted_slot].tracked.set(false); + self.sealed_read_state[evicted_slot] = SealedSegmentHandle::default(); + } + } + } + + /// Retire the oldest segment: pop the front of every parallel vec in + /// lockstep and purge the segment's sealed-LRU entry. Returns the pieces + /// the caller still needs (stats + file unlink), or `None` on an empty + /// log. The read-state handle is dropped here; an in-flight poll holding a + /// clone keeps it alive until it finishes (a cached fd reads the unlinked + /// inode). + pub fn retire_front(&mut self) -> Option<(Segment, SegmentStorage)> { + if self.segments.is_empty() { + return None; + } + self.debug_assert_lockstep(); + let segment = self.segments.remove(0); + let storage = self.storage.remove(0); + self.indexes.remove(0); + self.messages_writers.remove(0); + self.index_writers.remove(0); + self.sealed_read_state.remove(0); + self.sealed_lru + .retain(|&offset| offset != segment.start_offset); + Some((segment, storage)) + } + + fn debug_assert_lockstep(&self) { + debug_assert!( + self.segments.len() == self.storage.len() + && self.segments.len() == self.indexes.len() + && self.segments.len() == self.messages_writers.len() + && self.segments.len() == self.index_writers.len() + && self.segments.len() == self.sealed_read_state.len(), + "segment parallel vecs out of lockstep" + ); + } + + pub fn storages_mut(&mut self) -> &mut [SegmentStorage] { &mut self.storage } @@ -189,7 +299,7 @@ where &self.messages_writers } - pub const fn messages_writers_mut(&mut self) -> &mut Vec<Option<Rc<MessagesWriter>>> { + pub fn messages_writers_mut(&mut self) -> &mut [Option<Rc<MessagesWriter>>] { &mut self.messages_writers } @@ -197,7 +307,7 @@ where &self.index_writers } - pub const fn index_writers_mut(&mut self) -> &mut Vec<Option<Rc<IggyIndexWriter>>> { + pub fn index_writers_mut(&mut self) -> &mut [Option<Rc<IggyIndexWriter>>] { &mut self.index_writers } @@ -229,7 +339,7 @@ where &self.indexes } - pub const fn indexes_mut(&mut self) -> &mut Vec<Option<IggyIndexCache>> { + pub fn indexes_mut(&mut self) -> &mut [Option<IggyIndexCache>] { &mut self.indexes } @@ -283,6 +393,8 @@ where self.indexes.push(None); self.messages_writers.push(messages_writer); self.index_writers.push(index_writer); + self.sealed_read_state.push(SealedSegmentHandle::default()); + self.debug_assert_lockstep(); } pub fn set_segment_indexes(&mut self, segment_index: usize, indexes: IggyIndexCache) { @@ -321,3 +433,176 @@ where &self.journal } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::journal::{PartitionJournal, PartitionJournalMemStorage}; + + type TestLog = + SegmentedLog<PartitionJournal<PartitionJournalMemStorage>, PartitionJournalMemStorage>; + + /// Push a sealed segment with a resident (index-filled) read handle and + /// return a clone of that handle, standing in for an in-flight poll's clone. + /// Goes through `add_persisted_segment` so every parallel vec stays in + /// lockstep (segment start offsets equal their slot indexes in every test + /// that pushes ascending offsets from an empty log). + fn push_resident_sealed(log: &mut TestLog, start_offset: u64) -> SealedSegmentHandle { + log.add_persisted_segment( + Segment { + start_offset, + sealed: true, + ..Segment::default() + }, + SegmentStorage::default(), + None, + None, + ); + let slot = log.sealed_read_state.len() - 1; + *log.sealed_read_state[slot].index.borrow_mut() = Some(IggyIndexCache::with_capacity(1)); + Rc::clone(&log.sealed_read_state[slot]) + } + + #[test] + fn touch_sealed_read_state_evicts_least_recently_used_past_cap() { + let mut log = TestLog::default(); + let handles: Vec<_> = (0..=SEALED_READ_STATE_CAP as u64) + .map(|offset| push_resident_sealed(&mut log, offset)) + .collect(); + // Ascending touch order: slot/offset 0 is the least-recently used. + for slot in 0..=SEALED_READ_STATE_CAP { + log.touch_sealed_read_state(slot); + } + + // Slot 0 dropped: the pump handle was replaced with a fresh empty one. + assert!( + !Rc::ptr_eq(&handles[0], &log.sealed_read_state()[0]), + "least-recently-used handle must be dropped past the cap", + ); + assert!( + log.sealed_read_state()[0].index.borrow().is_none(), + "the dropped slot resets to an empty handle", + ); + // In-flight safety: the dropped handle's clone stays alive and still + // sees its cached index, so a poll holding it finishes without a UAF. + assert_eq!( + Rc::strong_count(&handles[0]), + 1, + "the dropped handle survives for an in-flight poll's clone", + ); + assert!( + handles[0].index.borrow().is_some(), + "the in-flight clone keeps reading the cached index", + ); + assert!( + !handles[0].tracked.get(), + "eviction untracks the orphaned handle so in-flight polls stop caching fds into it", + ); + // Every more-recently-used slot is retained (same Rc). + for (handle, resident) in handles.iter().zip(log.sealed_read_state().iter()).skip(1) { + assert!( + Rc::ptr_eq(handle, resident), + "recently-used handles stay resident", + ); + } + } + + #[test] + fn touch_sealed_read_state_reorders_eviction_on_reaccess() { + let mut log = TestLog::default(); + let mut handles: Vec<_> = (0..SEALED_READ_STATE_CAP as u64) + .map(|offset| push_resident_sealed(&mut log, offset)) + .collect(); + for slot in 0..SEALED_READ_STATE_CAP { + log.touch_sealed_read_state(slot); + } + // Re-access slot 0 -> now most-recently used, so slot 1 becomes the + // least-recently used and next to be evicted. + log.touch_sealed_read_state(0); + + handles.push(push_resident_sealed(&mut log, SEALED_READ_STATE_CAP as u64)); + log.touch_sealed_read_state(SEALED_READ_STATE_CAP); + + assert!( + !Rc::ptr_eq(&handles[1], &log.sealed_read_state()[1]), + "the least-recently-used segment is evicted, not the re-accessed one", + ); + assert!( + Rc::ptr_eq(&handles[0], &log.sealed_read_state()[0]), + "the re-accessed segment stays resident", + ); + } + + #[test] + fn evicted_sealed_slot_is_empty_and_refillable() { + let mut log = TestLog::default(); + for offset in 0..=SEALED_READ_STATE_CAP as u64 { + push_resident_sealed(&mut log, offset); + } + for slot in 0..=SEALED_READ_STATE_CAP { + log.touch_sealed_read_state(slot); + } + + // The evicted slot holds a fresh empty handle, so the next poll re-opens + // instead of reusing a stale descriptor. + let evicted = &log.sealed_read_state()[0]; + assert!(evicted.fd.borrow().is_none()); + assert!(evicted.index.borrow().is_none()); + + // Re-filling it (what the next sealed poll does) works. + *log.sealed_read_state()[0].index.borrow_mut() = Some(IggyIndexCache::with_capacity(1)); + assert!(log.sealed_read_state()[0].index.borrow().is_some()); + } + + #[test] + fn touch_sealed_read_state_ignores_active_and_out_of_range_slots() { + let mut log = TestLog::default(); + log.add_persisted_segment( + Segment { + start_offset: 0, + sealed: false, + ..Segment::default() + }, + SegmentStorage::default(), + None, + None, + ); + + // Active (unsealed) segment: no LRU entry, handle stays untracked, so + // its fd is never retained outside the cap. + log.touch_sealed_read_state(0); + assert!(log.sealed_lru.is_empty()); + assert!(!log.sealed_read_state()[0].tracked.get()); + + // Out-of-range slot (the purge drain window empties the vec across + // awaits): must be a no-op, not a panic. + log.touch_sealed_read_state(1); + assert!(log.sealed_lru.is_empty()); + } + + #[test] + fn retire_front_purges_lru_entry_and_keeps_vecs_lockstep() { + let mut log = TestLog::default(); + push_resident_sealed(&mut log, 0); + push_resident_sealed(&mut log, 5); + log.touch_sealed_read_state(0); + log.touch_sealed_read_state(1); + + let (segment, _storage) = log.retire_front().expect("log has segments"); + assert_eq!(segment.start_offset, 0); + assert!( + !log.sealed_lru.contains(&0), + "retire must purge the segment's LRU entry", + ); + assert!(log.sealed_lru.contains(&5), "the survivor's entry stays"); + assert_eq!(log.segments().len(), 1); + assert_eq!(log.sealed_read_state().len(), 1); + assert_eq!(log.storages().len(), 1); + + assert!(log.retire_front().is_some()); + assert!( + log.retire_front().is_none(), + "an empty log retires nothing instead of panicking", + ); + } +} diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs index 7c0e5afab..11c3d3e02 100644 --- a/core/partitions/src/poll_plan.rs +++ b/core/partitions/src/poll_plan.rs @@ -24,11 +24,15 @@ //! synchronously under the borrow into the owned types here, drops the borrow, //! then [`PollPlan::execute`] runs the disk read + the in-memory auto-commit //! apply on owned data alone: consumer offsets are already `Arc`, the journal -//! tail is a point-in-time `Frozen` snapshot, and segment files are re-opened -//! by path. No value in this module holds a partition reference, so executing a -//! plan is sound on a detached task concurrently with the pump's own writes. +//! tail is a point-in-time `Frozen` snapshot, and each sealed segment carries a +//! shared [`SealedSegmentReadState`] handle (a plain `Rc`, not a partition +//! reference) whose read fd + sparse index the read reuses or fills on a miss. +//! No value in this module holds a partition reference, so executing a plan is +//! sound on a detached task concurrently with the pump's own writes. use crate::PollFragments; +use crate::iggy_index::IggyIndexCache; +use crate::iggy_index_reader::IggyIndexReader; use crate::journal::{MessageLookup, push_selected_batch_fragments, select_batch_slice}; use compio::io::AsyncReadAtExt; use iggy_common::{ @@ -36,7 +40,9 @@ use iggy_common::{ }; use server_common::iobuf::{Frozen, Owned}; use server_common::send_messages2::{COMMAND_HEADER_SIZE, decode_batch_slice}; +use std::cell::{Cell, RefCell}; use std::hash::Hash; +use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::Ordering; use tracing::warn; @@ -58,9 +64,39 @@ pub enum PartitionDirResolution { Unresolvable, } -/// Owned, borrow-free inputs for the disk tier of a poll (see module docs). -/// Segment files are re-opened by path because sealed segments drop their -/// writer at rotation. +/// Per-sealed-segment read state, shared as a cheap `Rc` handle between the +/// owning partition and the off-borrow [`DiskReadPlan`] (a plain `Rc`, never a +/// partition reference, so the read runs off the pump). Both slots fill lazily +/// on the first sealed poll and are reused after. The pump drops its handle +/// when the segment is retired, so the state frees once any in-flight poll +/// holding a clone finishes (a cached fd meanwhile reads the unlinked inode, +/// which is fine). The active segment is never cached. +#[derive(Debug, Default)] +pub struct SealedSegmentReadState { + /// Read-only descriptor; compio `File` clones share the kernel fd, so a hit + /// avoids the per-poll `openat` (an `io_uring` op prone to io-wq punts) and + /// preserves kernel readahead. `None` until the first sealed poll opens it. + pub(crate) fd: RefCell<Option<compio::fs::File>>, + /// Sparse offset/timestamp index reloaded from the `.index` file the + /// segment dropped at rotation, so a poll resolves the start byte in + /// O(log n) instead of scanning the whole segment from byte 0 (the stall). + /// `None` until the first sealed poll loads it. + pub(crate) index: RefCell<Option<IggyIndexCache>>, + /// Whether the owning partition's sealed LRU currently tracks this handle. + /// Gates the fd store-back in `resolve_segment_file`: a walk crosses every + /// sealed segment from the poll's start onward, but only the start segment + /// is LRU-touched, so an untracked fill would retain a descriptor the + /// `SEALED_READ_STATE_CAP` budget never counts. Set on touch, cleared on + /// evict; plain `Cell`, all access is same-thread (`Rc` handle). + pub(crate) tracked: Cell<bool>, +} + +pub type SealedSegmentHandle = Rc<SealedSegmentReadState>; + +/// Owned, borrow-free inputs for the disk tier of a poll (see module docs). A +/// sealed segment reuses its cached [`SealedSegmentReadState`] (read fd + sparse +/// index); the active segment (and any cache miss) opens by path and resolves +/// from its resident index, because sealed segments drop both at rotation. pub struct DiskReadPlan { pub(crate) partition_dir: PartitionDirResolution, /// Segments to walk, snapshotted from the poll's starting segment onward @@ -74,6 +110,10 @@ pub struct DiskReadPlan { pub struct DiskSegment { pub(crate) start_offset: u64, pub(crate) persisted: u64, + /// Shared read state, cloned from the owning partition at plan time for a + /// SEALED segment; `None` for the active segment, which always opens fresh + /// and resolves from its resident index. See [`SealedSegmentReadState`]. + pub(crate) read_state: Option<SealedSegmentHandle>, } /// Owned auto-commit input, applied off the partition borrow after a poll (see @@ -427,7 +467,21 @@ impl DiskReadPlan { // `start_position` applies to the first snapshotted segment; each later // segment is walked from byte 0 (reset at the end of every iteration). - let mut position = self.start_position; + // + // A sealed first segment dropped its resident index at rotation, so + // `disk_poll_start` fell back to byte 0. Reload the sparse index (once, + // then cached) and resolve the start byte so the walk skips straight to + // the target instead of scanning the whole segment - the poll stall. A + // miss or load failure keeps `start_position` (the pre-existing + // full-scan fallback). The active segment carries no read state, so its + // resident-index-resolved `start_position` is left untouched. + let mut position = match self.segments.first() { + Some(first) => self + .resolve_sealed_start(first, query, partition_dir) + .await + .unwrap_or(self.start_position), + None => self.start_position, + }; let mut fragments = PollFragments::new(); let mut last_matching_offset = None; let mut matched: u32 = 0; @@ -448,7 +502,7 @@ impl DiskReadPlan { continue; } let path = format!("{partition_dir}/{:0>20}.log", segment.start_offset); - let Some(file) = self.open_segment_with_retry(&path).await else { + let Some(file) = self.resolve_segment_file(segment, &path).await else { // Open exhausted retries: the segment may hold present-but- // unreadable data. Stop here rather than walking past it. faulted = true; @@ -507,6 +561,99 @@ impl DiskReadPlan { } } + /// Resolve the read-only descriptor for `segment`'s file. A sealed segment + /// clones its cached fd on a hit (sharing the kernel fd, no syscall) and, on + /// a miss, opens by path and stores the fd back so later polls skip the + /// `openat`. The active segment (no cache slot) always opens fresh. Returns + /// `None` only when the open exhausts its retries (the caller fails closed). + async fn resolve_segment_file( + &self, + segment: &DiskSegment, + path: &str, + ) -> Option<compio::fs::File> { + let Some(handle) = &segment.read_state else { + 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() { + return Some(cached); + } + let file = self.open_segment_with_retry(path).await?; + // Store back only while the pump tracks this handle; an untracked + // fill (walk-through segment, or a slot evicted mid-poll) would pin an + // fd outside the LRU budget, so it opens transiently instead. Benign + // race: a concurrent poll of the same segment may have filled the slot + // while this open was in flight; overwriting with an equivalent fd + // (same inode) is harmless. + if handle.tracked.get() { + *handle.fd.borrow_mut() = Some(file.clone()); + } + Some(file) + } + + /// Resolve the start byte for the poll's target segment from its sparse + /// index, loading the `.index` file on the first sealed poll and caching it + /// on the shared handle. Returns `None` (keep the byte-0 fallback) for the + /// active segment (no handle), a below-range query, or a load failure. + async fn resolve_sealed_start( + &self, + segment: &DiskSegment, + query: MessageLookup, + partition_dir: &str, + ) -> Option<u64> { + // TODO: a per-consumer cursor hint (the previous sealed poll's resolved + // position) could seed this so a sequentially advancing consumer skips + // the sparse-index lookup on repeated polls of the same segment. + let handle = segment.read_state.as_ref()?; + // Cache hit: resolve under a short borrow, never across the await. + let cached = handle + .index + .borrow() + .as_ref() + .map(|index| resolve_index_position(index, query)); + if let Some(resolved) = cached { + return resolved; + } + let path = format!("{partition_dir}/{:0>20}.index", segment.start_offset); + let index = self.load_sealed_index(&path).await?; + let resolved = resolve_index_position(&index, query); + *handle.index.borrow_mut() = Some(index); + resolved + } + + /// Load a sealed segment's sparse index from its `.index` file. `None` on a + /// missing/unreadable file so the caller falls back to a byte-0 scan (the + /// pre-existing behavior); the load is retried on the next poll. + async fn load_sealed_index(&self, path: &str) -> Option<IggyIndexCache> { + match IggyIndexReader::new(path).await { + Ok(reader) => match reader.load_all().await { + Ok(index) => Some(index), + Err(error) => { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace_raw, + path, + %error, + "disk poll: failed to read sparse index; scanning from segment start" + ); + None + } + }, + Err(error) => { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace_raw, + path, + %error, + "disk poll: failed to open sparse index; scanning from segment start" + ); + None + } + } + } + /// Open a segment file for a disk poll, retrying transient IO failures (fd /// pressure under heavy parallel load) so one failed syscall does not /// silently collapse the poll into an empty result. @@ -565,6 +712,18 @@ impl DiskReadPlan { } } +/// Byte position of the sparse-index entry at or below the query's offset / +/// timestamp, or `None` when the query is below the first indexed entry (the +/// caller then scans from the segment start). Mirrors `disk_poll_start`'s +/// resident-index resolution for the sealed, off-pump path. +fn resolve_index_position(index: &IggyIndexCache, query: MessageLookup) -> Option<u64> { + match query { + MessageLookup::Offset { offset, .. } => index.offset_lower_bound(offset), + MessageLookup::Timestamp { timestamp, .. } => index.timestamp_lower_bound(timestamp), + } + .map(|entry| entry.position) +} + impl AutoCommitCtx { /// The offset key (kind + numeric id) this auto-commit targets, for the /// replicated `StoreConsumerOffset2` op the serving shard submits. diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index c118e8d02..af6d3c9ce 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -773,8 +773,7 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket = 8093, tcp_replica = # - numa settings: # + "numa:auto": Use all available numa node, cores # + "numa:nodes=0,1;cores=4;no_ht=true": Use NUMA node 0 and 1, each nodes use 4 cores, and no hyperthreads -# TODO(hubcio): revert to "numa:auto" once multi-shard server-ng is stable. -cpu_allocation = 1 +cpu_allocation = "numa:auto" # Whether shard threads are pinned to dedicated CPU cores (default: true). # Pinned cores are drawn from the process's allowed CPU set (affinity/cpuset diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index a131fdca3..ee155f09a 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -187,10 +187,10 @@ where { let shard_handle = Rc::clone(shard_handle); // Runs synchronously on the shard pump (see `process_lifecycle` -> - // `on_partition_read`). `build_poll_snapshot` takes the partition borrow via - // `with_partition` (closure-scoped, debug `BorrowGuard`) and returns an owned - // `PollPlan`; only owned data crosses into `spawn_poll_io`. A fully-resident - // poll replies here without spawning. See the `poll_plan` module docs. + // `on_partition_read`). `build_poll_snapshot` takes a pump-only `&mut` + // partition borrow (synchronous, so no sibling task can realloc under it) and + // returns an owned `PollPlan`; only owned data crosses into `spawn_poll_io`. A + // fully-resident poll replies here without spawning. See the `poll_plan` module docs. Rc::new(move |namespace, read, reply| { let Some(shard) = upgrade_shard_handle(&shard_handle) else { return;
