pub async fn maybe_persist<P: Persister>(&mut self, persister: &Arc<P>) {
        let LifecycleStats {
            mut total_bytes,
            partition_stats,
        } = self.stats();

        // get anything over the threshold size or age to persist
        let now = self.time_provider.now();

        let (mut to_persist, mut rest): (
            Vec<PartitionLifecycleStats>,
            Vec<PartitionLifecycleStats>,
        ) = partition_stats.into_iter().partition(|s| {
            let aged_out = now
                .checked_duration_since(s.first_write)
                .map(|age| age > self.config.partition_age_threshold)
                .unwrap_or(false);
            if aged_out {
                self.persist_age_counter.inc(1);
            }

            let is_cold = now
                .checked_duration_since(s.last_write)
                .map(|age| age > self.config.partition_cold_threshold)
                .unwrap_or(false);
            if is_cold {
                self.persist_cold_counter.inc(1);
            }

            let sized_out = s.bytes_written > self.config.partition_size_threshold;
            if sized_out {
                self.persist_size_counter.inc(1);
            }

            aged_out || sized_out || is_cold
        });

        // keep track of what we'll be evicting to see what else to drop
        for s in &to_persist {
            total_bytes -= s.bytes_written;
        }

        // if we're still over the memory threshold, persist as many of the largest partitions
        // until we're under. It's ok if this is stale, it'll just get handled on the next pass
        // through.
        if total_bytes > self.config.persist_memory_threshold {
            rest.sort_by(|a, b| b.bytes_written.cmp(&a.bytes_written));

            let mut remaining = vec![];

            let mut memory_persist_counter = 0;
            for s in rest {
                if total_bytes >= self.config.persist_memory_threshold {
                    total_bytes -= s.bytes_written;
                    to_persist.push(s);
                    memory_persist_counter += 1;
                } else {
                    remaining.push(s);
                }
            }

            self.persist_memory_counter.inc(memory_persist_counter);

            rest = remaining;
        }

        // for the sequencers that are getting data persisted, keep track of what
        // the highest seqeunce number was for each.
        let mut sequencer_maxes = BTreeMap::new();
        for s in &to_persist {
            sequencer_maxes
                .entry(s.sequencer_id)
                .and_modify(|sn| {
                    if *sn < s.first_sequence_number {
                        *sn = s.first_sequence_number;
                    }
                })
                .or_insert(s.first_sequence_number);
        }

        let persist_tasks: Vec<_> = to_persist
            .into_iter()
            .map(|s| {
                self.remove(s.partition_id);
                let persister = Arc::clone(persister);

                let (_tracker, registration) = self.job_registry.register(Job::Persist {
                    partition_id: s.partition_id,
                });
                tokio::task::spawn(async move {
                    persister.persist(s.partition_id).await;
                })
                .track(registration)
            })
            .collect();

        if !persist_tasks.is_empty() {
            let persists = futures::future::join_all(persist_tasks.into_iter());
            let results = persists.await;
            for res in results {
                res.expect("not aborted").expect("task finished");
            }

            // for the sequencers that had data persisted, update their min_unpersisted_sequence_number to
            // either the minimum remaining in everything that didn't get persisted, or the highest
            // number that was persisted. Marking the highest number as the state is ok because it
            // just needs to represent the farthest we'd have to seek back in the write buffer. Any
            // data replayed during recovery that has already been persisted will just be ignored.
            //
            // The calculation of the min unpersisted sequence number is:
            // - If there is any unpersisted data (i.e. `rest` contains entries for that sequencer) then we take the
            //   minimum sequence number from there. Note that there might be a gap between the data that we have just
            //   persisted and the unpersisted section.
            // - If there is NO unpersisted data, we take the max sequence number of the part that we have just
            //   persisted. Note that can cannot use "max + 1" because the lifecycle handler receives writes on a
            //   partition level and therefore might run while data for a single sequence number is added. So the max
            //   sequence number that we have just persisted might have more data.
            for (sequencer_id, sequence_number) in sequencer_maxes {
                let min = rest
                    .iter()
                    .filter(|s| s.sequencer_id == sequencer_id)
                    .map(|s| s.first_sequence_number)
                    .min()
                    .unwrap_or(sequence_number);
                persister
                    .update_min_unpersisted_sequence_number(sequencer_id, min)
                    .await;
            }
        }
    }
