/// Returns all data which can be flushed.  |allocate_buffer| is a callback which is used to
/// allocate Buffer objects.  Each pending data region will be copied into a Buffer and returned
/// to the caller in block-aligned ranges.  If there's already an ongoing flush, the caller
/// receives an Event that resolves when the other flush is complete.  This is not thread-safe
/// with respect to cache mutations; the caller must ensure that no changes can be made to the
/// cache for the duration of this call.
Returns all data which can be flushed.  |allocate_buffer| is a callback which is used to
allocate Buffer objects.  Each pending data region will be copied into a Buffer and returned
to the caller in block-aligned ranges.  If there's already an ongoing flush, the caller
receives an Event that resolves when the other flush is complete.  This is not thread-safe
with respect to cache mutations; the caller must ensure that no changes can be made to the
cache for the duration of this call.
Returns all data which can be flushed . |allocate_buffer| is a callback which is used to allocate Buffer objects . Each pending data region will be copied into a Buffer and returned to the caller in block - aligned ranges . If there ' s already an ongoing flush the caller receives an Event that resolves when the other flush is complete . This is not thread - safe with respect to cache mutations ; the caller must ensure that no changes can be made to the cache for the duration of this call .
pub fn take_flushable_data<'a, 'b, F>(
        &'b self,
        block_size: u64,
        last_known_size: u64,
        allocate_buffer: F,
        reserver: &'b dyn StorageReservation,
        reservation: &'b allocator::Reservation,
        progress_offset: u64,
        limit: u64,
    ) -> FlushableData<'a, 'b, B>
    where
        F: Fn(usize) -> Buffer<'a>,
    {
        // TODO(jfsulliv): Support using a single shared transfer buffer for all pending data.
        let mut inner = self.inner.lock().unwrap();
        assert!(inner.flushing_min_size.is_none()); // Ensure a flush isn't currently in-progress.
        let size = *inner.content_size.as_ref().unwrap_or(&(self.data.size() as u64));
        let intervals = inner.intervals.remove_interval(&(0..u64::MAX)).unwrap();
        let mut bytes_to_flush = 0;
        let mut flush_progress_offset =
            if progress_offset > 0 { Some(progress_offset) } else { None };
        let mut ranges = vec![];
        let mut i = 0;
        while let Some(mut interval) = intervals.get(i).cloned() {
            assert!(interval.range.start % block_size == 0);
            assert!(interval.range.end % block_size == 0);
            if let CacheState::Dirty = &interval.state {
            } else {
                panic!("Unexpected interval {:?}", interval);
            }
            let interval_end = std::cmp::min(size, interval.range.end);
            let end = std::cmp::min(interval_end, interval.range.start + limit - bytes_to_flush);
            if end < interval_end {
                // If we only flushed part of the interval due to limit being hit,
                // re-insert the tail of the interval.
                let mut interval_tail = interval.clone();
                interval_tail.range.start = round_down(end, block_size);
                inner.intervals.add_interval(&interval_tail).unwrap();
                interval.state = CacheState::Flushing;
                interval.range.end = end;
                inner.intervals.add_interval(&interval).unwrap();
            } else {
                interval.state = CacheState::Flushing;
                inner.intervals.add_interval(&interval).unwrap();
            }
            ranges.push(interval.range.clone());
            flush_progress_offset =
                if i == intervals.len() - 1 && end == interval_end { None } else { Some(end) };
            i += 1;
            bytes_to_flush += interval.range.end - interval.range.start;
            if bytes_to_flush >= limit || ranges.len() >= MAX_FLUSH_RANGE_COUNT {
                break;
            }
        }
        for interval in &intervals[i..] {
            // Add back any intervals we skipped flushing.
            inner.intervals.add_interval(interval).unwrap();
        }
        inner.dirty_bytes = inner.dirty_bytes.checked_sub(bytes_to_flush).unwrap();
        let mut buffer = allocate_buffer(bytes_to_flush as usize);
        let mut slice = buffer.as_mut_slice();
        for r in &ranges {
            let (head, tail) = slice.split_at_mut((r.end - r.start) as usize);
            self.data.raw_read(r.start, head);
            slice = tail;
        }
        let content_size = if let Some(offset) = flush_progress_offset {
            // If this is a partial flush, we can only update the content size as far as the data
            // we've flushed, and only update it if that size was an increase (otherwise this would
            // appear to be a false truncation).
            if offset > last_known_size {
                Some(offset)
            } else {
                None
            }
        } else {
            inner.content_size.take()
        };
        let metadata = FlushableMetadata {
            content_size,
            creation_time: std::mem::take(&mut inner.creation_time),
            modification_time: std::mem::take(&mut inner.modification_time),
        };
        inner.flushing_min_size = Some(inner.min_size);
        if let Some(s) = inner.content_size {
            inner.min_size = s;
        }
        let bytes_reserved = reserver.reservation_needed(bytes_to_flush);
        reservation.add(bytes_reserved);
        FlushableData {
            metadata,
            reservation: &reservation,
            reserver,
            ranges,
            buffer,
            flush_progress_offset,
            cache: Some(self),
        }
    }
