Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
58 kB · 1814 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815use std::collections::HashMap;use std::io;use std::sync::Arc;use std::sync::atomic::{AtomicU8, AtomicU32, Ordering};use std::thread;
use parking_lot::{Mutex, RwLock};
use crate::clock::{Clock, LogicalNanos};use crate::fsync_order::PostBlockstoreHook;
use super::BlocksSynced;use crate::io::{FileId, OpenOptions, StorageIO};
use super::data_file::{CID_SIZE, DataFileWriter, ReadBlockRecord, decode_block_record};use super::hash_index::{BlockIndex, BlockIndexError, CheckpointPositions};use super::hint::{HINT_RECORD_SIZE, HintFileWriter, hint_file_path};use super::manager::DataFileManager;use super::types::{ BlockLocation, BlockOffset, BlockstoreSnapshot, CommitEpoch, DataFileId, EpochCounter, HintOffset, ShardId, WriteCursor,};
pub struct FileIdAllocator { next: AtomicU32,}
impl FileIdAllocator { pub fn new(max_existing: DataFileId) -> Self { Self { next: AtomicU32::new(max_existing.raw().saturating_add(1)), } }
pub fn allocate(&self) -> DataFileId { let id = self .next .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| v.checked_add(1)) .expect("FileIdAllocator overflow: exhausted u32 file ID space"); DataFileId::new(id) }
pub fn peek(&self) -> DataFileId { DataFileId::new(self.next.load(Ordering::Relaxed)) }}
pub struct ActiveFileSet { files: RwLock<Vec<DataFileId>>,}
impl ActiveFileSet { pub fn with_capacity(n: usize) -> Self { Self { files: RwLock::new(Vec::with_capacity(n)), } }
pub fn register(&self, shard: ShardId, file_id: DataFileId) { let mut files = self.files.write(); let idx = shard.as_usize(); if idx >= files.len() { files.resize(idx.saturating_add(1), DataFileId::new(0)); } files[idx] = file_id; }
pub fn contains(&self, file_id: DataFileId) -> bool { self.files.read().contains(&file_id) }
pub fn snapshot(&self) -> Vec<DataFileId> { self.files.read().clone() }}
pub struct ShardHintPositions { shard_positions: RwLock<Vec<(DataFileId, HintOffset)>>, extra_positions: RwLock<HashMap<DataFileId, HintOffset>>,}
impl ShardHintPositions { pub fn new(shard_count: u8) -> Self { Self { shard_positions: RwLock::new( (0..shard_count as usize) .map(|_| (DataFileId::new(0), HintOffset::new(0))) .collect(), ), extra_positions: RwLock::new(HashMap::new()), } }
pub fn update(&self, shard_id: ShardId, file_id: DataFileId, offset: HintOffset) { let mut positions = self.shard_positions.write(); let idx = shard_id.as_usize(); if idx < positions.len() { positions[idx] = (file_id, offset); } }
pub fn record_extra(&self, file_id: DataFileId, offset: HintOffset) { self.extra_positions.write().insert(file_id, offset); }
pub fn forget_extra(&self, file_id: DataFileId) { self.extra_positions.write().remove(&file_id); }
pub fn snapshot(&self) -> CheckpointPositions { let shard = self.shard_positions.read().clone(); let extra = self.extra_positions.read().clone(); debug_assert!( shard .iter() .filter(|(fid, _)| fid.raw() != 0) .all(|(fid, _)| !extra.contains_key(fid)), "shard_positions and extra_positions must not overlap on the same DataFileId" ); CheckpointPositions(shard.into_iter().chain(extra).collect()) }}
#[derive(Debug, Clone)]pub enum CommitError { Io(Arc<io::Error>), Index(String), ChannelClosed, VerifyFailed { file_id: DataFileId, offset: BlockOffset, },}
impl std::fmt::Display for CommitError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Io(e) => write!(f, "io: {}", e.as_ref()), Self::Index(e) => write!(f, "index: {e}"), Self::ChannelClosed => write!(f, "commit channel closed"), Self::VerifyFailed { file_id, offset } => write!( f, "post-sync verify failed at {file_id}:{} (misdirected write or durable corruption)", offset.raw() ), } }}
impl std::error::Error for CommitError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::Io(e) => Some(e.as_ref()), Self::Index(_) | Self::ChannelClosed | Self::VerifyFailed { .. } => None, } }}
impl From<io::Error> for CommitError { fn from(e: io::Error) -> Self { Self::Io(Arc::new(e)) }}
impl From<BlockIndexError> for CommitError { fn from(e: BlockIndexError) -> Self { Self::Index(e.to_string()) }}
use super::compaction::{self, CompactionError};use super::types::{CidBytes, CompactionResult, RefCount};
type PutResponse = tokio::sync::oneshot::Sender<Result<Vec<BlockLocation>, CommitError>>;type ApplyResponse = tokio::sync::oneshot::Sender<Result<(), CommitError>>;type CompactResponse = tokio::sync::oneshot::Sender<Result<CompactionResult, CompactionError>>;type RepairResponse = tokio::sync::oneshot::Sender<Result<u64, CommitError>>;type RepairBlocksResponse = tokio::sync::oneshot::Sender<Result<u64, CompactionError>>;type RepairBlockSet = Vec<(CidBytes, Vec<u8>)>;type DeferredRepairBlock = (RepairBlockSet, RepairBlocksResponse);type QuiesceResponse = tokio::sync::oneshot::Sender<BlockstoreSnapshot>;type QuiesceResume = tokio::sync::oneshot::Receiver<()>;
pub enum CommitRequest { PutBlocks { blocks: Vec<([u8; CID_SIZE], Vec<u8>)>, response: PutResponse, }, ApplyCommit { blocks: Vec<([u8; CID_SIZE], Vec<u8>)>, deleted_cids: Vec<[u8; CID_SIZE]>, response: ApplyResponse, }, Compact { file_id: DataFileId, grace_period_ms: u64, response: CompactResponse, }, RepairLeaked { leaked_cids: Vec<(CidBytes, RefCount)>, response: RepairResponse, }, RepairBlocks { blocks: RepairBlockSet, response: RepairBlocksResponse, }, Quiesce { response: QuiesceResponse, resume: QuiesceResume, }, Shutdown,}
#[derive(Debug, Clone)]pub struct GroupCommitConfig { pub max_batch_size: usize, pub channel_capacity: usize, pub checkpoint_interval_ms: u64, pub checkpoint_write_threshold: u64, pub verify_persisted_blocks: bool, pub synchronous: bool,}
impl Default for GroupCommitConfig { fn default() -> Self { Self { max_batch_size: 1024, channel_capacity: 4096, checkpoint_interval_ms: 60_000, checkpoint_write_threshold: 100_000, verify_persisted_blocks: false, synchronous: false, } }}
struct ShardContext<C: Clock> { shard_id: ShardId, epoch: EpochCounter, file_ids: Arc<FileIdAllocator>, active_files: Arc<ActiveFileSet>, hint_positions: Arc<ShardHintPositions>, verify_persisted_blocks: bool, clock: C,}
struct ActiveState { file_id: DataFileId, fd: FileId, position: BlockOffset, hint_fd: FileId, hint_position: HintOffset,}
fn log_thread_panic(payload: Box<dyn std::any::Any + Send>, context: &str) { let msg = payload .downcast_ref::<&str>() .copied() .or_else(|| payload.downcast_ref::<String>().map(|s| s.as_str())) .unwrap_or("unknown panic"); tracing::error!(panic = msg, "{context}");}
trait InlineHandler: Send { fn handle(&mut self, req: CommitRequest); fn finalize(&mut self);}
struct InlineState<S: StorageIO + 'static, C: Clock> { manager: DataFileManager<S>, index: Arc<BlockIndex>, config: GroupCommitConfig, state: ActiveState, ctx: ShardContext<C>, post_sync_hook: Option<Arc<dyn PostBlockstoreHook>>, last_checkpoint: LogicalNanos, writes_since_checkpoint: u64,}
impl<S: StorageIO + 'static, C: Clock> InlineHandler for InlineState<S, C> { fn handle(&mut self, req: CommitRequest) { handle_request( req, &self.manager, &self.index, &self.config, &mut self.state, self.post_sync_hook.as_deref(), &self.ctx, &mut self.last_checkpoint, &mut self.writes_since_checkpoint, ); }
fn finalize(&mut self) { shutdown_checkpoint(&self.index, &self.ctx.epoch, &self.ctx.hint_positions); }}
enum SingleShardWriter { Threaded { sender: flume::Sender<CommitRequest>, handle: Option<thread::JoinHandle<()>>, }, Inline { handler: Mutex<Box<dyn InlineHandler>>, },}
impl SingleShardWriter { fn spawn<S: StorageIO + 'static, C: Clock>( ctx: ShardContext<C>, manager: DataFileManager<S>, index: Arc<BlockIndex>, config: GroupCommitConfig, post_sync_hook: Option<Arc<dyn PostBlockstoreHook>>, cursor: Option<WriteCursor>, ) -> Result<Self, CommitError> { let mut state = initialize_active_state(&manager, cursor, &ctx.file_ids)?; ctx.active_files.register(ctx.shard_id, state.file_id); ctx.hint_positions .update(ctx.shard_id, state.file_id, state.hint_position);
if config.synchronous { let last_checkpoint = ctx.clock.monotonic(); let inline = InlineState { manager, index, config, state, ctx, post_sync_hook, last_checkpoint, writes_since_checkpoint: 0, }; return Ok(Self::Inline { handler: Mutex::new(Box::new(inline)), }); }
let (sender, receiver) = flume::bounded(config.channel_capacity);
let handle = thread::Builder::new() .name(format!("blockstore-commit-{}", ctx.shard_id)) .spawn(move || { commit_loop( &manager, &index, &receiver, &config, &mut state, post_sync_hook.as_deref(), &ctx, ); }) .map_err(|e| CommitError::from(io::Error::other(e)))?;
Ok(Self::Threaded { sender, handle: Some(handle), }) }
fn submit(&self, request: CommitRequest) -> Result<(), CommitError> { match self { Self::Threaded { sender, .. } => { sender.send(request).map_err(|_| CommitError::ChannelClosed) } Self::Inline { handler } => { handler.lock().handle(request); Ok(()) } } }
fn shutdown(&mut self) { match self { Self::Threaded { sender, handle } => { let _ = sender.send(CommitRequest::Shutdown); if let Some(handle) = handle.take() && let Err(payload) = handle.join() { log_thread_panic(payload, "group commit thread panicked"); } } Self::Inline { handler } => handler.lock().finalize(), } }}
impl Drop for SingleShardWriter { fn drop(&mut self) { if let Self::Threaded { sender, handle } = self { let _ = sender.try_send(CommitRequest::Shutdown); if let Some(handle) = handle.take() && let Err(payload) = handle.join() { log_thread_panic(payload, "group commit thread panicked during drop"); } } }}
fn shard_for_cid(cid: &[u8; CID_SIZE], shard_count: u8) -> usize { match shard_count { 0 | 1 => 0, n => { let hash_bytes: [u8; 8] = cid[4..12].try_into().unwrap(); let hash = u64::from_le_bytes(hash_bytes); match n.is_power_of_two() { true => (hash & (n as u64 - 1)) as usize, false => (hash % n as u64) as usize, } } }}
fn pick_shard_for_blocks(blocks: &[([u8; CID_SIZE], Vec<u8>)], shard_count: u8) -> usize { match blocks.first() { Some((cid, _)) => shard_for_cid(cid, shard_count), None => 0, }}
fn pick_shard_for_apply( blocks: &[([u8; CID_SIZE], Vec<u8>)], deleted_cids: &[[u8; CID_SIZE]], shard_count: u8,) -> usize { match blocks.first() { Some((cid, _)) => shard_for_cid(cid, shard_count), None => match deleted_cids.first() { Some(cid) => shard_for_cid(cid, shard_count), None => 0, }, }}
pub struct GroupCommitWriter { shards: Vec<SingleShardWriter>, epoch: EpochCounter, shard_count: u8, round_robin: AtomicU8, _file_ids: Arc<FileIdAllocator>, _active_files: Arc<ActiveFileSet>, _hint_positions: Arc<ShardHintPositions>,}
impl GroupCommitWriter { pub fn spawn<S: StorageIO + 'static, C: Clock>( make_manager: impl Fn() -> DataFileManager<S>, index: Arc<BlockIndex>, config: GroupCommitConfig, clock: C, ) -> Result<Self, CommitError> { Self::spawn_sharded(make_manager, index, config, None, None, 1, None, clock) }
pub fn spawn_with_hook<S: StorageIO + 'static, C: Clock>( make_manager: impl Fn() -> DataFileManager<S>, index: Arc<BlockIndex>, config: GroupCommitConfig, post_sync_hook: Option<Arc<dyn PostBlockstoreHook>>, initial_epoch: Option<CommitEpoch>, clock: C, ) -> Result<Self, CommitError> { Self::spawn_sharded( make_manager, index, config, post_sync_hook, initial_epoch, 1, None, clock, ) }
#[allow(clippy::too_many_arguments)] pub fn spawn_sharded<F, S, C>( make_manager: F, index: Arc<BlockIndex>, config: GroupCommitConfig, post_sync_hook: Option<Arc<dyn PostBlockstoreHook>>, initial_epoch: Option<CommitEpoch>, shard_count: u8, checkpoint_positions: Option<&CheckpointPositions>, clock: C, ) -> Result<Self, CommitError> where S: StorageIO + 'static, C: Clock, F: Fn() -> DataFileManager<S>, { let shard_count = shard_count.max(1);
let probe_manager = make_manager(); let existing_files = probe_manager.list_files()?; let max_existing = existing_files.last().copied().unwrap_or(DataFileId::new(0)); let file_ids = Arc::new(FileIdAllocator::new(max_existing)); let active_files = Arc::new(ActiveFileSet::with_capacity(shard_count as usize)); let hint_positions = Arc::new(ShardHintPositions::new(shard_count)); drop(probe_manager);
let epoch = match initial_epoch { Some(e) => EpochCounter::from_raw(e.raw()), None => EpochCounter::new(), };
let global_cursor = index.read_write_cursor();
let shards: Result<Vec<SingleShardWriter>, CommitError> = (0..shard_count) .map(|i| { let shard_cursor = checkpoint_positions .and_then(|cp| cp.0.get(i as usize)) .filter(|(fid, _)| fid.raw() > 0) .map(|(fid, _)| WriteCursor { file_id: *fid, offset: BlockOffset::new(0), }) .or(match i { 0 => global_cursor, _ => None, }); let ctx = ShardContext { shard_id: ShardId::new(i), epoch: epoch.clone(), file_ids: Arc::clone(&file_ids), active_files: Arc::clone(&active_files), hint_positions: Arc::clone(&hint_positions), verify_persisted_blocks: config.verify_persisted_blocks, clock: clock.clone(), }; SingleShardWriter::spawn( ctx, make_manager(), Arc::clone(&index), config.clone(), post_sync_hook.clone(), shard_cursor, ) }) .collect();
Ok(Self { shards: shards?, epoch, shard_count, round_robin: AtomicU8::new(0), _file_ids: file_ids, _active_files: active_files, _hint_positions: hint_positions, }) }
pub fn epoch(&self) -> &EpochCounter { &self.epoch }
fn round_robin_index(&self) -> usize { self.round_robin .fetch_add(1, Ordering::Relaxed) .wrapping_rem(self.shard_count) as usize }
fn shard_index_for(&self, request: &CommitRequest) -> usize { match request { CommitRequest::PutBlocks { blocks, .. } => { pick_shard_for_blocks(blocks, self.shard_count) } CommitRequest::ApplyCommit { blocks, deleted_cids, .. } => pick_shard_for_apply(blocks, deleted_cids, self.shard_count), CommitRequest::Compact { .. } | CommitRequest::RepairLeaked { .. } | CommitRequest::RepairBlocks { .. } | CommitRequest::Quiesce { .. } | CommitRequest::Shutdown => self.round_robin_index(), } }
pub fn submit_blocking(&self, request: CommitRequest) -> Result<(), CommitError> { let idx = self.shard_index_for(&request); self.shards[idx].submit(request) }
fn threaded_sender(shard: &SingleShardWriter) -> &flume::Sender<CommitRequest> { match shard { SingleShardWriter::Threaded { sender, .. } => sender, SingleShardWriter::Inline { .. } => unreachable!( "async commit path requires threaded group-commit; synchronous mode is gauntlet-blocking-only" ), } }
pub fn sender_for_blocks( &self, blocks: &[([u8; CID_SIZE], Vec<u8>)], ) -> &flume::Sender<CommitRequest> { Self::threaded_sender(&self.shards[pick_shard_for_blocks(blocks, self.shard_count)]) }
pub fn sender_for_apply( &self, blocks: &[([u8; CID_SIZE], Vec<u8>)], deleted_cids: &[[u8; CID_SIZE]], ) -> &flume::Sender<CommitRequest> { Self::threaded_sender( &self.shards[pick_shard_for_apply(blocks, deleted_cids, self.shard_count)], ) }
pub fn quiesce_all( &self, ) -> Result<(BlockstoreSnapshot, Vec<tokio::sync::oneshot::Sender<()>>), CommitError> { let pending: Result<Vec<_>, CommitError> = self .shards .iter() .map(|shard| { let (response_tx, response_rx) = tokio::sync::oneshot::channel(); let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); let req = CommitRequest::Quiesce { response: response_tx, resume: resume_rx, }; match shard { SingleShardWriter::Threaded { sender, .. } => { sender.send(req).map_err(|_| CommitError::ChannelClosed)?; } SingleShardWriter::Inline { handler } => { handler.lock().handle(req); } } Ok((response_rx, resume_tx)) }) .collect();
let pairs: Result<Vec<_>, CommitError> = pending? .into_iter() .map(|(response_rx, resume_tx)| { let snapshot = response_rx .blocking_recv() .map_err(|_| CommitError::ChannelClosed)?; Ok((snapshot, resume_tx)) }) .collect(); let pairs = pairs?;
let mut all_data_files: Vec<DataFileId> = Vec::new(); let mut shard_cursors: Vec<WriteCursor> = Vec::new(); let mut max_epoch = CommitEpoch::zero();
pairs.iter().for_each(|(snap, _)| { shard_cursors.extend(&snap.shard_cursors); all_data_files.extend(&snap.data_files); if snap.epoch > max_epoch { max_epoch = snap.epoch; } });
all_data_files.sort(); all_data_files.dedup();
let resumes = pairs.into_iter().map(|(_, resume)| resume).collect();
Ok(( BlockstoreSnapshot { shard_cursors, epoch: max_epoch, data_files: all_data_files, }, resumes, )) }
pub fn shutdown(mut self) { self.shards.iter_mut().for_each(|s| s.shutdown()); }}
impl Drop for GroupCommitWriter { fn drop(&mut self) { self.shards.iter_mut().for_each(|s| { if let SingleShardWriter::Threaded { sender, .. } = s { let _ = sender.try_send(CommitRequest::Shutdown); } }); self.shards.iter_mut().for_each(|s| { if let SingleShardWriter::Threaded { handle, .. } = s && let Some(handle) = handle.take() && let Err(payload) = handle.join() { log_thread_panic(payload, "group commit thread panicked during drop"); } }); }}
fn initialize_active_state<S: StorageIO>( manager: &DataFileManager<S>, cursor: Option<WriteCursor>, file_ids: &FileIdAllocator,) -> Result<ActiveState, CommitError> { let data_dir = manager.data_dir();
match cursor { Some(wc) => { let handle = manager.open_for_append(wc.file_id)?; let fd = handle.fd(); let file_size = manager.io().file_size(fd)?;
if file_size < wc.offset.raw() { return Err(CommitError::from(io::Error::new( io::ErrorKind::InvalidData, "data file smaller than write cursor", ))); }
let header_end = super::data_file::BLOCK_HEADER_SIZE as u64; let position = match file_size < header_end { true => { let writer = DataFileWriter::new(manager.io(), fd, wc.file_id)?; writer.sync()?; writer.position() } false => BlockOffset::new(file_size), };
let hint_path = hint_file_path(data_dir, wc.file_id); let hint_fd = manager.io().open(&hint_path, OpenOptions::read_write())?; let hint_size = manager.io().file_size(hint_fd)?; let aligned_hint = hint_size - hint_size % HINT_RECORD_SIZE as u64; if aligned_hint != hint_size { manager.io().truncate(hint_fd, aligned_hint)?; manager.io().sync(hint_fd)?; }
Ok(ActiveState { file_id: wc.file_id, fd, position, hint_fd, hint_position: HintOffset::new(aligned_hint), }) } None => { let file_id = file_ids.allocate();
let handle = manager.open_for_append(file_id)?; let fd = handle.fd(); let writer = DataFileWriter::new(manager.io(), fd, file_id)?; writer.sync()?; let position = writer.position();
let hint_path = hint_file_path(data_dir, file_id); let hint_fd = manager.io().open(&hint_path, OpenOptions::read_write())?;
manager.io().sync_dir(data_dir)?;
Ok(ActiveState { file_id, fd, position, hint_fd, hint_position: HintOffset::new(0), }) } }}
enum BatchEntry { Put { blocks: Vec<([u8; CID_SIZE], Vec<u8>)>, response: PutResponse, }, Apply { blocks: Vec<([u8; CID_SIZE], Vec<u8>)>, deleted_cids: Vec<[u8; CID_SIZE]>, response: ApplyResponse, },}
enum ClassifyResult { Batch(BatchEntry), Shutdown, Compact { file_id: DataFileId, grace_period_ms: u64, response: CompactResponse, }, Repair { leaked_cids: Vec<(CidBytes, RefCount)>, response: RepairResponse, }, RepairBlocks { blocks: RepairBlockSet, response: RepairBlocksResponse, }, Quiesce { response: QuiesceResponse, resume: QuiesceResume, },}
fn classify_request(req: CommitRequest) -> ClassifyResult { match req { CommitRequest::PutBlocks { blocks, response } => { ClassifyResult::Batch(BatchEntry::Put { blocks, response }) } CommitRequest::ApplyCommit { blocks, deleted_cids, response, } => ClassifyResult::Batch(BatchEntry::Apply { blocks, deleted_cids, response, }), CommitRequest::Compact { file_id, grace_period_ms, response, } => ClassifyResult::Compact { file_id, grace_period_ms, response, }, CommitRequest::RepairLeaked { leaked_cids, response, } => ClassifyResult::Repair { leaked_cids, response, }, CommitRequest::RepairBlocks { blocks, response } => { ClassifyResult::RepairBlocks { blocks, response } } CommitRequest::Quiesce { response, resume } => ClassifyResult::Quiesce { response, resume }, CommitRequest::Shutdown => ClassifyResult::Shutdown, }}
fn batch_entry_block_count(entry: &BatchEntry) -> usize { match entry { BatchEntry::Put { blocks, .. } | BatchEntry::Apply { blocks, .. } => blocks.len(), }}
struct DrainResult { entries: Vec<BatchEntry>, shutdown: bool, deferred_compacts: Vec<(DataFileId, u64, CompactResponse)>, deferred_repairs: Vec<(Vec<(CidBytes, RefCount)>, RepairResponse)>, deferred_repair_blocks: Vec<DeferredRepairBlock>, deferred_quiesces: Vec<(QuiesceResponse, QuiesceResume)>,}
fn drain_batch( receiver: &flume::Receiver<CommitRequest>, first: CommitRequest, max_batch_size: usize,) -> DrainResult { let first_entry = match classify_request(first) { ClassifyResult::Shutdown => { return DrainResult { entries: Vec::new(), shutdown: true, deferred_compacts: Vec::new(), deferred_repairs: Vec::new(), deferred_repair_blocks: Vec::new(), deferred_quiesces: Vec::new(), }; } ClassifyResult::Compact { file_id, grace_period_ms, response, } => { return DrainResult { entries: Vec::new(), shutdown: false, deferred_compacts: vec![(file_id, grace_period_ms, response)], deferred_repairs: Vec::new(), deferred_repair_blocks: Vec::new(), deferred_quiesces: Vec::new(), }; } ClassifyResult::Repair { leaked_cids, response, } => { return DrainResult { entries: Vec::new(), shutdown: false, deferred_compacts: Vec::new(), deferred_repairs: vec![(leaked_cids, response)], deferred_repair_blocks: Vec::new(), deferred_quiesces: Vec::new(), }; } ClassifyResult::RepairBlocks { blocks, response } => { return DrainResult { entries: Vec::new(), shutdown: false, deferred_compacts: Vec::new(), deferred_repairs: Vec::new(), deferred_repair_blocks: vec![(blocks, response)], deferred_quiesces: Vec::new(), }; } ClassifyResult::Quiesce { response, resume } => { return DrainResult { entries: Vec::new(), shutdown: false, deferred_compacts: Vec::new(), deferred_repairs: Vec::new(), deferred_repair_blocks: Vec::new(), deferred_quiesces: vec![(response, resume)], }; } ClassifyResult::Batch(entry) => entry, };
let mut block_count = batch_entry_block_count(&first_entry); let mut result = DrainResult { entries: vec![first_entry], shutdown: false, deferred_compacts: Vec::new(), deferred_repairs: Vec::new(), deferred_repair_blocks: Vec::new(), deferred_quiesces: Vec::new(), };
let ingest = |req: CommitRequest, bc: &mut usize, r: &mut DrainResult| -> bool { match classify_request(req) { ClassifyResult::Shutdown => true, ClassifyResult::Compact { file_id, grace_period_ms, response, } => { r.deferred_compacts .push((file_id, grace_period_ms, response)); false } ClassifyResult::Repair { leaked_cids, response, } => { r.deferred_repairs.push((leaked_cids, response)); false } ClassifyResult::RepairBlocks { blocks, response } => { r.deferred_repair_blocks.push((blocks, response)); false } ClassifyResult::Quiesce { response, resume } => { r.deferred_quiesces.push((response, resume)); false } ClassifyResult::Batch(entry) => { *bc = bc.saturating_add(batch_entry_block_count(&entry)); r.entries.push(entry); false } } };
while block_count < max_batch_size { match receiver.try_recv() { Ok(req) => { if ingest(req, &mut block_count, &mut result) { result.shutdown = true; break; } } Err(_) => break, } }
result}
fn capture_snapshot<S: StorageIO>( manager: &DataFileManager<S>, state: &ActiveState, epoch: &EpochCounter,) -> BlockstoreSnapshot { BlockstoreSnapshot { shard_cursors: vec![WriteCursor { file_id: state.file_id, offset: state.position, }], epoch: epoch.current(), data_files: manager.list_files().unwrap_or_default(), }}
fn handle_quiesce<S: StorageIO>( manager: &DataFileManager<S>, state: &ActiveState, epoch: &EpochCounter, response: QuiesceResponse, resume: QuiesceResume,) { let snapshot = capture_snapshot(manager, state, epoch); let _ = response.send(snapshot); let _ = resume.blocking_recv();}
fn maybe_checkpoint<C: Clock>( index: &BlockIndex, epoch: &EpochCounter, config: &GroupCommitConfig, clock: &C, last_checkpoint: &mut LogicalNanos, writes_since_checkpoint: &mut u64, hint_positions: &ShardHintPositions,) { let interval = LogicalNanos::from_millis(config.checkpoint_interval_ms); let elapsed = clock.monotonic().saturating_sub(*last_checkpoint) >= interval; let threshold = *writes_since_checkpoint >= config.checkpoint_write_threshold; if !elapsed && !threshold { return; } match index.write_checkpoint(epoch.current(), hint_positions) { Ok(()) => { *last_checkpoint = clock.monotonic(); *writes_since_checkpoint = 0; tracing::debug!("periodic checkpoint written"); } Err(e) => { tracing::warn!(error = %e, "periodic checkpoint failed"); } }}
fn shutdown_checkpoint( index: &BlockIndex, epoch: &EpochCounter, hint_positions: &ShardHintPositions,) { match index.write_checkpoint(epoch.current(), hint_positions) { Ok(()) => tracing::debug!("shutdown checkpoint written"), Err(e) => tracing::warn!(error = %e, "shutdown checkpoint failed"), }}
fn commit_loop<S: StorageIO, C: Clock>( manager: &DataFileManager<S>, index: &BlockIndex, receiver: &flume::Receiver<CommitRequest>, config: &GroupCommitConfig, state: &mut ActiveState, post_sync_hook: Option<&dyn PostBlockstoreHook>, ctx: &ShardContext<C>,) { let epoch = &ctx.epoch; let mut last_checkpoint = ctx.clock.monotonic(); let mut writes_since_checkpoint: u64 = 0;
loop { let first = match receiver.recv() { Ok(CommitRequest::Shutdown) => { shutdown_checkpoint(index, epoch, &ctx.hint_positions); return; } Ok(CommitRequest::Compact { file_id, grace_period_ms, response, }) => { let result = compaction::compact_on_writer_thread( manager, index, file_id, epoch.current(), grace_period_ms, &ctx.file_ids, &ctx.active_files, &ctx.hint_positions, epoch, ctx.clock.wall_millis(), ); let _ = response.send(result); continue; } Ok(CommitRequest::RepairLeaked { leaked_cids, response, }) => { let repaired = index.repair_leaked_refcounts( &leaked_cids, epoch.current(), ctx.clock.wall_millis(), ); let _ = response.send(Ok(repaired)); continue; } Ok(CommitRequest::RepairBlocks { blocks, response }) => { let result = compaction::repair_blocks_on_writer_thread( manager, index, &blocks, epoch.current(), &ctx.file_ids, &ctx.hint_positions, epoch, ); let _ = response.send(result); continue; } Ok(CommitRequest::Quiesce { response, resume }) => { handle_quiesce(manager, state, epoch, response, resume); continue; } Ok(msg) => msg, Err(_) => { shutdown_checkpoint(index, epoch, &ctx.hint_positions); return; } };
let drain_start = std::time::Instant::now(); let drain = drain_batch(receiver, first, config.max_batch_size); let drain_us = drain_start.elapsed().as_nanos() as u64 / 1000;
if !drain.entries.is_empty() { tracing::debug!( batch_size = drain.entries.len(), drain_us, file_id = %state.file_id, "processing commit batch" );
let result = process_batch(manager, index, &drain.entries, state, ctx);
if let Ok((ref _dedup, ref proof)) = result { run_post_sync_hook(post_sync_hook, proof); }
if let Err(ref e) = result { tracing::warn!(error = %e, "commit batch failed"); }
if let Ok((ref dedup, _)) = result { writes_since_checkpoint = writes_since_checkpoint.saturating_add(dedup.len() as u64); }
dispatch_responses(drain.entries, result.map(|(dedup, _proof)| dedup)); }
drain .deferred_compacts .into_iter() .for_each(|(file_id, grace_period_ms, response)| { let result = compaction::compact_on_writer_thread( manager, index, file_id, epoch.current(), grace_period_ms, &ctx.file_ids, &ctx.active_files, &ctx.hint_positions, epoch, ctx.clock.wall_millis(), ); let _ = response.send(result); });
drain .deferred_repairs .into_iter() .for_each(|(leaked_cids, response)| { let repaired = index.repair_leaked_refcounts( &leaked_cids, epoch.current(), ctx.clock.wall_millis(), ); let _ = response.send(Ok(repaired)); });
drain .deferred_repair_blocks .into_iter() .for_each(|(blocks, response)| { let result = compaction::repair_blocks_on_writer_thread( manager, index, &blocks, epoch.current(), &ctx.file_ids, &ctx.hint_positions, epoch, ); let _ = response.send(result); });
maybe_checkpoint( index, epoch, config, &ctx.clock, &mut last_checkpoint, &mut writes_since_checkpoint, &ctx.hint_positions, );
if drain.shutdown { drain_and_process_remaining(manager, index, receiver, state, post_sync_hook, ctx); return; }
drain .deferred_quiesces .into_iter() .for_each(|(response, resume)| { handle_quiesce(manager, state, epoch, response, resume); }); }}
#[allow(clippy::too_many_arguments)]fn handle_request<S: StorageIO, C: Clock>( req: CommitRequest, manager: &DataFileManager<S>, index: &BlockIndex, config: &GroupCommitConfig, state: &mut ActiveState, post_sync_hook: Option<&dyn PostBlockstoreHook>, ctx: &ShardContext<C>, last_checkpoint: &mut LogicalNanos, writes_since_checkpoint: &mut u64,) { let epoch = &ctx.epoch; match classify_request(req) { ClassifyResult::Shutdown => {} ClassifyResult::Compact { file_id, grace_period_ms, response, } => { let result = compaction::compact_on_writer_thread( manager, index, file_id, epoch.current(), grace_period_ms, &ctx.file_ids, &ctx.active_files, &ctx.hint_positions, epoch, ctx.clock.wall_millis(), ); let _ = response.send(result); } ClassifyResult::Repair { leaked_cids, response, } => { let repaired = index.repair_leaked_refcounts( &leaked_cids, epoch.current(), ctx.clock.wall_millis(), ); let _ = response.send(Ok(repaired)); } ClassifyResult::RepairBlocks { blocks, response } => { let result = compaction::repair_blocks_on_writer_thread( manager, index, &blocks, epoch.current(), &ctx.file_ids, &ctx.hint_positions, epoch, ); let _ = response.send(result); } ClassifyResult::Quiesce { response, resume } => { let snapshot = capture_snapshot(manager, state, epoch); let _ = response.send(snapshot); drop(resume); } ClassifyResult::Batch(entry) => { let entries = vec![entry]; let result = process_batch(manager, index, &entries, state, ctx); if let Ok((ref _dedup, ref proof)) = result { run_post_sync_hook(post_sync_hook, proof); } if let Err(ref e) = result { tracing::warn!(error = %e, "commit batch failed"); } if let Ok((ref dedup, _)) = result { *writes_since_checkpoint = writes_since_checkpoint.saturating_add(dedup.len() as u64); } dispatch_responses(entries, result.map(|(dedup, _proof)| dedup)); maybe_checkpoint( index, epoch, config, &ctx.clock, last_checkpoint, writes_since_checkpoint, &ctx.hint_positions, ); } }}
fn run_post_sync_hook(hook: Option<&dyn PostBlockstoreHook>, proof: &BlocksSynced) { if let Some(hook) = hook && let Err(e) = hook.on_blocks_synced(proof) { tracing::error!(error = %e, "post-blockstore sync hook failed"); }}
fn drain_and_process_remaining<S: StorageIO, C: Clock>( manager: &DataFileManager<S>, index: &BlockIndex, receiver: &flume::Receiver<CommitRequest>, state: &mut ActiveState, post_sync_hook: Option<&dyn PostBlockstoreHook>, ctx: &ShardContext<C>,) { let epoch = &ctx.epoch; let mut entries: Vec<BatchEntry> = Vec::new(); let mut compacts: Vec<(DataFileId, u64, CompactResponse)> = Vec::new(); let mut repairs: Vec<(Vec<(CidBytes, RefCount)>, RepairResponse)> = Vec::new(); let mut repair_blocks: Vec<DeferredRepairBlock> = Vec::new();
std::iter::from_fn(|| receiver.try_recv().ok()).for_each(|req| match classify_request(req) { ClassifyResult::Batch(entry) => entries.push(entry), ClassifyResult::Compact { file_id, grace_period_ms, response, } => compacts.push((file_id, grace_period_ms, response)), ClassifyResult::Repair { leaked_cids, response, } => repairs.push((leaked_cids, response)), ClassifyResult::RepairBlocks { blocks, response } => repair_blocks.push((blocks, response)), ClassifyResult::Shutdown | ClassifyResult::Quiesce { .. } => {} });
if !entries.is_empty() { let result = process_batch(manager, index, &entries, state, ctx);
if let Ok((ref _dedup, ref proof)) = result { run_post_sync_hook(post_sync_hook, proof); }
dispatch_responses(entries, result.map(|(dedup, _proof)| dedup)); }
compacts .into_iter() .for_each(|(file_id, grace_period_ms, response)| { let result = compaction::compact_on_writer_thread( manager, index, file_id, epoch.current(), grace_period_ms, &ctx.file_ids, &ctx.active_files, &ctx.hint_positions, epoch, ctx.clock.wall_millis(), ); let _ = response.send(result); });
repairs.into_iter().for_each(|(leaked_cids, response)| { let repaired = index.repair_leaked_refcounts(&leaked_cids, epoch.current(), ctx.clock.wall_millis()); let _ = response.send(Ok(repaired)); });
repair_blocks.into_iter().for_each(|(blocks, response)| { let result = compaction::repair_blocks_on_writer_thread( manager, index, &blocks, epoch.current(), &ctx.file_ids, &ctx.hint_positions, epoch, ); let _ = response.send(result); });
shutdown_checkpoint(index, epoch, &ctx.hint_positions);}
struct RotationState<S: StorageIO> { file_id: DataFileId, handle: Arc<super::manager::CachedHandle<S>>, hint_fd: FileId,}
fn verify_persisted_blocks<S: StorageIO>( manager: &DataFileManager<S>, entries: &[([u8; CID_SIZE], BlockLocation)],) -> Result<(), CommitError> { use std::collections::BTreeMap; let by_file: BTreeMap<DataFileId, Vec<(&[u8; CID_SIZE], BlockLocation)>> = entries.iter().fold(BTreeMap::new(), |mut acc, (cid, loc)| { acc.entry(loc.file_id).or_default().push((cid, *loc)); acc });
by_file.into_iter().try_for_each(|(file_id, locations)| { let path = manager.data_file_path(file_id); let probe_offset = locations[0].1.offset; let fd = match (0..VERIFY_RETRY_ATTEMPTS).find_map(|_| { manager .io() .open(&path, OpenOptions::read_only_existing()) .ok() }) { Some(fd) => fd, None => { return Err(CommitError::VerifyFailed { file_id, offset: probe_offset, }); } }; let file_size = match (0..VERIFY_RETRY_ATTEMPTS).find_map(|_| manager.io().file_size(fd).ok()) { Some(s) => s, None => { let _ = manager.io().close(fd); return Err(CommitError::VerifyFailed { file_id, offset: probe_offset, }); } }; let result = locations.into_iter().try_for_each(|(expected_cid, loc)| { verify_block_at(manager, fd, file_size, expected_cid, loc) }); let _ = manager.io().close(fd); result })}
#[derive(Debug)]enum VerifyOutcome { NoFaultDetected, Faulted, Inconclusive,}
fn verify_block_at<S: StorageIO>( manager: &DataFileManager<S>, fd: FileId, file_size: u64, expected_cid: &[u8; CID_SIZE], loc: BlockLocation,) -> Result<(), CommitError> { let verdict = (0..VERIFY_RETRY_ATTEMPTS).find_map(|_| { match verify_once(manager, fd, file_size, expected_cid, loc) { VerifyOutcome::NoFaultDetected => Some(true), VerifyOutcome::Faulted => Some(false), VerifyOutcome::Inconclusive => None, } }); match verdict { Some(true) => Ok(()), Some(false) | None => Err(CommitError::VerifyFailed { file_id: loc.file_id, offset: loc.offset, }), }}
fn verify_once<S: StorageIO>( manager: &DataFileManager<S>, fd: FileId, file_size: u64, expected_cid: &[u8; CID_SIZE], loc: BlockLocation,) -> VerifyOutcome { match decode_block_record(manager.io(), fd, loc.offset, file_size) { Ok(Some(ReadBlockRecord::Valid { cid_bytes, .. })) if cid_bytes == *expected_cid => { VerifyOutcome::NoFaultDetected } Ok(Some(ReadBlockRecord::Valid { .. })) => { tracing::warn!( file_id = %loc.file_id, offset = loc.offset.raw(), "verify: stored CID mismatch (misdirected write)" ); VerifyOutcome::Faulted } Ok(Some(ReadBlockRecord::Corrupted { .. } | ReadBlockRecord::Truncated { .. })) | Ok(None) => { tracing::warn!( file_id = %loc.file_id, offset = loc.offset.raw(), "verify: block undecodable at location" ); VerifyOutcome::Faulted } Err(_) => VerifyOutcome::Inconclusive, }}
const VERIFY_RETRY_ATTEMPTS: u32 = 4;
const ROLLBACK_TRUNCATE_ATTEMPTS: u32 = 8;
fn truncate_and_sync_durably<S: StorageIO>(io: &S, fd: FileId, offset: u64) { let ok = (0..ROLLBACK_TRUNCATE_ATTEMPTS) .any(|_| io.truncate(fd, offset).and_then(|()| io.sync(fd)).is_ok()); if !ok { tracing::error!( offset, "rollback could not durably truncate uncommitted tail; recovery may resurrect it" ); }}
fn rollback_batch<S: StorageIO>( manager: &DataFileManager<S>, state: &ActiveState, rotations: &[RotationState<S>],) { truncate_and_sync_durably(manager.io(), state.fd, state.position.raw()); truncate_and_sync_durably(manager.io(), state.hint_fd, state.hint_position.raw()); let _ = manager.io().barrier(); rotations.iter().for_each(|rot| { manager.rollback_rotation(rot.file_id); let _ = manager.io().close(rot.hint_fd); let _ = manager .io() .delete(&hint_file_path(manager.data_dir(), rot.file_id)); });}
fn process_batch<S: StorageIO, C: Clock>( manager: &DataFileManager<S>, index: &BlockIndex, batch: &[BatchEntry], state: &mut ActiveState, ctx: &ShardContext<C>,) -> Result<(HashMap<[u8; CID_SIZE], BlockLocation>, BlocksSynced), CommitError> { let epoch = &ctx.epoch; let batch_start = std::time::Instant::now();
let mut dedup: HashMap<[u8; CID_SIZE], BlockLocation> = HashMap::new(); let mut index_entries: Vec<([u8; CID_SIZE], BlockLocation)> = Vec::new(); let mut all_decrements: Vec<[u8; CID_SIZE]> = Vec::new();
let mut current_hint_fd = state.hint_fd; let mut rotations: Vec<RotationState<S>> = Vec::new();
let mut data_writer = DataFileWriter::resume(manager.io(), state.fd, state.file_id, state.position); let mut hint_writer = HintFileWriter::resume(manager.io(), current_hint_fd, state.hint_position);
if manager.should_rotate(data_writer.position()) { data_writer.sync().map_err(CommitError::from)?; hint_writer.sync().map_err(CommitError::from)?;
let next_id = ctx.file_ids.allocate(); let next_handle = manager.open_for_append(next_id)?; let next_fd = next_handle.fd();
tracing::info!( from = %data_writer.file_id(), to = %next_id, trigger = "batch_boundary", "data file rotation" );
data_writer = DataFileWriter::new(manager.io(), next_fd, next_id)?;
let new_hint_path = hint_file_path(manager.data_dir(), next_id); let new_hint_fd = manager .io() .open(&new_hint_path, OpenOptions::read_write())?;
manager.io().sync_dir(manager.data_dir())?;
current_hint_fd = new_hint_fd; hint_writer = HintFileWriter::new(manager.io(), new_hint_fd); rotations.push(RotationState { file_id: next_id, handle: next_handle, hint_fd: new_hint_fd, }); }
let mut block_bytes: u64 = 0; let mut block_count: u64 = 0; let mut dedup_hits: u64 = 0;
let write_result: Result<(), CommitError> = batch.iter().try_for_each(|entry| { let (blocks, decrements) = match entry { BatchEntry::Put { blocks, .. } => (blocks.as_slice(), None), BatchEntry::Apply { blocks, deleted_cids, .. } => (blocks.as_slice(), Some(deleted_cids.as_slice())), };
blocks.iter().try_for_each(|(cid_bytes, data)| { let location = match dedup.get(cid_bytes) { Some(&loc) => { dedup_hits = dedup_hits.saturating_add(1); hint_writer.append_hint(cid_bytes, &loc)?; loc } None => match index.get(cid_bytes) { Some(existing) => { dedup_hits = dedup_hits.saturating_add(1); let loc = existing.location; hint_writer.append_hint(cid_bytes, &loc)?; dedup.insert(*cid_bytes, loc); loc } None => { let loc = data_writer.append_block(cid_bytes, data)?; hint_writer.append_hint(cid_bytes, &loc)?;
block_bytes = block_bytes.saturating_add(data.len() as u64); block_count = block_count.saturating_add(1); dedup.insert(*cid_bytes, loc); loc } }, };
index_entries.push((*cid_bytes, location)); Ok::<_, CommitError>(()) })?;
if let Some(decs) = decrements { all_decrements.extend_from_slice(decs); }
Ok::<_, CommitError>(()) });
if let Err(e) = write_result { rollback_batch(manager, state, &rotations); return Err(e); }
let write_nanos = batch_start.elapsed().as_nanos() as u64;
let current_epoch = epoch.current(); let now = ctx.clock.wall_millis();
let rollback_on_err = |e: CommitError| -> CommitError { rollback_batch(manager, state, &rotations); e };
all_decrements .iter() .try_for_each(|cid| hint_writer.append_decrement(cid, current_epoch, now)) .map_err(|e| rollback_on_err(CommitError::from(e)))?;
let t = std::time::Instant::now(); data_writer.sync().map_err(|e| rollback_on_err(e.into()))?; if ctx.verify_persisted_blocks { verify_persisted_blocks(manager, &index_entries).map_err(rollback_on_err)?; } let batch_record_count = u32::try_from( block_count .saturating_add(dedup_hits) .saturating_add(all_decrements.len() as u64), ) .unwrap_or(u32::MAX); hint_writer .append_commit_marker( current_epoch.raw(), batch_record_count, data_writer.file_id(), data_writer.position(), ) .map_err(|e| rollback_on_err(CommitError::from(e)))?; hint_writer.sync().map_err(|e| rollback_on_err(e.into()))?; manager .io() .barrier() .map_err(|e| rollback_on_err(e.into()))?; let sync_nanos = t.elapsed().as_nanos() as u64;
if !rotations.is_empty() { let old_file_id = state.file_id; let old_hint_fd = state.hint_fd; let last_idx = rotations.len() - 1; rotations.iter().enumerate().for_each(|(i, rot)| { if i == last_idx { manager.commit_rotation(rot.file_id, &rot.handle); ctx.active_files.register(ctx.shard_id, rot.file_id); } else { let _ = manager.io().close(rot.hint_fd); manager.evict_handle(rot.file_id); } }); manager.evict_handle(old_file_id); let _ = manager.io().close(old_hint_fd); }
state.file_id = data_writer.file_id(); state.fd = data_writer.fd(); state.position = data_writer.position(); state.hint_fd = current_hint_fd; state.hint_position = hint_writer.position();
let cursor = WriteCursor { file_id: state.file_id, offset: state.position, }; let t = std::time::Instant::now(); index .batch_put_and_advance_position( &index_entries, &all_decrements, cursor, current_epoch, now, super::hash_index::PositionUpdate { hint_positions: &ctx.hint_positions, shard_id: ctx.shard_id, file_id: state.file_id, offset: state.hint_position, }, ) .map_err(CommitError::from)?; let index_nanos = t.elapsed().as_nanos() as u64;
epoch.advance();
let total_nanos = batch_start.elapsed().as_nanos() as u64;
tracing::info!( blocks = block_count, bytes = block_bytes, dedup_hits, decrements = all_decrements.len(), entries = batch.len(), write_us = write_nanos / 1000, sync_us = sync_nanos / 1000, index_us = index_nanos / 1000, total_us = total_nanos / 1000, "commit batch profile" );
Ok((dedup, BlocksSynced::new()))}
fn dispatch_responses( batch: Vec<BatchEntry>, result: Result<HashMap<[u8; CID_SIZE], BlockLocation>, CommitError>,) { match result { Err(e) => { batch.into_iter().for_each(|entry| { let err = e.clone(); match entry { BatchEntry::Put { response, .. } => { let _ = response.send(Err(err)); } BatchEntry::Apply { response, .. } => { let _ = response.send(Err(err)); } } }); } Ok(written) => { batch.into_iter().for_each(|entry| match entry { BatchEntry::Put { blocks, response, .. } => { let result: Result<Vec<BlockLocation>, CommitError> = blocks .iter() .map(|(cid, _)| match written.get(cid) { Some(&loc) => Ok(loc), None => { tracing::error!( ?cid, "committed CID missing from dedup map, this is a bug" ); Err(CommitError::from(io::Error::other( "committed CID missing from dedup map", ))) } }) .collect(); let _ = response.send(result); } BatchEntry::Apply { response, .. } => { let _ = response.send(Ok(())); } }); } }}