mmap-sync: cloudflare 对于低延迟 IPC 的解决方案

              +----------------+
              |   left_right   |
              +----------------+
                       |
        left           |           right
          +------------+-------------+
          |                          |
          v                          v
 +-----------------+        +-----------------+
 |  left_instance  |        | right_instance  |
 +-----------------+        +-----------------+


+----------------------+----------------------+
| read_indicator[0][0] | read_indicator[1][0] |------->  thread_id = 0
+----------------------+----------------------+
| read_indicator[0][1] | read_indicator[1][1] |
+----------------------+----------------------+
| read_indicator[0][2] | read_indicator[1][2] |
+----------------------+----------------------+
| read_indicator[0][3] | read_indicator[1][3] |
+----------------------+----------------------+
            ^                      ^
          0 |                      | 1
            +----------+-----------+
                       |
                       |
              +-----------------+
              |  version_index  |
              +-----------------+

// vim src/synchronizer.rs + 73

/// Writes a given `entity` into the next available data file.
///
/// Returns the number of bytes written to the data file and a boolean flag, for diagnostic
/// purposes, indicating whether the reader counter was reset due to a reader exiting without
/// decrementing it.
///
/// # Parameters
/// - `entity`: The entity to be written to the data file.
/// - `grace_duration`: The maximum period to wait for readers to finish before resetting the
///                     reader count to 0. This handles scenarios where a reader process has
///                     crashed or exited abnormally, failing to decrement the reader count.
///                     After the `grace_duration` has elapsed, if there are still active
///                     readers, the reader count is reset to 0 to restore synchronization state.
///
/// # Returns
/// A result containing a tuple of the number of bytes written and a boolean indicating whether
/// the reader count was reset, or a `SynchronizerError` if the operation fails.
pub fn write<T>(
    &mut self,
    entity: &T,
    grace_duration: Duration,
) -> Result<(usize, bool), SynchronizerError>
where
    T: Serialize<DefaultSerializer>,
    T::Archived: for<'b> CheckBytes<DefaultValidator<'b>>,
{
    // serialize given entity into bytes
    let mut serializer = DefaultSerializer::default();
    let _ = serializer
        .serialize_value(entity)
        .map_err(|_| FailedEntityWrite)?;
    let data = serializer.into_serializer().into_inner();

    // ensure that serialized bytes can be deserialized back to `T` struct successfully
    check_archived_root::<T>(&data).map_err(|_| FailedEntityRead)?;

    // fetch current state from mapped memory
    let state = self.state_container.state(true)?;

    // calculate data checksum
    let mut hasher = SeaHasher::new();
    hasher.write(&data);
    let checksum = hasher.finish();

    // acquire next available data file idx and write data to it
    let (new_idx, reset) = state.acquire_next_idx(grace_duration);
    let new_version = InstanceVersion::new(new_idx, data.len(), checksum)?;
    let size = self.data_container.write(&data, new_version)?;

    // switch readers to new version
    state.switch_version(new_version);

    Ok((size, reset))
}

// vim src/state.rs +117

    /// Fetch state from existing memory mapped file or create new one
    #[inline]
    pub(crate) fn state(&mut self, create: bool) -> Result<&mut State, SynchronizerError> {
        if self.mmap.is_none() {
            let state_file = OpenOptions::new()
                .read(true)
                .write(true)
                .create(create)
                .mode(0o660) // set file mode to allow read/write from owner/group only
                .open(&self.state_path)
                .map_err(FailedStateRead)?;

            let mut need_init = false;
            // Reset state file size to match exactly `STATE_SIZE`
            if state_file.metadata().map_err(FailedStateRead)?.len() as usize != STATE_SIZE {
                state_file
                    .set_len(STATE_SIZE as u64)
                    .map_err(FailedStateRead)?;
                need_init = true;
            }

            let mut mmap = unsafe { MmapMut::map_mut(&state_file).map_err(FailedStateRead)? };
            if need_init {
                // Create new state and write it to mapped memory
                let new_state = State::default();
                unsafe {
                    mmap.as_mut_ptr()
                        .copy_from((&new_state as *const State) as *const u8, STATE_SIZE)
                };
            };

            self.mmap = Some(mmap);
        }
        Ok(unsafe { &mut *(self.mmap.as_ref().unwrap().as_ptr() as *mut State) })
    }

// vim src/state.rs +45

