/// 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
fn check_for_work(
        &mut self,
        now: DateTime<Utc>,
        now_instant: Instant,
    ) -> BoxFuture<'static, ()> {
        let rules = self.rules();
        let chunks = self.chunks(&rules.sort_order);
        let mut open_partitions: HashSet<String> = HashSet::new();
        // Only want to start a new move/write task if there isn't one already in-flight
        //
        // Fetch the trackers for tasks created by previous loop iterations. If the trackers exist
        // and haven't completed, we don't want to create a new task of that type as there
        // is still one in-flight.
        //
        // This is a very weak heuristic for preventing background work from starving query
        // workload - i.e. only use one thread for each type of background task.
        //
        // We may want to revisit this in future
        let mut move_tracker = self.move_tracker().filter(|x| !x.is_complete()).cloned();
        let mut write_tracker = self.write_tracker().filter(|x| !x.is_complete()).cloned();
        // Iterate through the chunks to determine
        // - total memory consumption
        // - any chunks to move
        for chunk in &chunks {
            let chunk_guard = chunk.read();
            let would_move = can_move(&rules, &*chunk_guard, now);
            let would_write = write_tracker.is_none() && rules.persist;
            if let Some(lifecycle_action) = chunk_guard.lifecycle_action() {
                if lifecycle_action.is_complete()
                    && now_instant.duration_since(lifecycle_action.start_instant())
                        >= LIFECYCLE_ACTION_BACKOFF
                {
                    std::mem::drop(chunk_guard);
                    chunk
                        .write()
                        .clear_lifecycle_action()
                        .expect("failed to clear lifecycle action");
                }
                continue;
            }
            match chunk_guard.stage() {
                ChunkStage::Open { .. } => {
                    open_partitions.insert(chunk_guard.key().to_string());
                    if move_tracker.is_none() && would_move {
                        let partition_key = chunk_guard.key().to_string();
                        let table_name = chunk_guard.table_name().to_string();
                        let chunk_id = chunk_guard.id();
                        std::mem::drop(chunk_guard);
                        move_tracker =
                            Some(self.move_to_read_buffer(partition_key, table_name, chunk_id));
                    }
                }
                ChunkStage::Frozen { representation, .. } => match &representation {
                    ChunkStageFrozenRepr::MutableBufferSnapshot(_) if move_tracker.is_none() => {
                        let partition_key = chunk_guard.key().to_string();
                        let table_name = chunk_guard.table_name().to_string();
                        let chunk_id = chunk_guard.id();
                        std::mem::drop(chunk_guard);
                        move_tracker =
                            Some(self.move_to_read_buffer(partition_key, table_name, chunk_id));
                    }
                    ChunkStageFrozenRepr::ReadBuffer { .. } if would_write => {
                        let partition_key = chunk_guard.key().to_string();
                        let table_name = chunk_guard.table_name().to_string();
                        let chunk_id = chunk_guard.id();
                        std::mem::drop(chunk_guard);
                        write_tracker =
                            Some(self.write_to_object_store(partition_key, table_name, chunk_id));
                    }
                    _ => {}
                },
                // TODO: unload read buffer (https://github.com/influxdata/influxdb_iox/issues/1400)
                _ => {}
            }
        }
        if let Some(soft_limit) = rules.buffer_size_soft {
            let mut chunks = chunks.iter();
            loop {
                let buffer_size = self.buffer_size();
                if buffer_size < soft_limit.get() {
                    break;
                }
                // Dropping chunks that are kept in memory by queries frees no memory
                // TODO: LRU drop policy
                match chunks.next() {
                    Some(chunk) => {
                        let chunk_guard = chunk.read();
                        if chunk_guard.lifecycle_action().is_some() {
                            // Cannot drop chunk with in-progress lifecycle action
                            continue;
                        }
                        if (rules.drop_non_persisted
                            && matches!(
                                chunk_guard.stage(),
                                ChunkStage::Frozen {
                                    representation: ChunkStageFrozenRepr::ReadBuffer(_),
                                    ..
                                }
                            ))
                            || matches!(chunk_guard.stage(), ChunkStage::Persisted { .. })
                        {
                            let partition_key = chunk_guard.key().to_string();
                            let table_name = chunk_guard.table_name().to_string();
                            let chunk_id = chunk_guard.id();
                            std::mem::drop(chunk_guard);
                            if open_partitions.contains(&partition_key) {
                                warn!(db_name=self.db_name(), %partition_key, chunk_id, soft_limit, buffer_size,
                                      "dropping chunk from partition containing open chunk. Consider increasing the soft buffer limit");
                            }
                            self.drop_chunk(partition_key, table_name, chunk_id)
                        }
                    }
                    None => {
                        warn!(db_name=self.db_name(), soft_limit, buffer_size,
                              "soft limited exceeded, but no chunks found that can be evicted. Check lifecycle rules");
                        break;
                    }
                }
            }
        }
        Box::pin(async move {
            let backoff = rules
                .worker_backoff_millis
                .map(|x| Duration::from_millis(x.get()))
                .unwrap_or(DEFAULT_LIFECYCLE_BACKOFF);
            // `check_for_work` should be called again if either 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
            tokio::select! {
                _ = tokio::time::sleep(backoff) => {}
                _ = wait_optional_tracker(move_tracker) => {}
                _ = wait_optional_tracker(write_tracker) => {}
            };
        })
    }
