pub fn check_for_work(
        &mut self,
        now: DateTime<Utc>,
        now_instant: Instant,
    ) -> BoxFuture<'_, ()> {
        // Any time-consuming work should be spawned as tokio tasks and not
        // run directly within this loop

        // TODO: Add loop iteration count and duration metrics

        let db_name = self.db.name();
        let rules = self.db.rules();
        let partitions = self.db.partitions();

        for partition in &partitions {
            self.maybe_cleanup_failed(&db_name, partition, now_instant);

            // Persistence cannot split chunks if they are currently being compacted
            //
            // To avoid compaction "starving" persistence we employ a
            // heavy-handed approach of temporarily pausing compaction
            // if the criteria for persistence have been satisfied,
            // but persistence cannot proceed because of in-progress
            // compactions
            let stall_compaction_persisting = if rules.persist && !self.suppress_persistence {
                let persisting =
                    self.maybe_persist_chunks(&db_name, partition, &rules, now_instant);
                if persisting {
                    debug!(%db_name, %partition, reason="persisting", "stalling compaction");
                }
                persisting
            } else {
                false
            };

            // Until we have a more sophisticated compaction policy that can
            // allocate resources appropriately, we limit the number of
            // compactions that may run concurrently. Compactions are
            // completely disabled if max_compactions is Some(0), whilst if
            // it is None then the compaction limiter is disabled (unlimited
            // concurrent compactions).
            let stall_compaction_no_slots = {
                let max_compactions = self.db.rules().max_active_compactions.get();
                let slots_full = self.active_compactions >= max_compactions as usize;
                if slots_full {
                    debug!(%db_name, %partition, ?max_compactions, reason="slots_full", "stalling compaction");
                }
                slots_full
            };

            // conditions where no compactions will be scheduled.
            if stall_compaction_persisting || stall_compaction_no_slots {
                continue;
            }

            // possibly do a compaction
            self.maybe_compact_chunks(partition, &rules, now);
        }

        if let Some(soft_limit) = rules.buffer_size_soft {
            self.maybe_free_memory(
                &db_name,
                &partitions,
                soft_limit.get(),
                rules.drop_non_persisted,
            )
        }

        // Clear out completed tasks
        let mut completed_compactions = 0;
        self.trackers.retain(|x| {
            let completed = x.is_complete();
            if completed && matches!(x.metadata(), ChunkLifecycleAction::Compacting) {
                // free up slot for another compaction
                completed_compactions += 1;
            }

            !completed
        });

        // update active compactions
        if completed_compactions > 0 {
            debug!(?completed_compactions, active_compactions=?self.active_compactions,
                max_compactions=?self.db.rules().max_active_compactions, "releasing compaction slots")
        }

        assert!(completed_compactions <= self.active_compactions);
        self.active_compactions -= completed_compactions;

        let tracker_fut = match self.trackers.is_empty() {
            false => futures::future::Either::Left(futures::future::select_all(
                self.trackers.iter().map(|x| Box::pin(x.join())),
            )),
            true => futures::future::Either::Right(futures::future::pending()),
        };

        Box::pin(async move {
            let backoff = rules.worker_backoff_millis.get();

            // `check_for_work` should be called again if any of the tasks completes
            // or the backoff expires.
            //
            // This formulation ensures that the loop will run again in backoff
            // number of milliseconds regardless of if any tasks have finished
            //
            // Consider the situation where we have an in-progress write but no
            // in-progress move. We will look again for new tasks as soon
            // as either the backoff expires or the write task finishes
            //
            // Similarly if there are no in-progress tasks, we will look again
            // after the backoff interval
            //
            // Finally if all tasks are running we still want to be invoked
            // after the backoff interval, even if no tasks have completed by then,
            // as we may need to drop chunks to free up memory
            tokio::select! {
                _ = tokio::time::sleep(tokio::time::Duration::from_millis(backoff)) => {}
                _ = tracker_fut => {}
            };
        })
    }