/// Acquire next `idx` of the state for writing
#[inline]
pub(crate) fn acquire_next_idx(&self, grace_duration: Duration) -> (usize, bool) {
    // calculate `next_idx` to acquire, in case of uninitialized version use 0
    let next_idx = match InstanceVersion::try_from(self.version.load(Ordering::SeqCst)) {
        Ok(version) => (version.idx() + 1) % 2,
        Err(_) => 0,
    };

    // check number of readers using `next_idx`
    let num_readers = &self.idx_readers[next_idx];

    // wait until either no more readers left for `next_idx` or grace period has expired
    let grace_expiring_at = Instant::now().add(grace_duration);
    let mut reset = false;
    while num_readers.load(Ordering::SeqCst) > 0 {
        // we should reach here only when one of the readers dies without decrement
        if Instant::now().gt(&grace_expiring_at) {
            // reset number of readers after expired grace period
            num_readers.store(0, Ordering::SeqCst);
            reset = true;
            break;
        } else {
            thread::sleep(SLEEP_DURATION);
        }
    }

    (next_idx, reset)
}

// vim src/data.rs +24

/// Write `data` into mapped data file with given `version`
pub(crate) fn write(
    &mut self,
    data: &[u8],
    version: InstanceVersion,
) -> Result<usize, SynchronizerError> {
    let data_file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .mode(0o640) // set file mode to allow read/write from owner, read from group only
        .open(version.path(&self.base_path))
        .map_err(FailedDataWrite)?;

    // grow data file when its current length exceeded
    let data_len = data.len() as u64;
    if data_len > data_file.metadata().map_err(FailedDataWrite)?.len() {
        data_file.set_len(data_len).map_err(FailedDataWrite)?;
    }

    // copy data to mapped file and ensure it's been flushed
    let mut mmap = unsafe { MmapMut::map_mut(&data_file).map_err(FailedDataWrite)? };
    mmap[..data.len()].copy_from_slice(data);
    mmap.flush().map_err(FailedDataWrite)?;

    Ok(data.len())
}

// vim src/synchronizer.rs +159

/// Reads and returns an `entity` struct from mapped memory wrapped in `ReadGuard`.
///
/// # Safety
///
/// This method is marked as unsafe due to the potential for memory corruption if the returned
/// result is used beyond the `grace_duration` set in the `write` method. The caller must ensure
/// the `ReadGuard` (and any references derived from it) are dropped before this time period
/// elapses to ensure safe operation.
///
/// Additionally, the use of `unsafe` here is related to the internal use of the
/// `rkyv::archived_root` function, which has its own safety considerations. Particularly, it
/// assumes the byte slice provided to it accurately represents an archived object, and that the
/// root of the object is stored at the end of the slice.
pub unsafe fn read<T>(&mut self) -> Result<ReadResult<T>, SynchronizerError>
where
    T: Archive,
    T::Archived: for<'b> CheckBytes<DefaultValidator<'b>>,
{
    // fetch current state from mapped memory
    let state = self.state_container.state(false)?;

    // fetch current version
    let version = state.version()?;

    // create and lock state guard for reading
    let guard = ReadGuard::new(state, version)?;

    // fetch data for current version from mapped memory
    let (data, switched) = self.data_container.data(version)?;

    // fetch entity from data using zero-copy deserialization
    let entity = archived_root::<T>(data);

    Ok(ReadResult::new(guard, entity, switched))
}

// vim src/state.rs +33

    /// Return state's current instance version
    #[inline]
    pub(crate) fn version(&self) -> Result<InstanceVersion, SynchronizerError> {
        self.version.load(Ordering::SeqCst).try_into()
    }

// vim src/guard.rs +22

/// An RAII implementation of a “scoped read lock” of a `State`
pub(crate) struct ReadGuard<'a> {
    state: &'a mut State,
    version: InstanceVersion,
}

impl<'a> ReadGuard<'a> {
    /// Creates new `ReadGuard` with specified parameters
    pub(crate) fn new(
        state: &'a mut State,
        version: InstanceVersion,
    ) -> Result<Self, SynchronizerError> {
        state.rlock(version);
        Ok(ReadGuard { version, state })
    }
}

// vim src/state.rs +39

    /// Locks given `version` of the state for reading
    #[inline]
    pub(crate) fn rlock(&mut self, version: InstanceVersion) {
        self.idx_readers[version.idx()].fetch_add(1, Ordering::SeqCst);
    }

// vim src/state.rs +75

    /// Unlocks given `version` from reading
    #[inline]
    pub(crate) fn runlock(&mut self, version: InstanceVersion) {
        self.idx_readers[version.idx()].fetch_sub(1, Ordering::SeqCst);
    }