Append commit instead of individual transactions to commitlog (#4140)

Changes the commitlog (and durability) write API, such that the caller
decides how many transactions are in a single commit, and has to supply
the transaction offsets.

This simplifies commitlog-side buffering logic to essentially a
`BufWriter` (which, of course, we must not forget to flush). This will
help throughput, but offers less opportunity to retry failed writes.
This is probably a good thing, as disks can fail in erratic ways, and we
should rather crash and re-verify the commitlog (suffix) than continue
writing.

To that end, this patch liberally raises panics when there is a chance
that internal state could be "poisoned" by partial writes, which may be
debatable.


# Motivation

The main motivation is to avoid maintaining the transaction offset in
two places in such a way that they could diverge. As ordering commits is
the responsibility of the datastore, we make it authoritative on this
matter -- the commitlog will still check that offsets are contiguous,
and refuse to commit if that's not the case.

A secondary, related motivation is the following:

A "commit" is an atomic unit of storage, meaning that a torn (partial)
write of a commit will render the entire commit corrupt. There hasn't
been a compelling case where we would want this, and have always
configured the server to write exactly one transaction per commit.
The code to handle buffering of transactions is, however, rather
complex, as it tries hard to allow the caller to retry writes at commit
boundaries. An unfortunate consequence of this is that we'd flush to the
OS very often, leaving throughput performance on the table.

So, if there is a compelling case for batching multiple transactions in
a commit, it should be the datastore's responsibility.


# API and ABI breaking changes

Breaks internal APIs only.

# Expected complexity level and risk

5 - Mostly for the risk

# Testing

Existing tests.

Author: kim

crates/commitlog/src/commitlog.rs

@@ -1,6 +1,6 @@
 use std::{
     io,
-    num::{NonZeroU16, NonZeroU64},
+    num::NonZeroU64,
     ops::RangeBounds,
     sync::{Arc, RwLock},
 };
@@ -17,10 +17,11 @@ pub mod segment;
 mod varchar;
 mod varint;
 
+use crate::segment::Committed;
 pub use crate::{
     commit::{Commit, StoredCommit},
     payload::{Decoder, Encode},
-    repo::fs::SizeOnDisk,
+    repo::{fs::SizeOnDisk, TxOffset},
     segment::{Transaction, DEFAULT_LOG_FORMAT_VERSION},
     varchar::Varchar,
 };
@@ -57,14 +58,6 @@ pub struct Options {
     /// Default: 1GiB
     #[cfg_attr(feature = "serde", serde(default = "Options::default_max_segment_size"))]
     pub max_segment_size: u64,
-    /// The maximum number of records in a commit.
-    ///
-    /// If this number is exceeded, the commit is flushed to disk even without
-    /// explicitly calling [`Commitlog::flush`].
-    ///
-    /// Default: 1
-    #[cfg_attr(feature = "serde", serde(default = "Options::default_max_records_in_commit"))]
-    pub max_records_in_commit: NonZeroU16,
     /// Whenever at least this many bytes have been written to the currently
     /// active segment, an entry is added to its offset index.
     ///
@@ -96,6 +89,12 @@ pub struct Options {
     /// Has no effect if the `fallocate` feature is not enabled.
     #[cfg_attr(feature = "serde", serde(default = "Options::default_preallocate_segments"))]
     pub preallocate_segments: bool,
+    /// Size in bytes of the memory buffer holding commit data before flushing
+    /// to storage.
+    ///
+    /// Default: 4KiB
+    #[cfg_attr(feature = "serde", serde(default = "Options::default_write_buffer_size"))]
+    pub write_buffer_size: usize,
 }
 
 impl Default for Options {
@@ -106,18 +105,18 @@ impl Default for Options {
 
 impl Options {
     pub const DEFAULT_MAX_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
-    pub const DEFAULT_MAX_RECORDS_IN_COMMIT: NonZeroU16 = NonZeroU16::new(1).expect("1 > 0, qed");
     pub const DEFAULT_OFFSET_INDEX_INTERVAL_BYTES: NonZeroU64 = NonZeroU64::new(4096).expect("4096 > 0, qed");
     pub const DEFAULT_OFFSET_INDEX_REQUIRE_SEGMENT_FSYNC: bool = false;
     pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
+    pub const DEFAULT_WRITE_BUFFER_SIZE: usize = 4 * 1024;
 
     pub const DEFAULT: Self = Self {
         log_format_version: DEFAULT_LOG_FORMAT_VERSION,
         max_segment_size: Self::default_max_segment_size(),
-        max_records_in_commit: Self::default_max_records_in_commit(),
         offset_index_interval_bytes: Self::default_offset_index_interval_bytes(),
         offset_index_require_segment_fsync: Self::default_offset_index_require_segment_fsync(),
         preallocate_segments: Self::default_preallocate_segments(),
+        write_buffer_size: Self::default_write_buffer_size(),
     };
 
     pub const fn default_log_format_version() -> u8 {
@@ -128,10 +127,6 @@ impl Options {
         Self::DEFAULT_MAX_SEGMENT_SIZE
     }
 
-    pub const fn default_max_records_in_commit() -> NonZeroU16 {
-        Self::DEFAULT_MAX_RECORDS_IN_COMMIT
-    }
-
     pub const fn default_offset_index_interval_bytes() -> NonZeroU64 {
         Self::DEFAULT_OFFSET_INDEX_INTERVAL_BYTES
     }
@@ -144,6 +139,10 @@ impl Options {
         Self::DEFAULT_PREALLOCATE_SEGMENTS
     }
 
+    pub const fn default_write_buffer_size() -> usize {
+        Self::DEFAULT_WRITE_BUFFER_SIZE
+    }
+
     /// Compute the length in bytes of an offset index based on the settings in
     /// `self`.
     pub fn offset_index_len(&self) -> u64 {
@@ -262,7 +261,7 @@ impl<T> Commitlog<T> {
     pub fn flush(&self) -> io::Result<Option<u64>> {
         let mut inner = self.inner.write().unwrap();
         trace!("flush commitlog");
-        inner.commit()?;
+        inner.flush()?;
 
         Ok(inner.max_committed_offset())
     }
@@ -282,7 +281,7 @@ impl<T> Commitlog<T> {
     pub fn flush_and_sync(&self) -> io::Result<Option<u64>> {
         let mut inner = self.inner.write().unwrap();
         trace!("flush and sync commitlog");
-        inner.commit()?;
+        inner.flush()?;
         inner.sync();
 
         Ok(inner.max_committed_offset())
@@ -383,57 +382,47 @@ impl<T> Commitlog<T> {
 }
 
 impl<T: Encode> Commitlog<T> {
-    /// Append the record `txdata` to the log.
+    /// Write `transactions` to the log.
     ///
-    /// If the internal buffer exceeds [`Options::max_records_in_commit`], the
-    /// argument is returned in an `Err`. The caller should [`Self::flush`] the
-    /// log and try again.
+    /// This will store all `transactions` as a single [Commit]
+    /// (note that `transactions` must not yield more than [u16::MAX] elements).
     ///
-    /// In case the log is appended to from multiple threads, this may result in
-    /// a busy loop trying to acquire a slot in the buffer. In such scenarios,
-    /// [`Self::append_maybe_flush`] is preferable.
-    pub fn append(&self, txdata: T) -> Result<(), T> {
-        let mut inner = self.inner.write().unwrap();
-        inner.append(txdata)
-    }
-
-    /// Append the record `txdata` to the log.
+    /// Data is buffered internally, call [Self::flush] to force flushing to
+    /// the underlying storage.
+    ///
+    /// Returns `Ok(None)` if `transactions` was empty, otherwise [Committed],
+    /// which contains the offset range and checksum of the commit.
     ///
-    /// The `txdata` payload is buffered in memory until either:
+    /// # Errors
     ///
-    /// - [`Self::flush`] is called explicitly, or
-    /// - [`Options::max_records_in_commit`] is exceeded
+    /// An `Err` value is returned in the following cases:
     ///
-    /// In the latter case, [`Self::append`] flushes implicitly, _before_
-    /// appending the `txdata` argument.
+    /// - if the transaction sequence is invalid, e.g. because the transaction
+    ///   offsets are not contiguous.
     ///
-    /// I.e. the argument is not guaranteed to be flushed after the method
-    /// returns. If that is desired, [`Self::flush`] must be called explicitly.
+    ///   In this case, **none** of the `transactions` will be written.
     ///
-    /// If writing `txdata` to the commitlog results in a new segment file being opened,
-    /// we will send a message down `on_new_segment`.
-    /// This will be hooked up to the `request_snapshot` channel of a `SnapshotWorker`.
+    /// - if creating the new segment fails due to an I/O error.
     ///
-    /// # Errors
+    /// # Panics
+    ///
+    /// The method panics if:
+    ///
+    /// - `transactions` exceeds [u16::MAX] elements
+    ///
+    /// - [Self::flush] or writing to the underlying buffered writer fails
     ///
-    /// If the log needs to be flushed, but an I/O error occurs, ownership of
-    /// `txdata` is returned back to the caller alongside the [`io::Error`].
+    ///   This is likely caused by some storage issue. As we cannot tell with
+    ///   certainty how much data (if any) has been written, the internal state
+    ///   becomes invalid and thus a panic is raised.
     ///
-    /// The value can then be used to retry appending.
-    pub fn append_maybe_flush(&self, txdata: T) -> Result<(), error::Append<T>> {
+    /// - [Self::sync] panics (called when rotating segments)
+    pub fn commit<U: Into<Transaction<T>>>(
+        &self,
+        transactions: impl IntoIterator<Item = U>,
+    ) -> io::Result<Option<Committed>> {
         let mut inner = self.inner.write().unwrap();
-
-        if let Err(txdata) = inner.append(txdata) {
-            if let Err(source) = inner.commit() {
-                return Err(error::Append { txdata, source });
-            }
-
-            // `inner.commit.n` must be zero at this point
-            let res = inner.append(txdata);
-            debug_assert!(res.is_ok(), "failed to append while holding write lock");
-        }
-
-        Ok(())
+        inner.commit(transactions)
     }
 
     /// Obtain an iterator which traverses the log from the start, yielding

crates/commitlog/src/lib.rs

@@ -214,13 +214,11 @@ pub fn create_segment_writer<R: Repo>(
             records: Vec::new(),
             epoch,
         },
-        inner: io::BufWriter::new(storage),
+        inner: io::BufWriter::with_capacity(opts.write_buffer_size, storage),
 
         min_tx_offset: offset,
         bytes_written: Header::LEN as u64,
 
-        max_records_in_commit: opts.max_records_in_commit,
-
         offset_index_head: create_offset_index_writer(repo, offset, opts),
     })
 }
@@ -293,8 +291,6 @@ pub fn resume_segment_writer<R: Repo>(
         min_tx_offset: tx_range.start,
         bytes_written: size_in_bytes,
 
-        max_records_in_commit: opts.max_records_in_commit,
-
         offset_index_head: create_offset_index_writer(repo, offset, opts),
     }))
 }