/// The core policy logic
///
/// Returns a future that resolves when this method should be called next
The core policy logic
Returns a future that resolves when this method should be called next
The core policy logic Returns a future that resolves when this method should be called next
pub fn check_for_work(
        &mut self,
        now: DateTime<Utc>,
        now_instant: Instant,
    ) -> BoxFuture<'_, ()> {
        // Any time-consuming work should be spawned as tokio tasks and not
        // run directly within this loop
        // TODO: Add loop iteration count and duration metrics
        let db_name = self.db.name();
        let rules = self.db.rules();
        let partitions = self.db.partitions();
        for partition in &partitions {
            self.maybe_cleanup_failed(&db_name, partition, now_instant);
            // Persistence cannot split chunks if they are currently being compacted
            //
            // To avoid compaction "starving" persistence we employ a
            // heavy-handed approach of temporarily pausing compaction
            // if the criteria for persistence have been satisfied,
            // but persistence cannot proceed because of in-progress
            // compactions
            let stall_compaction_persisting = if rules.persist && !self.suppress_persistence {
                let persisting =
                    self.maybe_persist_chunks(&db_name, partition, &rules, now_instant);
                if persisting {
                    debug!(%db_name, %partition, reason="persisting", "stalling compaction");
                }
                persisting
            } else {
                false
            };
            // Until we have a more sophisticated compaction policy that can
            // allocate resources appropriately, we limit the number of
            // compactions that may run concurrently. Compactions are
            // completely disabled if max_compactions is Some(0), whilst if
            // it is None then the compaction limiter is disabled (unlimited
            // concurrent compactions).
            let stall_compaction_no_slots = {
                let max_compactions = self.db.rules().max_active_compactions.get();
                let slots_full = self.active_compactions >= max_compactions as usize;
                if slots_full {
                    debug!(%db_name, %partition, ?max_compactions, reason="slots_full", "stalling compaction");
                }
                slots_full
            };
            // conditions where no compactions will be scheduled.
            if stall_compaction_persisting || stall_compaction_no_slots {
                continue;
            }
            // possibly do a compaction
            self.maybe_compact_chunks(partition, &rules, now);
        }
        if let Some(soft_limit) = rules.buffer_size_soft {
            self.maybe_free_memory(
                &db_name,
                &partitions,
                soft_limit.get(),
                rules.drop_non_persisted,
            )
        }
        // Clear out completed tasks
        let mut completed_compactions = 0;
        self.trackers.retain(|x| {
            let completed = x.is_complete();
            if completed && matches!(x.metadata(), ChunkLifecycleAction::Compacting) {
                // free up slot for another compaction
                completed_compactions += 1;
            }
            !completed
        });
        // update active compactions
        if completed_compactions > 0 {
            debug!(?completed_compactions, active_compactions=?self.active_compactions,
                max_compactions=?self.db.rules().max_active_compactions, "releasing compaction slots")
        }
        assert!(completed_compactions <= self.active_compactions);
        self.active_compactions -= completed_compactions;
        let tracker_fut = match self.trackers.is_empty() {
            false => futures::future::Either::Left(futures::future::select_all(
                self.trackers.iter().map(|x| Box::pin(x.join())),
            )),
            true => futures::future::Either::Right(futures::future::pending()),
        };
        Box::pin(async move {
            let backoff = rules.worker_backoff_millis.get();
            // `check_for_work` should be called again if any of the tasks completes
            // or the backoff expires.
            //
            // This formulation ensures that the loop will run again in backoff
            // number of milliseconds regardless of if any tasks have finished
            //
            // Consider the situation where we have an in-progress write but no
            // in-progress move. We will look again for new tasks as soon
            // as either the backoff expires or the write task finishes
            //
            // Similarly if there are no in-progress tasks, we will look again
            // after the backoff interval
            //
            // Finally if all tasks are running we still want to be invoked
            // after the backoff interval, even if no tasks have completed by then,
            // as we may need to drop chunks to free up memory
            tokio::select! {
                _ = tokio::time::sleep(tokio::time::Duration::from_millis(backoff)) => {}
                _ = tracker_fut => {}
            };
        })
    }
