pub(crate) fn write_to_log(&self, iobuf: &IoBuf) -> Result<()> {
        let _measure = Measure::new(&M.write_to_log);
        let header = iobuf.get_header();
        let log_offset = iobuf.offset;
        let base_lsn = iobuf.lsn;
        let capacity = iobuf.capacity;

        let segment_size = self.config.segment_size;

        assert_eq!(
            Lsn::try_from(log_offset % segment_size as LogOffset).unwrap(),
            base_lsn % segment_size as Lsn
        );

        assert_ne!(
            log_offset,
            LogOffset::max_value(),
            "created reservation for uninitialized slot",
        );

        assert!(is_sealed(header));

        let bytes_to_write = offset(header);

        trace!(
            "write_to_log log_offset {} lsn {} len {}",
            log_offset,
            base_lsn,
            bytes_to_write
        );

        let maxed = iobuf.linearized(|| iobuf.get_maxed());
        let unused_space = capacity - bytes_to_write;
        let should_pad = maxed && unused_space >= MAX_MSG_HEADER_LEN;

        // a pad is a null message written to the end of a buffer
        // to signify that nothing else will be written into it
        if should_pad {
            let pad_len = capacity - bytes_to_write - MAX_MSG_HEADER_LEN;
            let data = iobuf.get_mut_range(bytes_to_write, pad_len);

            // take the crc of the random bytes already after where we
            // would place our header.
            let padding_bytes = vec![MessageKind::Corrupted.into(); pad_len];

            let segment_number = SegmentNumber(
                u64::try_from(base_lsn).unwrap()
                    / u64::try_from(self.config.segment_size).unwrap(),
            );

            let header = MessageHeader {
                kind: MessageKind::Cap,
                pid: PageId::max_value(),
                segment_number,
                len: u64::try_from(pad_len).unwrap(),
                crc32: 0,
            };

            let header_bytes = header.serialize();

            #[allow(unsafe_code)]
            unsafe {
                std::ptr::copy_nonoverlapping(
                    header_bytes.as_ptr(),
                    data.as_mut_ptr(),
                    header_bytes.len(),
                );
                std::ptr::copy_nonoverlapping(
                    padding_bytes.as_ptr(),
                    data.as_mut_ptr().add(header_bytes.len()),
                    pad_len,
                );
            }

            // this as to stay aligned with the hashing
            let crc32_arr = u32_to_arr(calculate_message_crc32(
                &header_bytes,
                &padding_bytes,
            ));

            #[allow(unsafe_code)]
            unsafe {
                std::ptr::copy_nonoverlapping(
                    crc32_arr.as_ptr(),
                    // the crc32 is the first part of the buffer
                    data.as_mut_ptr(),
                    std::mem::size_of::<u32>(),
                );
            }
        }

        let total_len = if maxed { capacity } else { bytes_to_write };

        let data = iobuf.get_mut_range(0, total_len);
        let stored_max_stable_lsn = iobuf.stored_max_stable_lsn;

        io_fail!(self, "buffer write");
        #[cfg(feature = "io_uring")]
        {
            let mut wrote = 0;
            while wrote < total_len {
                let to_write = &data[wrote..];
                let offset = log_offset + wrote as u64;

                // we take out this mutex to guarantee
                // that our `Link` write operation below
                // is serialized with the following sync.
                // we don't put the `Rio` instance into
                // the `Mutex` because we want to drop the
                // `Mutex` right after beginning the async
                // submission.
                let link_mu = self.submission_mutex.lock();

                // using the `Link` ordering, we specify
                // that `io_uring` should not begin
                // the following `sync_file_range`
                // until the previous write is
                // complete.
                let wrote_completion = self.io_uring.write_at_ordered(
                    &*self.config.file,
                    &to_write,
                    offset,
                    rio::Ordering::Link,
                );

                let sync_completion = self.io_uring.sync_file_range(
                    &*self.config.file,
                    offset,
                    to_write.len(),
                );

                sync_completion.wait()?;

                // TODO we want to move this above the previous `wait`
                // but there seems to be an issue in `rio` that is
                // triggered when multiple threads are submitting
                // events while events from other threads are in play.
                drop(link_mu);

                wrote += wrote_completion.wait()?;
            }
        }
        #[cfg(not(feature = "io_uring"))]
        {
            let f = &self.config.file;
            pwrite_all(f, data, log_offset)?;
            if !self.config.temporary {
                #[cfg(target_os = "linux")]
                {
                    use std::os::unix::io::AsRawFd;
                    let ret = unsafe {
                        libc::sync_file_range(
                            f.as_raw_fd(),
                            i64::try_from(log_offset).unwrap(),
                            i64::try_from(total_len).unwrap(),
                            libc::SYNC_FILE_RANGE_WAIT_BEFORE
                                | libc::SYNC_FILE_RANGE_WRITE
                                | libc::SYNC_FILE_RANGE_WAIT_AFTER,
                        )
                    };
                    if ret < 0 {
                        let err = std::io::Error::last_os_error();
                        if let Some(libc::ENOSYS) = err.raw_os_error() {
                            f.sync_all()?;
                        } else {
                            return Err(err.into());
                        }
                    }
                }

                #[cfg(not(target_os = "linux"))]
                f.sync_all()?;
            }
        }
        io_fail!(self, "buffer write post");

        if total_len > 0 {
            let complete_len = if maxed {
                let lsn_idx = base_lsn / segment_size as Lsn;
                let next_seg_beginning = (lsn_idx + 1) * segment_size as Lsn;
                assert_usize(next_seg_beginning - base_lsn)
            } else {
                total_len
            };

            debug!(
                "wrote lsns {}-{} to disk at offsets {}-{}, maxed {} complete_len {}",
                base_lsn,
                base_lsn + total_len as Lsn - 1,
                log_offset,
                log_offset + total_len as LogOffset - 1,
                maxed,
                complete_len
            );
            self.mark_interval(base_lsn, complete_len);
        }

        #[allow(clippy::cast_precision_loss)]
        M.written_bytes.measure(total_len as f64);

        // NB the below deferred logic is important to ensure
        // that we never actually free a segment until all threads
        // that may have witnessed a DiskPtr that points into it
        // have completed their (crossbeam-epoch)-pinned operations.
        let guard = pin();
        let max_header_stable_lsn = self.max_header_stable_lsn.clone();
        guard.defer(move || {
            trace!("bumping atomic header lsn to {}", stored_max_stable_lsn);
            bump_atomic_lsn(&max_header_stable_lsn, stored_max_stable_lsn)
        });

        guard.flush();

        let current_max_header_stable_lsn =
            self.max_header_stable_lsn.load(SeqCst);

        self.sa_stabilize(current_max_header_stable_lsn, &guard)
    }
