pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
    iobufs: &Arc<IoBufs>,
    iobuf: &Arc<IoBuf>,
    header: Header,
    from_reserve: bool,
) -> Result<()> {
    if is_sealed(header) {
        // this buffer is already sealed. nothing to do here.
        return Ok(());
    }

    // NB need to do this before CAS because it can get
    // written and reset by another thread afterward
    let lid = iobuf.offset;
    let lsn = iobuf.lsn;
    let capacity = iobuf.capacity;
    let segment_size = iobufs.config.segment_size;

    if offset(header) > capacity {
        // a race happened, nothing we can do
        return Ok(());
    }

    let sealed = mk_sealed(header);
    let res_len = offset(sealed);

    let maxed = from_reserve || capacity - res_len < MAX_MSG_HEADER_LEN;

    let worked = iobuf.linearized(|| {
        if iobuf.cas_header(header, sealed).is_err() {
            // cas failed, don't try to continue
            return false;
        }

        trace!("sealed iobuf with lsn {}", lsn);

        if maxed {
            // NB we linearize this together with sealing
            // the header here to guarantee that in write_to_log,
            // which may be executing as soon as the seal is set
            // by another thread, the thread that calls
            // iobuf.get_maxed() is linearized with this one!
            trace!("setting maxed to true for iobuf with lsn {}", lsn);
            iobuf.set_maxed(true);
        }
        true
    });
    if !worked {
        return Ok(());
    }

    assert!(
        capacity + SEG_HEADER_LEN >= res_len,
        "res_len of {} higher than buffer capacity {}",
        res_len,
        capacity
    );

    assert_ne!(
        lid,
        LogOffset::max_value(),
        "sealing something that should never have \
         been claimed (iobuf lsn {})\n{:?}",
        lsn,
        iobufs
    );

    // open new slot
    let mut next_lsn = lsn;

    let measure_assign_offset = Measure::new(&M.assign_offset);

    let next_offset = if maxed {
        // roll lsn to the next offset
        let lsn_idx = lsn / segment_size as Lsn;
        next_lsn = (lsn_idx + 1) * segment_size as Lsn;

        // mark unused as clear
        debug!(
            "rolling to new segment after clearing {}-{}",
            lid,
            lid + res_len as LogOffset,
        );

        match iobufs.with_sa(|sa| sa.next(next_lsn)) {
            Ok(ret) => ret,
            Err(e) => {
                iobufs.config.set_global_error(e.clone());
                let intervals = iobufs.intervals.lock();

                // having held the mutex makes this linearized
                // with the notify below.
                drop(intervals);

                let _notified = iobufs.interval_updated.notify_all();
                return Err(e);
            }
        }
    } else {
        debug!(
            "advancing offset within the current segment from {} to {}",
            lid,
            lid + res_len as LogOffset
        );
        next_lsn += res_len as Lsn;

        lid + res_len as LogOffset
    };

    // NB as soon as the "sealed" bit is 0, this allows new threads
    // to start writing into this buffer, so do that after it's all
    // set up. expect this thread to block until the buffer completes
    // its entire life cycle as soon as we do that.
    let next_iobuf = if maxed {
        let mut next_iobuf = IoBuf {
            buf: Arc::new(AlignedBuf::new(segment_size)),
            header: CachePadded::new(AtomicU64::new(0)),
            base: 0,
            offset: next_offset,
            lsn: next_lsn,
            capacity: segment_size,
            maxed: AtomicBool::new(false),
            linearizer: Mutex::new(()),
            stored_max_stable_lsn: -1,
        };

        next_iobuf.store_segment_header(sealed, next_lsn, iobufs.stable());

        next_iobuf
    } else {
        let new_cap = capacity - res_len;
        assert_ne!(new_cap, 0);
        let last_salt = salt(sealed);
        let new_salt = bump_salt(last_salt);

        IoBuf {
            // reuse the previous io buffer
            buf: iobuf.buf.clone(),
            header: CachePadded::new(AtomicU64::new(new_salt)),
            base: iobuf.base + res_len,
            offset: next_offset,
            lsn: next_lsn,
            capacity: new_cap,
            maxed: AtomicBool::new(false),
            linearizer: Mutex::new(()),
            stored_max_stable_lsn: -1,
        }
    };

    // we acquire this mutex to guarantee that any threads that
    // are going to wait on the condition variable will observe
    // the change.
    debug_delay();
    let intervals = iobufs.intervals.lock();
    let old_ptr = iobufs
        .iobuf
        .swap(Arc::into_raw(Arc::new(next_iobuf)) as *mut IoBuf, SeqCst);

    let old_arc = unsafe { Arc::from_raw(old_ptr) };

    pin().defer(move || drop(old_arc));

    // having held the mutex makes this linearized
    // with the notify below.
    drop(intervals);

    let _notified = iobufs.interval_updated.notify_all();

    drop(measure_assign_offset);

    // if writers is 0, it's our responsibility to write the buffer.
    if n_writers(sealed) == 0 {
        iobufs.config.global_error()?;
        trace!(
            "asynchronously writing iobuf with lsn {} to log from maybe_seal",
            lsn
        );
        let iobufs = iobufs.clone();
        let iobuf = iobuf.clone();
        let _result = threadpool::spawn(move || {
            if let Err(e) = iobufs.write_to_log(&iobuf) {
                error!(
                    "hit error while writing iobuf with lsn {}: {:?}",
                    lsn, e
                );
                let intervals = iobufs.intervals.lock();

                // having held the mutex makes this linearized
                // with the notify below.
                drop(intervals);

                let _notified = iobufs.interval_updated.notify_all();
                iobufs.config.set_global_error(e);
            }
        });

        #[cfg(feature = "event_log")]
        _result.unwrap();

        Ok(())
    } else {
        Ok(())
    }
}
/// Attempt to seal the current IO buffer, possibly
/// writing it to disk if there are no other writers
/// operating on it.
Attempt to seal the current IO buffer, possibly
writing it to disk if there are no other writers
operating on it.
Attempt to seal the current IO buffer possibly writing it to disk if there are no other writers operating on it .
pub(in crate::pagecache) fn maybe_seal_and_write_iobuf(
    iobufs: &Arc<IoBufs>,
    iobuf: &Arc<IoBuf>,
    header: Header,
    from_reserve: bool,
) -> Result<()> {
    if is_sealed(header) {
        // this buffer is already sealed. nothing to do here.
        return Ok(());
    }
    // NB need to do this before CAS because it can get
    // written and reset by another thread afterward
    let lid = iobuf.offset;
    let lsn = iobuf.lsn;
    let capacity = iobuf.capacity;
    let segment_size = iobufs.config.segment_size;
    if offset(header) > capacity {
        // a race happened, nothing we can do
        return Ok(());
    }
    let sealed = mk_sealed(header);
    let res_len = offset(sealed);
    let maxed = from_reserve || capacity - res_len < MAX_MSG_HEADER_LEN;
    let worked = iobuf.linearized(|| {
        if iobuf.cas_header(header, sealed).is_err() {
            // cas failed, don't try to continue
            return false;
        }
        trace!("sealed iobuf with lsn {}", lsn);
        if maxed {
            // NB we linearize this together with sealing
            // the header here to guarantee that in write_to_log,
            // which may be executing as soon as the seal is set
            // by another thread, the thread that calls
            // iobuf.get_maxed() is linearized with this one!
            trace!("setting maxed to true for iobuf with lsn {}", lsn);
            iobuf.set_maxed(true);
        }
        true
    });
    if !worked {
        return Ok(());
    }
    assert!(
        capacity + SEG_HEADER_LEN >= res_len,
        "res_len of {} higher than buffer capacity {}",
        res_len,
        capacity
    );
    assert_ne!(
        lid,
        LogOffset::max_value(),
        "sealing something that should never have \
         been claimed (iobuf lsn {})\n{:?}",
        lsn,
        iobufs
    );
    // open new slot
    let mut next_lsn = lsn;
    let measure_assign_offset = Measure::new(&M.assign_offset);
    let next_offset = if maxed {
        // roll lsn to the next offset
        let lsn_idx = lsn / segment_size as Lsn;
        next_lsn = (lsn_idx + 1) * segment_size as Lsn;
        // mark unused as clear
        debug!(
            "rolling to new segment after clearing {}-{}",
            lid,
            lid + res_len as LogOffset,
        );
        match iobufs.with_sa(|sa| sa.next(next_lsn)) {
            Ok(ret) => ret,
            Err(e) => {
                iobufs.config.set_global_error(e.clone());
                let intervals = iobufs.intervals.lock();
                // having held the mutex makes this linearized
                // with the notify below.
                drop(intervals);
                let _notified = iobufs.interval_updated.notify_all();
                return Err(e);
            }
        }
    } else {
        debug!(
            "advancing offset within the current segment from {} to {}",
            lid,
            lid + res_len as LogOffset
        );
        next_lsn += res_len as Lsn;
        lid + res_len as LogOffset
    };
    // NB as soon as the "sealed" bit is 0, this allows new threads
    // to start writing into this buffer, so do that after it's all
    // set up. expect this thread to block until the buffer completes
    // its entire life cycle as soon as we do that.
    let next_iobuf = if maxed {
        let mut next_iobuf = IoBuf {
            buf: Arc::new(AlignedBuf::new(segment_size)),
            header: CachePadded::new(AtomicU64::new(0)),
            base: 0,
            offset: next_offset,
            lsn: next_lsn,
            capacity: segment_size,
            maxed: AtomicBool::new(false),
            linearizer: Mutex::new(()),
            stored_max_stable_lsn: -1,
        };
        next_iobuf.store_segment_header(sealed, next_lsn, iobufs.stable());
        next_iobuf
    } else {
        let new_cap = capacity - res_len;
        assert_ne!(new_cap, 0);
        let last_salt = salt(sealed);
        let new_salt = bump_salt(last_salt);
        IoBuf {
            // reuse the previous io buffer
            buf: iobuf.buf.clone(),
            header: CachePadded::new(AtomicU64::new(new_salt)),
            base: iobuf.base + res_len,
            offset: next_offset,
            lsn: next_lsn,
            capacity: new_cap,
            maxed: AtomicBool::new(false),
            linearizer: Mutex::new(()),
            stored_max_stable_lsn: -1,
        }
    };
    // we acquire this mutex to guarantee that any threads that
    // are going to wait on the condition variable will observe
    // the change.
    debug_delay();
    let intervals = iobufs.intervals.lock();
    let old_ptr = iobufs
        .iobuf
        .swap(Arc::into_raw(Arc::new(next_iobuf)) as *mut IoBuf, SeqCst);
    let old_arc = unsafe { Arc::from_raw(old_ptr) };
    pin().defer(move || drop(old_arc));
    // having held the mutex makes this linearized
    // with the notify below.
    drop(intervals);
    let _notified = iobufs.interval_updated.notify_all();
    drop(measure_assign_offset);
    // if writers is 0, it's our responsibility to write the buffer.
    if n_writers(sealed) == 0 {
        iobufs.config.global_error()?;
        trace!(
            "asynchronously writing iobuf with lsn {} to log from maybe_seal",
            lsn
        );
        let iobufs = iobufs.clone();
        let iobuf = iobuf.clone();
        let _result = threadpool::spawn(move || {
            if let Err(e) = iobufs.write_to_log(&iobuf) {
                error!(
                    "hit error while writing iobuf with lsn {}: {:?}",
                    lsn, e
                );
                let intervals = iobufs.intervals.lock();
                // having held the mutex makes this linearized
                // with the notify below.
                drop(intervals);
                let _notified = iobufs.interval_updated.notify_all();
                iobufs.config.set_global_error(e);
            }
        });
        #[cfg(feature = "event_log")]
        _result.unwrap();
        Ok(())
    } else {
        Ok(())
    }
}
pub ( in crate :: pagecache ) fn maybe_seal_and_write_iobuf ( iobufs : & Arc < IoBufs > , iobuf : & Arc < IoBuf > , header : Header , from_reserve : bool , ) -> Result < ( ) > { if is_sealed ( header ) { // this buffer is already sealed. nothing to do here. return Ok ( ( ) ) ; } // NB need to do this before CAS because it can get // written and reset by another thread afterward let lid = iobuf . offset ; let lsn = iobuf . lsn ; let capacity = iobuf . capacity ; let segment_size = iobufs . config . segment_size ; if offset ( header ) > capacity { // a race happened, nothing we can do return Ok ( ( ) ) ; } let sealed = mk_sealed ( header ) ; let res_len = offset ( sealed ) ; let maxed = from_reserve || capacity - res_len < MAX_MSG_HEADER_LEN ; let worked = iobuf . linearized ( | | { if iobuf . cas_header ( header , sealed ) . is_err ( ) { // cas failed, don't try to continue return false ; } trace ! ( " " lsn ) ; if maxed { // NB we linearize this together with sealing // the header here to guarantee that in write_to_log, // which may be executing as soon as the seal is set // by another thread, the thread that calls // iobuf.get_maxed() is linearized with this one! trace ! ( " " lsn ) ; iobuf . set_maxed ( true ) ; } true } ) ; if ! worked { return Ok ( ( ) ) ; } assert ! ( capacity SEG_HEADER_LEN res_len " " res_len capacity ) ; assert_ne ! ( lid LogOffset max_value ( ) " \
 \n " lsn iobufs ) ; // open new slot let mut next_lsn = lsn ; let measure_assign_offset = Measure :: new ( & M . assign_offset ) ; let next_offset = if maxed { // roll lsn to the next offset let lsn_idx = lsn / segment_size as Lsn ; next_lsn = ( lsn_idx + 1 ) * segment_size as Lsn ; // mark unused as clear debug ! ( " " lid lid res_len as LogOffset ) ; match iobufs . with_sa ( | sa | sa . next ( next_lsn ) ) { Ok ( ret ) => ret , Err ( e ) => { iobufs . config . set_global_error ( e . clone ( ) ) ; let intervals = iobufs . intervals . lock ( ) ; // having held the mutex makes this linearized // with the notify below. drop ( intervals ) ; let _notified = iobufs . interval_updated . notify_all ( ) ; return Err ( e ) ; } } } else { debug ! ( " " lid lid res_len as LogOffset ) ; next_lsn += res_len as Lsn ; lid + res_len as LogOffset } ; // NB as soon as the "sealed" bit is 0, this allows new threads // to start writing into this buffer, so do that after it's all // set up. expect this thread to block until the buffer completes // its entire life cycle as soon as we do that. let next_iobuf = if maxed { let mut next_iobuf = IoBuf { buf : Arc :: new ( AlignedBuf :: new ( segment_size ) ) , header : CachePadded :: new ( AtomicU64 :: new ( 0 ) ) , base : 0 , offset : next_offset , lsn : next_lsn , capacity : segment_size , maxed : AtomicBool :: new ( false ) , linearizer : Mutex :: new ( ( ) ) , stored_max_stable_lsn : - 1 , } ; next_iobuf . store_segment_header ( sealed , next_lsn , iobufs . stable ( ) ) ; next_iobuf } else { let new_cap = capacity - res_len ; assert_ne ! ( new_cap 0 ) ; let last_salt = salt ( sealed ) ; let new_salt = bump_salt ( last_salt ) ; IoBuf { // reuse the previous io buffer buf : iobuf . buf . clone ( ) , header : CachePadded :: new ( AtomicU64 :: new ( new_salt ) ) , base : iobuf . base + res_len , offset : next_offset , lsn : next_lsn , capacity : new_cap , maxed : AtomicBool :: new ( false ) , linearizer : Mutex :: new ( ( ) ) , stored_max_stable_lsn : - 1 , } } ; // we acquire this mutex to guarantee that any threads that // are going to wait on the condition variable will observe // the change. debug_delay ( ) ; let intervals = iobufs . intervals . lock ( ) ; let old_ptr = iobufs . iobuf . swap ( Arc :: into_raw ( Arc :: new ( next_iobuf ) ) as * mut IoBuf , SeqCst ) ; let old_arc = unsafe { Arc :: from_raw ( old_ptr ) } ; pin ( ) . defer ( move | | drop ( old_arc ) ) ; // having held the mutex makes this linearized // with the notify below. drop ( intervals ) ; let _notified = iobufs . interval_updated . notify_all ( ) ; drop ( measure_assign_offset ) ; // if writers is 0, it's our responsibility to write the buffer. if n_writers ( sealed ) == 0 { iobufs . config . global_error ( ) ? ; trace ! ( " " lsn ) ; let iobufs = iobufs . clone ( ) ; let iobuf = iobuf . clone ( ) ; let _result = threadpool :: spawn ( move | | { if let Err ( e ) = iobufs . write_to_log ( & iobuf ) { error ! ( " " lsn e ) ; let intervals = iobufs . intervals . lock ( ) ; // having held the mutex makes this linearized // with the notify below. drop ( intervals ) ; let _notified = iobufs . interval_updated . notify_all ( ) ; iobufs . config . set_global_error ( e ) ; } } ) ; # [ cfg ( feature " " ) ] _result . unwrap ( ) ; Ok ( ( ) ) } else { Ok ( ( ) ) } }
Attempt to seal the current IO buffer, possibly
writing it to disk if there are no other writers
operating on it.
Attempt to seal the current IO buffer possibly writing it to disk if there are no other writers operating on it .
iobufs
&Arc<IoBufs>
iobuf
&Arc<IoBuf>
header
Header
from_reserve
bool