/// This will persist any partitions that are over their size or age thresholds and
/// persist as many partitions as necessary (largest first) to get below the memory threshold.
/// The persist operations are spawned in new tasks and run at the same time, but the
/// function waits for all to return before completing.
This will persist any partitions that are over their size or age thresholds and
persist as many partitions as necessary (largest first) to get below the memory threshold.
The persist operations are spawned in new tasks and run at the same time, but the
function waits for all to return before completing.
This will persist any partitions that are over their size or age thresholds and persist as many partitions as necessary ( largest first ) to get below the memory threshold . The persist operations are spawned in new tasks and run at the same time but the function waits for all to return before completing .
pub async fn maybe_persist<P: Persister>(&mut self, persister: &Arc<P>) {
        let LifecycleStats {
            mut total_bytes,
            partition_stats,
        } = self.stats();
        // get anything over the threshold size or age to persist
        let now = self.time_provider.now();
        let (mut to_persist, mut rest): (
            Vec<PartitionLifecycleStats>,
            Vec<PartitionLifecycleStats>,
        ) = partition_stats.into_iter().partition(|s| {
            let aged_out = now
                .checked_duration_since(s.first_write)
                .map(|age| age > self.config.partition_age_threshold)
                .unwrap_or(false);
            if aged_out {
                self.persist_age_counter.inc(1);
            }
            let is_cold = now
                .checked_duration_since(s.last_write)
                .map(|age| age > self.config.partition_cold_threshold)
                .unwrap_or(false);
            if is_cold {
                self.persist_cold_counter.inc(1);
            }
            let sized_out = s.bytes_written > self.config.partition_size_threshold;
            if sized_out {
                self.persist_size_counter.inc(1);
            }
            aged_out || sized_out || is_cold
        });
        // keep track of what we'll be evicting to see what else to drop
        for s in &to_persist {
            total_bytes -= s.bytes_written;
        }
        // if we're still over the memory threshold, persist as many of the largest partitions
        // until we're under. It's ok if this is stale, it'll just get handled on the next pass
        // through.
        if total_bytes > self.config.persist_memory_threshold {
            rest.sort_by(|a, b| b.bytes_written.cmp(&a.bytes_written));
            let mut remaining = vec![];
            let mut memory_persist_counter = 0;
            for s in rest {
                if total_bytes >= self.config.persist_memory_threshold {
                    total_bytes -= s.bytes_written;
                    to_persist.push(s);
                    memory_persist_counter += 1;
                } else {
                    remaining.push(s);
                }
            }
            self.persist_memory_counter.inc(memory_persist_counter);
            rest = remaining;
        }
        // for the sequencers that are getting data persisted, keep track of what
        // the highest seqeunce number was for each.
        let mut sequencer_maxes = BTreeMap::new();
        for s in &to_persist {
            sequencer_maxes
                .entry(s.sequencer_id)
                .and_modify(|sn| {
                    if *sn < s.first_sequence_number {
                        *sn = s.first_sequence_number;
                    }
                })
                .or_insert(s.first_sequence_number);
        }
        let persist_tasks: Vec<_> = to_persist
            .into_iter()
            .map(|s| {
                self.remove(s.partition_id);
                let persister = Arc::clone(persister);
                let (_tracker, registration) = self.job_registry.register(Job::Persist {
                    partition_id: s.partition_id,
                });
                tokio::task::spawn(async move {
                    persister.persist(s.partition_id).await;
                })
                .track(registration)
            })
            .collect();
        if !persist_tasks.is_empty() {
            let persists = futures::future::join_all(persist_tasks.into_iter());
            let results = persists.await;
            for res in results {
                res.expect("not aborted").expect("task finished");
            }
            // for the sequencers that had data persisted, update their min_unpersisted_sequence_number to
            // either the minimum remaining in everything that didn't get persisted, or the highest
            // number that was persisted. Marking the highest number as the state is ok because it
            // just needs to represent the farthest we'd have to seek back in the write buffer. Any
            // data replayed during recovery that has already been persisted will just be ignored.
            //
            // The calculation of the min unpersisted sequence number is:
            // - If there is any unpersisted data (i.e. `rest` contains entries for that sequencer) then we take the
            //   minimum sequence number from there. Note that there might be a gap between the data that we have just
            //   persisted and the unpersisted section.
            // - If there is NO unpersisted data, we take the max sequence number of the part that we have just
            //   persisted. Note that can cannot use "max + 1" because the lifecycle handler receives writes on a
            //   partition level and therefore might run while data for a single sequence number is added. So the max
            //   sequence number that we have just persisted might have more data.
            for (sequencer_id, sequence_number) in sequencer_maxes {
                let min = rest
                    .iter()
                    .filter(|s| s.sequencer_id == sequencer_id)
                    .map(|s| s.first_sequence_number)
                    .min()
                    .unwrap_or(sequence_number);
                persister
                    .update_min_unpersisted_sequence_number(sequencer_id, min)
                    .await;
            }
        }
    }