pub fn check_for_work ( & mut self , now : DateTime < Utc > , now_instant : Instant , ) -> BoxFuture < ' _ , ( ) > { // Any time-consuming work should be spawned as tokio tasks and not // run directly within this loop // TODO: Add loop iteration count and duration metrics let db_name = self . db . name ( ) ; let rules = self . db . rules ( ) ; let partitions = self . db . partitions ( ) ; for partition in & partitions { self . maybe_cleanup_failed ( & db_name , partition , now_instant ) ; // Persistence cannot split chunks if they are currently being compacted // // To avoid compaction "starving" persistence we employ a // heavy-handed approach of temporarily pausing compaction // if the criteria for persistence have been satisfied, // but persistence cannot proceed because of in-progress // compactions let stall_compaction_persisting = if rules . persist && ! self . suppress_persistence { let persisting = self . maybe_persist_chunks ( & db_name , partition , & rules , now_instant ) ; if persisting { debug ! ( db_name partition reason " " " " ) ; } persisting } else { false } ; // Until we have a more sophisticated compaction policy that can // allocate resources appropriately, we limit the number of // compactions that may run concurrently. Compactions are // completely disabled if max_compactions is Some(0), whilst if // it is None then the compaction limiter is disabled (unlimited // concurrent compactions). let stall_compaction_no_slots = { let max_compactions = self . db . rules ( ) . max_active_compactions . get ( ) ; let slots_full = self . active_compactions >= max_compactions as usize ; if slots_full { debug ! ( db_name partition max_compactions reason " " " " ) ; } slots_full } ; // conditions where no compactions will be scheduled. if stall_compaction_persisting || stall_compaction_no_slots { continue ; } // possibly do a compaction self . maybe_compact_chunks ( partition , & rules , now ) ; } if let Some ( soft_limit ) = rules . buffer_size_soft { self . maybe_free_memory ( & db_name , & partitions , soft_limit . get ( ) , rules . drop_non_persisted , ) } // Clear out completed tasks let mut completed_compactions = 0 ; self . trackers . retain ( | x | { let completed = x . is_complete ( ) ; if completed && matches ! ( x metadata ( ) ChunkLifecycleAction Compacting ) { // free up slot for another compaction completed_compactions += 1 ; } ! completed } ) ; // update active compactions if completed_compactions > 0 { debug ! ( completed_compactions active_compactions self active_compactions max_compactions self db rules ( ) max_active_compactions " " ) } assert ! ( completed_compactions self active_compactions ) ; self . active_compactions -= completed_compactions ; let tracker_fut = match self . trackers . is_empty ( ) { false => futures :: future :: Either :: Left ( futures :: future :: select_all ( self . trackers . iter ( ) . map ( | x | Box :: pin ( x . join ( ) ) ) , ) ) , true => futures :: future :: Either :: Right ( futures :: future :: pending ( ) ) , } ; Box :: pin ( async move { let backoff = rules . worker_backoff_millis . get ( ) ; // `check_for_work` should be called again if any of the tasks completes // or the backoff expires. // // This formulation ensures that the loop will run again in backoff // number of milliseconds regardless of if any tasks have finished // // Consider the situation where we have an in-progress write but no // in-progress move. We will look again for new tasks as soon // as either the backoff expires or the write task finishes // // Similarly if there are no in-progress tasks, we will look again // after the backoff interval // // Finally if all tasks are running we still want to be invoked // after the backoff interval, even if no tasks have completed by then, // as we may need to drop chunks to free up memory tokio :: select ! { tokio time sleep ( tokio time Duration from_millis ( backoff ) ) { } tracker_fut { } } ; } ) }
The core policy logic
Returns a future that resolves when this method should be called next
The core policy logic Returns a future that resolves when this method should be called next
&mut self
now
DateTime<Utc>
now_instant
Instant
&mut self
now
DateTime<Utc>
now_instant
Instant