// Write an IO buffer's data to stable storage and set up the
// next IO buffer for writing.
Write an IO buffer's data to stable storage and set up the
next IO buffer for writing.
Write an IO buffer ' s data to stable storage and set up the next IO buffer for writing .
pub(crate) fn write_to_log(&self, iobuf: &IoBuf) -> Result<()> {
        let _measure = Measure::new(&M.write_to_log);
        let header = iobuf.get_header();
        let log_offset = iobuf.offset;
        let base_lsn = iobuf.lsn;
        let capacity = iobuf.capacity;
        let segment_size = self.config.segment_size;
        assert_eq!(
            Lsn::try_from(log_offset % segment_size as LogOffset).unwrap(),
            base_lsn % segment_size as Lsn
        );
        assert_ne!(
            log_offset,
            LogOffset::max_value(),
            "created reservation for uninitialized slot",
        );
        assert!(is_sealed(header));
        let bytes_to_write = offset(header);
        trace!(
            "write_to_log log_offset {} lsn {} len {}",
            log_offset,
            base_lsn,
            bytes_to_write
        );
        let maxed = iobuf.linearized(|| iobuf.get_maxed());
        let unused_space = capacity - bytes_to_write;
        let should_pad = maxed && unused_space >= MAX_MSG_HEADER_LEN;
        // a pad is a null message written to the end of a buffer
        // to signify that nothing else will be written into it
        if should_pad {
            let pad_len = capacity - bytes_to_write - MAX_MSG_HEADER_LEN;
            let data = iobuf.get_mut_range(bytes_to_write, pad_len);
            // take the crc of the random bytes already after where we
            // would place our header.
            let padding_bytes = vec![MessageKind::Corrupted.into(); pad_len];
            let segment_number = SegmentNumber(
                u64::try_from(base_lsn).unwrap()
                    / u64::try_from(self.config.segment_size).unwrap(),
            );
            let header = MessageHeader {
                kind: MessageKind::Cap,
                pid: PageId::max_value(),
                segment_number,
                len: u64::try_from(pad_len).unwrap(),
                crc32: 0,
            };
            let header_bytes = header.serialize();
            #[allow(unsafe_code)]
            unsafe {
                std::ptr::copy_nonoverlapping(
                    header_bytes.as_ptr(),
                    data.as_mut_ptr(),
                    header_bytes.len(),
                );
                std::ptr::copy_nonoverlapping(
                    padding_bytes.as_ptr(),
                    data.as_mut_ptr().add(header_bytes.len()),
                    pad_len,
                );
            }
            // this as to stay aligned with the hashing
            let crc32_arr = u32_to_arr(calculate_message_crc32(
                &header_bytes,
                &padding_bytes,
            ));
            #[allow(unsafe_code)]
            unsafe {
                std::ptr::copy_nonoverlapping(
                    crc32_arr.as_ptr(),
                    // the crc32 is the first part of the buffer
                    data.as_mut_ptr(),
                    std::mem::size_of::<u32>(),
                );
            }
        }
        let total_len = if maxed { capacity } else { bytes_to_write };
        let data = iobuf.get_mut_range(0, total_len);
        let stored_max_stable_lsn = iobuf.stored_max_stable_lsn;
        io_fail!(self, "buffer write");
        #[cfg(feature = "io_uring")]
        {
            let mut wrote = 0;
            while wrote < total_len {
                let to_write = &data[wrote..];
                let offset = log_offset + wrote as u64;
                // we take out this mutex to guarantee
                // that our `Link` write operation below
                // is serialized with the following sync.
                // we don't put the `Rio` instance into
                // the `Mutex` because we want to drop the
                // `Mutex` right after beginning the async
                // submission.
                let link_mu = self.submission_mutex.lock();
                // using the `Link` ordering, we specify
                // that `io_uring` should not begin
                // the following `sync_file_range`
                // until the previous write is
                // complete.
                let wrote_completion = self.io_uring.write_at_ordered(
                    &*self.config.file,
                    &to_write,
                    offset,
                    rio::Ordering::Link,
                );
                let sync_completion = self.io_uring.sync_file_range(
                    &*self.config.file,
                    offset,
                    to_write.len(),
                );
                sync_completion.wait()?;
                // TODO we want to move this above the previous `wait`
                // but there seems to be an issue in `rio` that is
                // triggered when multiple threads are submitting
                // events while events from other threads are in play.
                drop(link_mu);
                wrote += wrote_completion.wait()?;
            }
        }
        #[cfg(not(feature = "io_uring"))]
        {
            let f = &self.config.file;
            pwrite_all(f, data, log_offset)?;
            if !self.config.temporary {
                #[cfg(target_os = "linux")]
                {
                    use std::os::unix::io::AsRawFd;
                    let ret = unsafe {
                        libc::sync_file_range(
                            f.as_raw_fd(),
                            i64::try_from(log_offset).unwrap(),
                            i64::try_from(total_len).unwrap(),
                            libc::SYNC_FILE_RANGE_WAIT_BEFORE
                                | libc::SYNC_FILE_RANGE_WRITE
                                | libc::SYNC_FILE_RANGE_WAIT_AFTER,
                        )
                    };
                    if ret < 0 {
                        let err = std::io::Error::last_os_error();
                        if let Some(libc::ENOSYS) = err.raw_os_error() {
                            f.sync_all()?;
                        } else {
                            return Err(err.into());
                        }
                    }
                }
                #[cfg(not(target_os = "linux"))]
                f.sync_all()?;
            }
        }
        io_fail!(self, "buffer write post");
        if total_len > 0 {
            let complete_len = if maxed {
                let lsn_idx = base_lsn / segment_size as Lsn;
                let next_seg_beginning = (lsn_idx + 1) * segment_size as Lsn;
                assert_usize(next_seg_beginning - base_lsn)
            } else {
                total_len
            };
            debug!(
                "wrote lsns {}-{} to disk at offsets {}-{}, maxed {} complete_len {}",
                base_lsn,
                base_lsn + total_len as Lsn - 1,
                log_offset,
                log_offset + total_len as LogOffset - 1,
                maxed,
                complete_len
            );
            self.mark_interval(base_lsn, complete_len);
        }
        #[allow(clippy::cast_precision_loss)]
        M.written_bytes.measure(total_len as f64);
        // NB the below deferred logic is important to ensure
        // that we never actually free a segment until all threads
        // that may have witnessed a DiskPtr that points into it
        // have completed their (crossbeam-epoch)-pinned operations.
        let guard = pin();
        let max_header_stable_lsn = self.max_header_stable_lsn.clone();
        guard.defer(move || {
            trace!("bumping atomic header lsn to {}", stored_max_stable_lsn);
            bump_atomic_lsn(&max_header_stable_lsn, stored_max_stable_lsn)
        });
        guard.flush();
        let current_max_header_stable_lsn =
            self.max_header_stable_lsn.load(SeqCst);
        self.sa_stabilize(current_max_header_stable_lsn, &guard)
    }