pub async fn maybe_persist < P : Persister > ( & mut self , persister : & Arc < P > ) { let LifecycleStats { mut total_bytes , partition_stats , } = self . stats ( ) ; // get anything over the threshold size or age to persist let now = self . time_provider . now ( ) ; let ( mut to_persist , mut rest ) : ( Vec < PartitionLifecycleStats > , Vec < PartitionLifecycleStats > , ) = partition_stats . into_iter ( ) . partition ( | s | { let aged_out = now . checked_duration_since ( s . first_write ) . map ( | age | age > self . config . partition_age_threshold ) . unwrap_or ( false ) ; if aged_out { self . persist_age_counter . inc ( 1 ) ; } let is_cold = now . checked_duration_since ( s . last_write ) . map ( | age | age > self . config . partition_cold_threshold ) . unwrap_or ( false ) ; if is_cold { self . persist_cold_counter . inc ( 1 ) ; } let sized_out = s . bytes_written > self . config . partition_size_threshold ; if sized_out { self . persist_size_counter . inc ( 1 ) ; } aged_out || sized_out || is_cold } ) ; // keep track of what we'll be evicting to see what else to drop for s in & to_persist { total_bytes -= s . bytes_written ; } // if we're still over the memory threshold, persist as many of the largest partitions // until we're under. It's ok if this is stale, it'll just get handled on the next pass // through. if total_bytes > self . config . persist_memory_threshold { rest . sort_by ( | a , b | b . bytes_written . cmp ( & a . bytes_written ) ) ; let mut remaining = vec ! [ ] ; let mut memory_persist_counter = 0 ; for s in rest { if total_bytes >= self . config . persist_memory_threshold { total_bytes -= s . bytes_written ; to_persist . push ( s ) ; memory_persist_counter += 1 ; } else { remaining . push ( s ) ; } } self . persist_memory_counter . inc ( memory_persist_counter ) ; rest = remaining ; } // for the sequencers that are getting data persisted, keep track of what // the highest seqeunce number was for each. let mut sequencer_maxes = BTreeMap :: new ( ) ; for s in & to_persist { sequencer_maxes . entry ( s . sequencer_id ) . and_modify ( | sn | { if * sn < s . first_sequence_number { * sn = s . first_sequence_number ; } } ) . or_insert ( s . first_sequence_number ) ; } let persist_tasks : Vec < _ > = to_persist . into_iter ( ) . map ( | s | { self . remove ( s . partition_id ) ; let persister = Arc :: clone ( persister ) ; let ( _tracker , registration ) = self . job_registry . register ( Job :: Persist { partition_id : s . partition_id , } ) ; tokio :: task :: spawn ( async move { persister . persist ( s . partition_id ) . await ; } ) . track ( registration ) } ) . collect ( ) ; if ! persist_tasks . is_empty ( ) { let persists = futures :: future :: join_all ( persist_tasks . into_iter ( ) ) ; let results = persists . await ; for res in results { res . expect ( " " ) . expect ( " " ) ; } // for the sequencers that had data persisted, update their min_unpersisted_sequence_number to // either the minimum remaining in everything that didn't get persisted, or the highest // number that was persisted. Marking the highest number as the state is ok because it // just needs to represent the farthest we'd have to seek back in the write buffer. Any // data replayed during recovery that has already been persisted will just be ignored. // // The calculation of the min unpersisted sequence number is: // - If there is any unpersisted data (i.e. `rest` contains entries for that sequencer) then we take the //   minimum sequence number from there. Note that there might be a gap between the data that we have just //   persisted and the unpersisted section. // - If there is NO unpersisted data, we take the max sequence number of the part that we have just //   persisted. Note that can cannot use "max + 1" because the lifecycle handler receives writes on a //   partition level and therefore might run while data for a single sequence number is added. So the max //   sequence number that we have just persisted might have more data. for ( sequencer_id , sequence_number ) in sequencer_maxes { let min = rest . iter ( ) . filter ( | s | s . sequencer_id == sequencer_id ) . map ( | s | s . first_sequence_number ) . min ( ) . unwrap_or ( sequence_number ) ; persister . update_min_unpersisted_sequence_number ( sequencer_id , min ) . await ; } } }
This will persist any partitions that are over their size or age thresholds and
persist as many partitions as necessary (largest first) to get below the memory threshold.
This will persist any partitions that are over their size or age thresholds and persist as many partitions as necessary ( largest first ) to get below the memory threshold .
&mut self
persister
&Arc<P>
&mut self
persister
&Arc<P>