turbo_persistence/
write_batch.rs

1use std::{
2    cell::SyncUnsafeCell,
3    io::Write,
4    mem::{replace, take},
5    path::PathBuf,
6    sync::atomic::{AtomicU32, AtomicU64, Ordering},
7};
8
9use anyhow::{Context, Result};
10use byteorder::{BE, WriteBytesExt};
11use either::Either;
12use fs_err::File;
13use parking_lot::Mutex;
14use smallvec::SmallVec;
15use thread_local::ThreadLocal;
16
17use crate::{
18    FamilyConfig, ValueBuffer,
19    collector::Collector,
20    collector_entry::CollectorEntry,
21    compression::{checksum_block, compress_into_buffer},
22    constants::{MAX_MEDIUM_VALUE_SIZE, THREAD_LOCAL_SIZE_SHIFT},
23    db::WriteOperationGuard,
24    key::StoreKey,
25    meta_file::MetaEntryFlags,
26    meta_file_builder::MetaFileBuilder,
27    parallel_scheduler::ParallelScheduler,
28    static_sorted_file_builder::{StaticSortedFileBuilderMeta, write_static_stored_file},
29};
30
31/// A newly created database file (meta, SST, or blob), carrying its on-disk size so commit can sum
32/// written bytes without stat'ing the files afterwards.
33pub(crate) struct NewFile {
34    pub(crate) seq: u32,
35    pub(crate) file: File,
36    /// On-disk size in bytes.
37    pub(crate) size: u64,
38}
39
40/// The thread local state of a `WriteBatch`. `FAMILIES` should fit within a `u32`.
41//
42// NOTE: This type *must* use `usize`, even though the real type used in storage is `u32` because
43// there's no way to cast a `u32` to `usize` when declaring an array without the nightly
44// `min_generic_const_args` feature.
45struct ThreadLocalState<K: StoreKey + Send, const FAMILIES: usize> {
46    /// The collectors for each family.
47    collectors: [Option<Collector<K, THREAD_LOCAL_SIZE_SHIFT>>; FAMILIES],
48    /// The list of new blob files that have been created.
49    new_blob_files: Vec<NewFile>,
50}
51
52const COLLECTOR_SHARDS: usize = 4;
53const COLLECTOR_SHARD_SHIFT: usize =
54    u64::BITS as usize - COLLECTOR_SHARDS.trailing_zeros() as usize;
55
56/// The result of a `WriteBatch::finish` operation.
57pub(crate) struct FinishResult {
58    pub(crate) sequence_number: u32,
59    pub(crate) new_meta_files: Vec<NewFile>,
60    pub(crate) new_sst_files: Vec<NewFile>,
61    pub(crate) new_blob_files: Vec<NewFile>,
62    /// Number of keys written in this batch.
63    pub(crate) keys_written: u64,
64}
65
66enum GlobalCollectorState<K: StoreKey + Send> {
67    /// Initial state. Single collector. Once the collector is full, we switch to sharded mode.
68    Unsharded(Collector<K>),
69    /// Sharded mode.
70    /// We use multiple collectors, and select one based on the first bits of the key hash.
71    Sharded([Collector<K>; COLLECTOR_SHARDS]),
72}
73
74/// A write batch.
75pub struct WriteBatch<'db, K: StoreKey + Send, S: ParallelScheduler, const FAMILIES: usize> {
76    /// RAII guard that releases the write-operation slot (and rolls back on failure) on drop.
77    _write_guard: WriteOperationGuard<'db>,
78    /// Parallel scheduler
79    parallel_scheduler: S,
80    /// The database path
81    db_path: PathBuf,
82    /// Per-family configuration (kind: SingleValue/MultiValue).
83    #[cfg_attr(not(feature = "verify_sst_content"), allow(dead_code))]
84    family_configs: [FamilyConfig; FAMILIES],
85    /// The current sequence number counter. Increased for every new SST file or blob file.
86    current_sequence_number: AtomicU32,
87    /// The thread local state.
88    thread_locals: ThreadLocal<SyncUnsafeCell<ThreadLocalState<K, FAMILIES>>>,
89    /// Collectors in use. The thread local collectors flush into these when they are full.
90    collectors: [Mutex<GlobalCollectorState<K>>; FAMILIES],
91    /// Meta file builders for each family.
92    meta_collectors: [Mutex<Vec<(u32, StaticSortedFileBuilderMeta<'static>)>>; FAMILIES],
93    /// The list of new SST files that have been created.
94    new_sst_files: Mutex<Vec<NewFile>>,
95}
96
97impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize>
98    WriteBatch<'db, K, S, FAMILIES>
99{
100    /// Creates a new write batch for a database with per-family configuration.
101    pub(crate) fn new(
102        write_guard: WriteOperationGuard<'db>,
103        path: PathBuf,
104        current: u32,
105        parallel_scheduler: S,
106        family_configs: [FamilyConfig; FAMILIES],
107    ) -> Self {
108        const {
109            assert!(FAMILIES <= usize_from_u32(u32::MAX));
110        };
111        Self {
112            _write_guard: write_guard,
113            parallel_scheduler,
114            db_path: path,
115            family_configs,
116            current_sequence_number: AtomicU32::new(current),
117            thread_locals: ThreadLocal::new(),
118            collectors: [(); FAMILIES]
119                .map(|_| Mutex::new(GlobalCollectorState::Unsharded(Collector::new()))),
120            meta_collectors: [(); FAMILIES].map(|_| Mutex::new(Vec::new())),
121            new_sst_files: Mutex::new(Vec::new()),
122        }
123    }
124
125    /// Marks the write operation as successfully completed.
126    ///
127    /// Must be called before dropping the `WriteBatch` to skip the rollback in the guard's `Drop`
128    /// impl. Typically called by `TurboPersistence::commit_write_batch` after a successful commit.
129    pub(crate) fn mark_succeeded(&mut self) {
130        self._write_guard.success();
131    }
132
133    /// Returns the thread local state for the current thread.
134    #[allow(clippy::mut_from_ref)]
135    fn thread_local_state(&self) -> &mut ThreadLocalState<K, FAMILIES> {
136        let cell = self.thread_locals.get_or(|| {
137            SyncUnsafeCell::new(ThreadLocalState {
138                collectors: [const { None }; FAMILIES],
139                new_blob_files: Vec::new(),
140            })
141        });
142        // Safety: We know that the cell is only accessed from the current thread.
143        unsafe { &mut *cell.get() }
144    }
145
146    /// Returns the collector for a family for the current thread.
147    fn thread_local_collector_mut<'l>(
148        &self,
149        state: &'l mut ThreadLocalState<K, FAMILIES>,
150        family: u32,
151    ) -> Result<&'l mut Collector<K, THREAD_LOCAL_SIZE_SHIFT>> {
152        debug_assert!(usize_from_u32(family) < FAMILIES);
153        let collector =
154            state.collectors[usize_from_u32(family)].get_or_insert_with(|| Collector::new());
155        if collector.is_full() {
156            self.flush_thread_local_collector(family, collector)?;
157        }
158        Ok(collector)
159    }
160
161    #[tracing::instrument(level = "trace", skip(self, collector), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
162    fn flush_thread_local_collector(
163        &self,
164        family: u32,
165        collector: &mut Collector<K, THREAD_LOCAL_SIZE_SHIFT>,
166    ) -> Result<()> {
167        let mut full_collectors = SmallVec::<[_; 2]>::new();
168        {
169            let mut global_collector_state = self.collectors[usize_from_u32(family)].lock();
170            for entry in collector.drain() {
171                match &mut *global_collector_state {
172                    GlobalCollectorState::Unsharded(collector) => {
173                        collector.add_entry(entry);
174                        if collector.is_full() {
175                            // When full, split the entries into shards.
176                            let mut shards: [Collector<K>; 4] =
177                                [(); COLLECTOR_SHARDS].map(|_| Collector::new());
178                            for entry in collector.drain() {
179                                let shard = (entry.key.hash >> COLLECTOR_SHARD_SHIFT) as usize;
180                                shards[shard].add_entry(entry);
181                            }
182                            // There is a rare edge case where all entries are in the same shard,
183                            // and the collector is full after the split.
184                            for collector in shards.iter_mut() {
185                                if collector.is_full() {
186                                    full_collectors
187                                        .push(replace(&mut *collector, Collector::new()));
188                                }
189                            }
190                            *global_collector_state = GlobalCollectorState::Sharded(shards);
191                        }
192                    }
193                    GlobalCollectorState::Sharded(shards) => {
194                        let shard = (entry.key.hash >> COLLECTOR_SHARD_SHIFT) as usize;
195                        let collector = &mut shards[shard];
196                        collector.add_entry(entry);
197                        if collector.is_full() {
198                            full_collectors.push(replace(&mut *collector, Collector::new()));
199                        }
200                    }
201                }
202            }
203        }
204        // After flushing write all the full global collectors to disk.
205        // TODO: This can distribute work unfairly
206        // * a thread could fill up multiple global collectors and then get stuck writing them all
207        //   out, if multiple threads could work on it we could take care of spare IO parallism
208        // * we can also have too much IO parallism with many threads concurrently writing files.
209        //
210        // Ideally we would limit the amount of data buffered in memory and control the amount of IO
211        // parallism.  Consider:
212        // * store full-buffers as a field on WireBatch (queued writes)
213        // * each thread will attempt to poll and flush a full buffer after flushing its local
214        //   buffer.
215        // This will distribute the writing work more fairly, but now we have the problem of to
216        // many concurrent writes contending for filesystem locks.  So we could also use a semaphore
217        // to restrict how many concurrent writes occur.  But then we would accumulate 'fullBuffers'
218        // leading to too much memory consumption.  So really we also need to slow down the threads
219        // submitting work data.  To do this we could simply use a tokio semaphore and make all
220        // these operations async, or we could integrate with the parallel::map operation that is
221        // driving the work to slow down task submission in this case.
222        for mut global_collector in full_collectors {
223            // When the global collector is full, we create a new SST file.
224            let sst = self.create_sst_file(
225                family,
226                global_collector.sorted(self.family_configs[usize_from_u32(family)].kind),
227            )?;
228            self.new_sst_files.lock().push(sst);
229            drop(global_collector);
230        }
231        Ok(())
232    }
233
234    /// Puts a key-value pair into the write batch.
235    pub fn put(&self, family: u32, key: K, value: ValueBuffer<'_>) -> Result<()> {
236        let state = self.thread_local_state();
237        let collector = self.thread_local_collector_mut(state, family)?;
238        if value.len() <= MAX_MEDIUM_VALUE_SIZE {
239            collector.put(key, value);
240        } else {
241            let blob = self.create_blob(&value)?;
242            collector.put_blob(key, blob.seq);
243            state.new_blob_files.push(blob);
244        }
245        Ok(())
246    }
247
248    /// Puts a delete operation into the write batch.
249    pub fn delete(&self, family: u32, key: K) -> Result<()> {
250        let state = self.thread_local_state();
251        let collector = self.thread_local_collector_mut(state, family)?;
252        collector.delete(key);
253        Ok(())
254    }
255
256    /// Flushes a family of the write batch, reducing the amount of buffered memory used.
257    /// Does not commit any data persistently.
258    ///
259    /// # Safety
260    ///
261    /// Caller must ensure that no concurrent put or delete operation is happening on the flushed
262    /// family.
263    #[tracing::instrument(level = "trace", skip(self), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
264    pub unsafe fn flush(&self, family: u32) -> Result<()> {
265        // Flush the thread local collectors to the global collector.
266        let mut collectors = Vec::new();
267        for cell in self.thread_locals.iter() {
268            let state = unsafe { &mut *cell.get() };
269            if let Some(collector) = state.collectors[usize_from_u32(family)].take()
270                && !collector.is_empty()
271            {
272                collectors.push(collector);
273            }
274        }
275
276        self.parallel_scheduler
277            .try_parallel_for_each_owned(collectors, |mut collector| {
278                self.flush_thread_local_collector(family, &mut collector)?;
279                drop(collector);
280                anyhow::Ok(())
281            })?;
282
283        // Now we flush the global collector(s).
284        let mut collector_state = self.collectors[usize_from_u32(family)].lock();
285        match &mut *collector_state {
286            GlobalCollectorState::Unsharded(collector) => {
287                if !collector.is_empty() {
288                    let sst = self.create_sst_file(
289                        family,
290                        collector.sorted(self.family_configs[usize_from_u32(family)].kind),
291                    )?;
292                    collector.clear();
293                    self.new_sst_files.lock().push(sst);
294                }
295            }
296            GlobalCollectorState::Sharded(_) => {
297                let GlobalCollectorState::Sharded(mut shards) = replace(
298                    &mut *collector_state,
299                    GlobalCollectorState::Unsharded(Collector::new()),
300                ) else {
301                    unreachable!();
302                };
303                self.parallel_scheduler
304                    .try_parallel_for_each_mut(&mut shards, |collector| {
305                        if !collector.is_empty() {
306                            let sst = self.create_sst_file(
307                                family,
308                                collector.sorted(self.family_configs[usize_from_u32(family)].kind),
309                            )?;
310                            collector.clear();
311                            self.new_sst_files.lock().push(sst);
312                            collector.drop_contents();
313                        }
314                        anyhow::Ok(())
315                    })?;
316            }
317        }
318
319        Ok(())
320    }
321
322    /// Finishes the write batch by returning the new sequence number and the new SST files. This
323    /// writes all outstanding thread local data to disk.
324    #[tracing::instrument(level = "trace", skip_all)]
325    pub(crate) fn finish(
326        &mut self,
327        get_accessed_key_hashes: impl Fn(u32) -> qfilter::Filter + Send + Sync,
328    ) -> Result<FinishResult> {
329        let mut new_blob_files = Vec::new();
330
331        // First, we flush all thread local collectors to the global collectors.
332        {
333            let _span = tracing::trace_span!("flush thread local collectors").entered();
334            let mut collectors = [const { Vec::new() }; FAMILIES];
335            for cell in self.thread_locals.iter_mut() {
336                let state = cell.get_mut();
337                new_blob_files.append(&mut state.new_blob_files);
338                for (family, thread_local_collector) in state.collectors.iter_mut().enumerate() {
339                    if let Some(collector) = thread_local_collector.take()
340                        && !collector.is_empty()
341                    {
342                        collectors[family].push(collector);
343                    }
344                }
345            }
346            let to_flush = collectors
347                .into_iter()
348                .enumerate()
349                .flat_map(|(family, collector)| {
350                    collector
351                        .into_iter()
352                        .map(move |collector| (family as u32, collector))
353                })
354                .collect::<Vec<_>>();
355            self.parallel_scheduler.try_parallel_for_each_owned(
356                to_flush,
357                |(family, mut collector)| {
358                    self.flush_thread_local_collector(family, &mut collector)?;
359                    drop(collector);
360                    anyhow::Ok(())
361                },
362            )?;
363        }
364
365        let _span = tracing::trace_span!("flush collectors").entered();
366
367        // Now we reduce the global collectors in parallel
368        let mut new_sst_files = take(self.new_sst_files.get_mut());
369        let shared_new_sst_files = Mutex::new(&mut new_sst_files);
370
371        let new_collectors =
372            [(); FAMILIES].map(|_| Mutex::new(GlobalCollectorState::Unsharded(Collector::new())));
373        let collectors = replace(&mut self.collectors, new_collectors);
374        let collectors = collectors
375            .into_iter()
376            .enumerate()
377            .flat_map(|(family, state)| {
378                let collector = state.into_inner();
379                match collector {
380                    GlobalCollectorState::Unsharded(collector) => {
381                        Either::Left([(family, collector)].into_iter())
382                    }
383                    GlobalCollectorState::Sharded(shards) => {
384                        Either::Right(shards.into_iter().map(move |collector| (family, collector)))
385                    }
386                }
387            })
388            .collect::<Vec<_>>();
389        self.parallel_scheduler.try_parallel_for_each_owned(
390            collectors,
391            |(family, mut collector)| {
392                let family = family as u32;
393                if !collector.is_empty() {
394                    let sst = self.create_sst_file(
395                        family,
396                        collector.sorted(self.family_configs[usize_from_u32(family)].kind),
397                    )?;
398                    collector.clear();
399                    drop(collector);
400                    shared_new_sst_files.lock().push(sst);
401                }
402                anyhow::Ok(())
403            },
404        )?;
405
406        // Now we need to write the new meta files.
407        let new_meta_collectors = [(); FAMILIES].map(|_| Mutex::new(Vec::new()));
408        let meta_collectors = replace(&mut self.meta_collectors, new_meta_collectors);
409        let keys_written = AtomicU64::new(0);
410        let file_to_write = meta_collectors
411            .into_iter()
412            .map(|mutex| mutex.into_inner())
413            .enumerate()
414            .filter(|(_, sst_files)| !sst_files.is_empty())
415            .collect::<Vec<_>>();
416        let new_meta_files = self
417            .parallel_scheduler
418            .parallel_map_collect_owned::<_, _, Result<Vec<_>>>(
419                file_to_write,
420                |(family, sst_files)| {
421                    let family = family as u32;
422                    let mut entries = 0;
423                    let mut builder = MetaFileBuilder::new(family);
424                    for (seq, sst) in sst_files {
425                        entries += sst.entries;
426                        builder.add(seq, sst);
427                    }
428                    keys_written.fetch_add(entries, Ordering::Relaxed);
429                    let accessed_key_hashes = get_accessed_key_hashes(family);
430                    builder.set_used_key_hashes_amqf(accessed_key_hashes);
431                    let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
432                    let (file, size) = builder.write(&self.db_path, seq)?;
433                    Ok(NewFile { seq, file, size })
434                },
435            )?;
436
437        // Finally we return the new files and sequence number.
438        let seq = self.current_sequence_number.load(Ordering::SeqCst);
439        Ok(FinishResult {
440            sequence_number: seq,
441            new_meta_files,
442            new_sst_files,
443            new_blob_files,
444            keys_written: keys_written.into_inner(),
445        })
446    }
447
448    /// Creates a new blob file with the given value.
449    #[tracing::instrument(level = "trace", skip(self, value), fields(value_len = value.len()))]
450    fn create_blob(&self, value: &[u8]) -> Result<NewFile> {
451        let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
452        let mut compressed = Vec::new();
453        compress_into_buffer(value, &mut compressed)
454            .context("Compression of value for blob file failed")?;
455
456        let mut buffer = Vec::with_capacity(8 + compressed.len());
457        buffer.write_u32::<BE>(value.len() as u32)?;
458        buffer.write_u32::<BE>(checksum_block(&compressed))?;
459        buffer.extend_from_slice(&compressed);
460
461        let size = buffer.len() as u64;
462        let file = self.db_path.join(format!("{seq:08}.blob"));
463        let mut file = File::create(&file)?;
464        file.write_all(&buffer)?;
465        file.flush()?;
466        Ok(NewFile { seq, file, size })
467    }
468
469    /// Creates a new SST file with the given collector data.
470    #[tracing::instrument(level = "trace", skip(self, collector_data), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
471    fn create_sst_file(
472        &self,
473        family: u32,
474        collector_data: (&[CollectorEntry<K>], usize),
475    ) -> Result<NewFile> {
476        let (entries, _total_key_size) = collector_data;
477        let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
478
479        let path = self.db_path.join(format!("{seq:08}.sst"));
480        let (meta, file) = self
481            .parallel_scheduler
482            .block_in_place(|| write_static_stored_file(entries, &path, MetaEntryFlags::FRESH))
483            .with_context(|| format!("Unable to write SST file {seq:08}.sst"))?;
484
485        #[cfg(feature = "verify_sst_content")]
486        {
487            use core::panic;
488
489            use crate::{
490                collector_entry::CollectorEntryValue,
491                key::hash_key,
492                lookup_entry::LookupValue,
493                static_sorted_file::{
494                    BlockCache, SstLookupResult, StaticSortedFile, StaticSortedFileMetaData,
495                },
496                static_sorted_file_builder::Entry,
497            };
498
499            file.sync_all()?;
500            let sst = StaticSortedFile::open(
501                &self.db_path,
502                StaticSortedFileMetaData {
503                    sequence_number: seq,
504                    block_count: meta.block_count,
505                },
506            )?;
507            let cache2 = BlockCache::with(
508                10,
509                u64::MAX,
510                Default::default(),
511                Default::default(),
512                Default::default(),
513            );
514            let cache3 = BlockCache::with(
515                10,
516                u64::MAX,
517                Default::default(),
518                Default::default(),
519                Default::default(),
520            );
521            let mut key_buf = Vec::new();
522            let family_config = self.family_configs[usize_from_u32(family)].kind;
523            for entry in entries {
524                entry.write_key_to(&mut key_buf);
525                let result = sst
526                    .lookup::<_, true>(hash_key(&key_buf), &key_buf, &cache2, &cache3)
527                    .expect("key found");
528                key_buf.clear();
529                match result {
530                    SstLookupResult::Found(values) => {
531                        if values.len() > 1 {
532                            use crate::FamilyKind;
533
534                            assert!(
535                                values.len() == 1 || family_config == FamilyKind::MultiValue,
536                                "only multi-value tables can have more than one value, got {} \
537                                 values",
538                                values.len()
539                            )
540                        }
541                        match &entry.value {
542                            CollectorEntryValue::Large { blob } => {
543                                assert!(
544                                    values.contains(&LookupValue::Blob {
545                                        sequence_number: *blob
546                                    }),
547                                    "we wrote a blob but did not read it"
548                                );
549                            }
550                            CollectorEntryValue::Deleted => assert!(
551                                values.first() == Some(&LookupValue::Deleted),
552                                "we wrote a deleted tombstone but it was not first in results"
553                            ),
554                            v => {
555                                assert!(
556                                    values.into_iter().any(|lv| {
557                                        if let LookupValue::Slice { value } = lv {
558                                            &*value == v.as_bytes().unwrap()
559                                        } else {
560                                            false
561                                        }
562                                    }),
563                                    "we wrote a slice of bytes but did not read it"
564                                )
565                            }
566                        }
567                    }
568                    SstLookupResult::NotFound => panic!("All keys must exist"),
569                }
570            }
571        }
572
573        let size = meta.size;
574        self.meta_collectors[usize_from_u32(family)]
575            .lock()
576            .push((seq, meta));
577
578        Ok(NewFile { seq, file, size })
579    }
580}
581
582#[inline(always)]
583const fn usize_from_u32(value: u32) -> usize {
584    // This should always be true, as we assume at least a 32-bit width architecture for Turbopack.
585    // Since this is a const expression, we expect it to be compiled away.
586    const {
587        assert!(u32::BITS < usize::BITS);
588    };
589    value as usize
590}
turbo_persistence/write_batch.rs

turbo_persistence/
write_batch.rs