pub ( crate ) fn write_to_log ( & self , iobuf : & IoBuf ) -> Result < ( ) > { let _measure = Measure :: new ( & M . write_to_log ) ; let header = iobuf . get_header ( ) ; let log_offset = iobuf . offset ; let base_lsn = iobuf . lsn ; let capacity = iobuf . capacity ; let segment_size = self . config . segment_size ; assert_eq ! ( Lsn try_from ( log_offset segment_size as LogOffset ) unwrap ( ) base_lsn segment_size as Lsn ) ; assert_ne ! ( log_offset LogOffset max_value ( ) " " ) ; assert ! ( is_sealed ( header ) ) ; let bytes_to_write = offset ( header ) ; trace ! ( " " log_offset base_lsn bytes_to_write ) ; let maxed = iobuf . linearized ( | | iobuf . get_maxed ( ) ) ; let unused_space = capacity - bytes_to_write ; let should_pad = maxed && unused_space >= MAX_MSG_HEADER_LEN ; // a pad is a null message written to the end of a buffer // to signify that nothing else will be written into it if should_pad { let pad_len = capacity - bytes_to_write - MAX_MSG_HEADER_LEN ; let data = iobuf . get_mut_range ( bytes_to_write , pad_len ) ; // take the crc of the random bytes already after where we // would place our header. let padding_bytes = vec ! [ MessageKind Corrupted into ( ) pad_len ] ; let segment_number = SegmentNumber ( u64 :: try_from ( base_lsn ) . unwrap ( ) / u64 :: try_from ( self . config . segment_size ) . unwrap ( ) , ) ; let header = MessageHeader { kind : MessageKind :: Cap , pid : PageId :: max_value ( ) , segment_number , len : u64 :: try_from ( pad_len ) . unwrap ( ) , crc32 : 0 , } ; let header_bytes = header . serialize ( ) ; # [ allow ( unsafe_code ) ] unsafe { std :: ptr :: copy_nonoverlapping ( header_bytes . as_ptr ( ) , data . as_mut_ptr ( ) , header_bytes . len ( ) , ) ; std :: ptr :: copy_nonoverlapping ( padding_bytes . as_ptr ( ) , data . as_mut_ptr ( ) . add ( header_bytes . len ( ) ) , pad_len , ) ; } // this as to stay aligned with the hashing let crc32_arr = u32_to_arr ( calculate_message_crc32 ( & header_bytes , & padding_bytes , ) ) ; # [ allow ( unsafe_code ) ] unsafe { std :: ptr :: copy_nonoverlapping ( crc32_arr . as_ptr ( ) , // the crc32 is the first part of the buffer data . as_mut_ptr ( ) , std :: mem :: size_of :: < u32 > ( ) , ) ; } } let total_len = if maxed { capacity } else { bytes_to_write } ; let data = iobuf . get_mut_range ( 0 , total_len ) ; let stored_max_stable_lsn = iobuf . stored_max_stable_lsn ; io_fail ! ( self " " ) ; # [ cfg ( feature " " ) ] { let mut wrote = 0 ; while wrote < total_len { let to_write = & data [ wrote .. ] ; let offset = log_offset + wrote as u64 ; // we take out this mutex to guarantee // that our `Link` write operation below // is serialized with the following sync. // we don't put the `Rio` instance into // the `Mutex` because we want to drop the // `Mutex` right after beginning the async // submission. let link_mu = self . submission_mutex . lock ( ) ; // using the `Link` ordering, we specify // that `io_uring` should not begin // the following `sync_file_range` // until the previous write is // complete. let wrote_completion = self . io_uring . write_at_ordered ( & * self . config . file , & to_write , offset , rio :: Ordering :: Link , ) ; let sync_completion = self . io_uring . sync_file_range ( & * self . config . file , offset , to_write . len ( ) , ) ; sync_completion . wait ( ) ? ; // TODO we want to move this above the previous `wait` // but there seems to be an issue in `rio` that is // triggered when multiple threads are submitting // events while events from other threads are in play. drop ( link_mu ) ; wrote += wrote_completion . wait ( ) ? ; } } # [ cfg ( not ( feature " " ) ) ] { let f = & self . config . file ; pwrite_all ( f , data , log_offset ) ? ; if ! self . config . temporary { # [ cfg ( target_os " " ) ] { use std :: os :: unix :: io :: AsRawFd ; let ret = unsafe { libc :: sync_file_range ( f . as_raw_fd ( ) , i64 :: try_from ( log_offset ) . unwrap ( ) , i64 :: try_from ( total_len ) . unwrap ( ) , libc :: SYNC_FILE_RANGE_WAIT_BEFORE | libc :: SYNC_FILE_RANGE_WRITE | libc :: SYNC_FILE_RANGE_WAIT_AFTER , ) } ; if ret < 0 { let err = std :: io :: Error :: last_os_error ( ) ; if let Some ( libc :: ENOSYS ) = err . raw_os_error ( ) { f . sync_all ( ) ? ; } else { return Err ( err . into ( ) ) ; } } } # [ cfg ( not ( target_os " " ) ) ] f . sync_all ( ) ? ; } } io_fail ! ( self " " ) ; if total_len > 0 { let complete_len = if maxed { let lsn_idx = base_lsn / segment_size as Lsn ; let next_seg_beginning = ( lsn_idx + 1 ) * segment_size as Lsn ; assert_usize ( next_seg_beginning - base_lsn ) } else { total_len } ; debug ! ( " " base_lsn base_lsn total_len as Lsn 1 log_offset log_offset total_len as LogOffset 1 maxed complete_len ) ; self . mark_interval ( base_lsn , complete_len ) ; } # [ allow ( clippy cast_precision_loss ) ] M . written_bytes . measure ( total_len as f64 ) ; // NB the below deferred logic is important to ensure // that we never actually free a segment until all threads // that may have witnessed a DiskPtr that points into it // have completed their (crossbeam-epoch)-pinned operations. let guard = pin ( ) ; let max_header_stable_lsn = self . max_header_stable_lsn . clone ( ) ; guard . defer ( move | | { trace ! ( " " stored_max_stable_lsn ) ; bump_atomic_lsn ( & max_header_stable_lsn , stored_max_stable_lsn ) } ) ; guard . flush ( ) ; let current_max_header_stable_lsn = self . max_header_stable_lsn . load ( SeqCst ) ; self . sa_stabilize ( current_max_header_stable_lsn , & guard ) }
Write an IO buffer's data to stable storage and set up the
next IO buffer for writing.
Write an IO buffer ' s data to stable storage and set up the next IO buffer for writing .
&self
iobuf
&IoBuf