pub fn take_flushable_data < ' a , ' b , F > ( & ' b self , block_size : u64 , last_known_size : u64 , allocate_buffer : F , reserver : & ' b dyn StorageReservation , reservation : & ' b allocator :: Reservation , progress_offset : u64 , limit : u64 , ) -> FlushableData < ' a , ' b , B > where F : Fn ( usize ) -> Buffer < ' a > , { // TODO(jfsulliv): Support using a single shared transfer buffer for all pending data. let mut inner = self . inner . lock ( ) . unwrap ( ) ; assert ! ( inner flushing_min_size is_none ( ) ) ; // Ensure a flush isn't currently in-progress. let size = * inner . content_size . as_ref ( ) . unwrap_or ( & ( self . data . size ( ) as u64 ) ) ; let intervals = inner . intervals . remove_interval ( & ( 0 .. u64 :: MAX ) ) . unwrap ( ) ; let mut bytes_to_flush = 0 ; let mut flush_progress_offset = if progress_offset > 0 { Some ( progress_offset ) } else { None } ; let mut ranges = vec ! [ ] ; let mut i = 0 ; while let Some ( mut interval ) = intervals . get ( i ) . cloned ( ) { assert ! ( interval range start block_size 0 ) ; assert ! ( interval range end block_size 0 ) ; if let CacheState :: Dirty = & interval . state { } else { panic ! ( " " interval ) ; } let interval_end = std :: cmp :: min ( size , interval . range . end ) ; let end = std :: cmp :: min ( interval_end , interval . range . start + limit - bytes_to_flush ) ; if end < interval_end { // If we only flushed part of the interval due to limit being hit, // re-insert the tail of the interval. let mut interval_tail = interval . clone ( ) ; interval_tail . range . start = round_down ( end , block_size ) ; inner . intervals . add_interval ( & interval_tail ) . unwrap ( ) ; interval . state = CacheState :: Flushing ; interval . range . end = end ; inner . intervals . add_interval ( & interval ) . unwrap ( ) ; } else { interval . state = CacheState :: Flushing ; inner . intervals . add_interval ( & interval ) . unwrap ( ) ; } ranges . push ( interval . range . clone ( ) ) ; flush_progress_offset = if i == intervals . len ( ) - 1 && end == interval_end { None } else { Some ( end ) } ; i += 1 ; bytes_to_flush += interval . range . end - interval . range . start ; if bytes_to_flush >= limit || ranges . len ( ) >= MAX_FLUSH_RANGE_COUNT { break ; } } for interval in & intervals [ i .. ] { // Add back any intervals we skipped flushing. inner . intervals . add_interval ( interval ) . unwrap ( ) ; } inner . dirty_bytes = inner . dirty_bytes . checked_sub ( bytes_to_flush ) . unwrap ( ) ; let mut buffer = allocate_buffer ( bytes_to_flush as usize ) ; let mut slice = buffer . as_mut_slice ( ) ; for r in & ranges { let ( head , tail ) = slice . split_at_mut ( ( r . end - r . start ) as usize ) ; self . data . raw_read ( r . start , head ) ; slice = tail ; } let content_size = if let Some ( offset ) = flush_progress_offset { // If this is a partial flush, we can only update the content size as far as the data // we've flushed, and only update it if that size was an increase (otherwise this would // appear to be a false truncation). if offset > last_known_size { Some ( offset ) } else { None } } else { inner . content_size . take ( ) } ; let metadata = FlushableMetadata { content_size , creation_time : std :: mem :: take ( & mut inner . creation_time ) , modification_time : std :: mem :: take ( & mut inner . modification_time ) , } ; inner . flushing_min_size = Some ( inner . min_size ) ; if let Some ( s ) = inner . content_size { inner . min_size = s ; } let bytes_reserved = reserver . reservation_needed ( bytes_to_flush ) ; reservation . add ( bytes_reserved ) ; FlushableData { metadata , reservation : & reservation , reserver , ranges , buffer , flush_progress_offset , cache : Some ( self ) , } }
Returns all data which can be flushed.
Returns all data which can be flushed .
&'b self
block_size
u64
last_known_size
u64
allocate_buffer
F
reserver
&'b dyn StorageReservation
reservation
&'b allocator::Reservation
progress_offset
u64
limit
u64