fn check_for_work ( & mut self , now : DateTime < Utc > , now_instant : Instant , ) -> BoxFuture < ' static , ( ) > { let rules = self . rules ( ) ; let chunks = self . chunks ( & rules . sort_order ) ; let mut open_partitions : HashSet < String > = HashSet :: new ( ) ; // Only want to start a new move/write task if there isn't one already in-flight // // Fetch the trackers for tasks created by previous loop iterations. If the trackers exist // and haven't completed, we don't want to create a new task of that type as there // is still one in-flight. // // This is a very weak heuristic for preventing background work from starving query // workload - i.e. only use one thread for each type of background task. // // We may want to revisit this in future let mut move_tracker = self . move_tracker ( ) . filter ( | x | ! x . is_complete ( ) ) . cloned ( ) ; let mut write_tracker = self . write_tracker ( ) . filter ( | x | ! x . is_complete ( ) ) . cloned ( ) ; // Iterate through the chunks to determine // - total memory consumption // - any chunks to move for chunk in & chunks { let chunk_guard = chunk . read ( ) ; let would_move = can_move ( & rules , & * chunk_guard , now ) ; let would_write = write_tracker . is_none ( ) && rules . persist ; if let Some ( lifecycle_action ) = chunk_guard . lifecycle_action ( ) { if lifecycle_action . is_complete ( ) && now_instant . duration_since ( lifecycle_action . start_instant ( ) ) >= LIFECYCLE_ACTION_BACKOFF { std :: mem :: drop ( chunk_guard ) ; chunk . write ( ) . clear_lifecycle_action ( ) . expect ( " " ) ; } continue ; } match chunk_guard . stage ( ) { ChunkStage :: Open { .. } => { open_partitions . insert ( chunk_guard . key ( ) . to_string ( ) ) ; if move_tracker . is_none ( ) && would_move { let partition_key = chunk_guard . key ( ) . to_string ( ) ; let table_name = chunk_guard . table_name ( ) . to_string ( ) ; let chunk_id = chunk_guard . id ( ) ; std :: mem :: drop ( chunk_guard ) ; move_tracker = Some ( self . move_to_read_buffer ( partition_key , table_name , chunk_id ) ) ; } } ChunkStage :: Frozen { representation , .. } => match & representation { ChunkStageFrozenRepr :: MutableBufferSnapshot ( _ ) if move_tracker . is_none ( ) => { let partition_key = chunk_guard . key ( ) . to_string ( ) ; let table_name = chunk_guard . table_name ( ) . to_string ( ) ; let chunk_id = chunk_guard . id ( ) ; std :: mem :: drop ( chunk_guard ) ; move_tracker = Some ( self . move_to_read_buffer ( partition_key , table_name , chunk_id ) ) ; } ChunkStageFrozenRepr :: ReadBuffer { .. } if would_write => { let partition_key = chunk_guard . key ( ) . to_string ( ) ; let table_name = chunk_guard . table_name ( ) . to_string ( ) ; let chunk_id = chunk_guard . id ( ) ; std :: mem :: drop ( chunk_guard ) ; write_tracker = Some ( self . write_to_object_store ( partition_key , table_name , chunk_id ) ) ; } _ => { } } , // TODO: unload read buffer (https://github.com/influxdata/influxdb_iox/issues/1400) _ => { } } } if let Some ( soft_limit ) = rules . buffer_size_soft { let mut chunks = chunks . iter ( ) ; loop { let buffer_size = self . buffer_size ( ) ; if buffer_size < soft_limit . get ( ) { break ; } // Dropping chunks that are kept in memory by queries frees no memory // TODO: LRU drop policy match chunks . next ( ) { Some ( chunk ) => { let chunk_guard = chunk . read ( ) ; if chunk_guard . lifecycle_action ( ) . is_some ( ) { // Cannot drop chunk with in-progress lifecycle action continue ; } if ( rules . drop_non_persisted && matches ! ( chunk_guard stage ( ) ChunkStage Frozen { representation ChunkStageFrozenRepr ReadBuffer ( ) } ) ) || matches ! ( chunk_guard stage ( ) ChunkStage Persisted { } ) { let partition_key = chunk_guard . key ( ) . to_string ( ) ; let table_name = chunk_guard . table_name ( ) . to_string ( ) ; let chunk_id = chunk_guard . id ( ) ; std :: mem :: drop ( chunk_guard ) ; if open_partitions . contains ( & partition_key ) { warn ! ( db_name self db_name ( ) partition_key chunk_id soft_limit buffer_size " " ) ; } self . drop_chunk ( partition_key , table_name , chunk_id ) } } None => { warn ! ( db_name self db_name ( ) soft_limit buffer_size " " ) ; break ; } } } } Box :: pin ( async move { let backoff = rules . worker_backoff_millis . map ( | x | Duration :: from_millis ( x . get ( ) ) ) . unwrap_or ( DEFAULT_LIFECYCLE_BACKOFF ) ; // `check_for_work` should be called again if either 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 tokio :: select ! { tokio time sleep ( backoff ) { } wait_optional_tracker ( move_tracker ) { } wait_optional_tracker ( write_tracker ) { } } ; } ) }
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