fn maybe_free_memory<P: LockablePartition>(
        &mut self,
        db_name: &DatabaseName<'static>,
        partitions: &[P],
        soft_limit: usize,
        drop_non_persisted: bool,
    ) {
        let buffer_size = self.db.buffer_size();
        if buffer_size < soft_limit {
            debug!(%db_name, buffer_size, %soft_limit, "memory use under soft limit");
            return;
        }

        // Collect a list of candidates to free memory
        let mut candidates = Vec::new();
        for partition in partitions {
            let guard = partition.read();
            for (_, chunk) in LockablePartition::chunks(&guard) {
                let chunk = chunk.read();
                if chunk.lifecycle_action().is_some() {
                    continue;
                }

                let action = match chunk.storage() {
                    ChunkStorage::ReadBuffer | ChunkStorage::ClosedMutableBuffer
                        if drop_non_persisted =>
                    {
                        FreeAction::Drop
                    }
                    ChunkStorage::ReadBufferAndObjectStore => FreeAction::Unload,
                    _ => continue,
                };

                candidates.push(FreeCandidate {
                    partition,
                    action,
                    chunk_id: chunk.addr().chunk_id,
                    access_metrics: chunk.access_metrics(),
                })
            }
        }

        sort_free_candidates(&mut candidates);
        let mut candidates = candidates.into_iter();

        // Loop through trying to free memory
        //
        // There is an intentional lock gap here, to avoid holding read locks on all
        // the droppable chunks within the database. The downside is we have to
        // re-check pre-conditions in case they no longer hold
        //
        loop {
            let buffer_size = self.db.buffer_size();
            if buffer_size < soft_limit {
                info!(%db_name, buffer_size, %soft_limit, "memory use under soft limit");
                break;
            }
            info!(%db_name, buffer_size, %soft_limit, "memory use over soft limit");

            match candidates.next() {
                Some(candidate) => {
                    let partition = candidate.partition.read();
                    match LockablePartition::chunk(&partition, candidate.chunk_id) {
                        Some(chunk) => {
                            let chunk = chunk.read();
                            if chunk.lifecycle_action().is_some() {
                                info!(
                                    %db_name,
                                    chunk_id = candidate.chunk_id,
                                    %partition,
                                    "cannot mutate chunk with in-progress lifecycle action"
                                );
                                continue;
                            }

                            match candidate.action {
                                FreeAction::Drop => match chunk.storage() {
                                    ChunkStorage::ReadBuffer
                                    | ChunkStorage::ClosedMutableBuffer => {
                                        let tracker = LockablePartition::drop_chunk(
                                            partition.upgrade(),
                                            chunk.upgrade(),
                                        )
                                        .expect("failed to drop")
                                        .with_metadata(ChunkLifecycleAction::Dropping);
                                        self.trackers.push(tracker);
                                    }
                                    storage => warn!(
                                        %db_name,
                                        chunk_id = candidate.chunk_id,
                                        %partition,
                                        ?storage,
                                        "unexpected storage for drop"
                                    ),
                                },
                                FreeAction::Unload => match chunk.storage() {
                                    ChunkStorage::ReadBufferAndObjectStore => {
                                        LockableChunk::unload_read_buffer(chunk.upgrade())
                                            .expect("failed to unload")
                                    }
                                    storage => warn!(
                                        %db_name,
                                        chunk_id = candidate.chunk_id,
                                        %partition,
                                        ?storage,
                                        "unexpected storage for unload"
                                    ),
                                },
                            }
                        }
                        None => info!(
                            %db_name,
                            chunk_id = candidate.chunk_id,
                            %partition,
                            "cannot drop chunk that no longer exists on partition"
                        ),
                    }
                }
                None => {
                    warn!(%db_name, soft_limit, buffer_size,
                          "soft limited exceeded, but no chunks found that can be evicted. Check lifecycle rules");
                    break;
                }
            }
        }
    }
/// Check if database exceeds memory limits and free memory if necessary
///
/// The policy will first try to unload persisted chunks in order of creation
/// time, starting with the oldest.
///
/// If permitted by the lifecycle policy, it will then drop unpersisted chunks,
/// also in order of creation time, starting with the oldest.
///
/// TODO: use LRU instead of creation time
///
Check if database exceeds memory limits and free memory if necessary
The policy will first try to unload persisted chunks in order of creation
time, starting with the oldest.

If permitted by the lifecycle policy, it will then drop unpersisted chunks,
also in order of creation time, starting with the oldest.

use LRU instead of creation time
Check if database exceeds memory limits and free memory if necessary The policy will first try to unload persisted chunks in order of creation time starting with the oldest . If permitted by the lifecycle policy it will then drop unpersisted chunks also in order of creation time starting with the oldest . use LRU instead of creation time
fn maybe_free_memory<P: LockablePartition>(
        &mut self,
        db_name: &DatabaseName<'static>,
        partitions: &[P],
        soft_limit: usize,
        drop_non_persisted: bool,
    ) {
        let buffer_size = self.db.buffer_size();
        if buffer_size < soft_limit {
            debug!(%db_name, buffer_size, %soft_limit, "memory use under soft limit");
            return;
        }
        // Collect a list of candidates to free memory
        let mut candidates = Vec::new();
        for partition in partitions {
            let guard = partition.read();
            for (_, chunk) in LockablePartition::chunks(&guard) {
                let chunk = chunk.read();
                if chunk.lifecycle_action().is_some() {
                    continue;
                }
                let action = match chunk.storage() {
                    ChunkStorage::ReadBuffer | ChunkStorage::ClosedMutableBuffer
                        if drop_non_persisted =>
                    {
                        FreeAction::Drop
                    }
                    ChunkStorage::ReadBufferAndObjectStore => FreeAction::Unload,
                    _ => continue,
                };
                candidates.push(FreeCandidate {
                    partition,
                    action,
                    chunk_id: chunk.addr().chunk_id,
                    access_metrics: chunk.access_metrics(),
                })
            }
        }
        sort_free_candidates(&mut candidates);
        let mut candidates = candidates.into_iter();
        // Loop through trying to free memory
        //
        // There is an intentional lock gap here, to avoid holding read locks on all
        // the droppable chunks within the database. The downside is we have to
        // re-check pre-conditions in case they no longer hold
        //
        loop {
            let buffer_size = self.db.buffer_size();
            if buffer_size < soft_limit {
                info!(%db_name, buffer_size, %soft_limit, "memory use under soft limit");
                break;
            }
            info!(%db_name, buffer_size, %soft_limit, "memory use over soft limit");
            match candidates.next() {
                Some(candidate) => {
                    let partition = candidate.partition.read();
                    match LockablePartition::chunk(&partition, candidate.chunk_id) {
                        Some(chunk) => {
                            let chunk = chunk.read();
                            if chunk.lifecycle_action().is_some() {
                                info!(
                                    %db_name,
                                    chunk_id = candidate.chunk_id,
                                    %partition,
                                    "cannot mutate chunk with in-progress lifecycle action"
                                );
                                continue;
                            }
                            match candidate.action {
                                FreeAction::Drop => match chunk.storage() {
                                    ChunkStorage::ReadBuffer
                                    | ChunkStorage::ClosedMutableBuffer => {
                                        let tracker = LockablePartition::drop_chunk(
                                            partition.upgrade(),
                                            chunk.upgrade(),
                                        )
                                        .expect("failed to drop")
                                        .with_metadata(ChunkLifecycleAction::Dropping);
                                        self.trackers.push(tracker);
                                    }
                                    storage => warn!(
                                        %db_name,
                                        chunk_id = candidate.chunk_id,
                                        %partition,
                                        ?storage,
                                        "unexpected storage for drop"
                                    ),
                                },
                                FreeAction::Unload => match chunk.storage() {
                                    ChunkStorage::ReadBufferAndObjectStore => {
                                        LockableChunk::unload_read_buffer(chunk.upgrade())
                                            .expect("failed to unload")
                                    }
                                    storage => warn!(
                                        %db_name,
                                        chunk_id = candidate.chunk_id,
                                        %partition,
                                        ?storage,
                                        "unexpected storage for unload"
                                    ),
                                },
                            }
                        }
                        None => info!(
                            %db_name,
                            chunk_id = candidate.chunk_id,
                            %partition,
                            "cannot drop chunk that no longer exists on partition"
                        ),
                    }
                }
                None => {
                    warn!(%db_name, soft_limit, buffer_size,
                          "soft limited exceeded, but no chunks found that can be evicted. Check lifecycle rules");
                    break;
                }
            }
        }
    }
fn maybe_free_memory < P : LockablePartition > ( & mut self , db_name : & DatabaseName < ' static > , partitions : & [ P ] , soft_limit : usize , drop_non_persisted : bool , ) { let buffer_size = self . db . buffer_size ( ) ; if buffer_size < soft_limit { debug ! ( db_name buffer_size soft_limit " " ) ; return ; } // Collect a list of candidates to free memory let mut candidates = Vec :: new ( ) ; for partition in partitions { let guard = partition . read ( ) ; for ( _ , chunk ) in LockablePartition :: chunks ( & guard ) { let chunk = chunk . read ( ) ; if chunk . lifecycle_action ( ) . is_some ( ) { continue ; } let action = match chunk . storage ( ) { ChunkStorage :: ReadBuffer | ChunkStorage :: ClosedMutableBuffer if drop_non_persisted => { FreeAction :: Drop } ChunkStorage :: ReadBufferAndObjectStore => FreeAction :: Unload , _ => continue , } ; candidates . push ( FreeCandidate { partition , action , chunk_id : chunk . addr ( ) . chunk_id , access_metrics : chunk . access_metrics ( ) , } ) } } sort_free_candidates ( & mut candidates ) ; let mut candidates = candidates . into_iter ( ) ; // Loop through trying to free memory // // There is an intentional lock gap here, to avoid holding read locks on all // the droppable chunks within the database. The downside is we have to // re-check pre-conditions in case they no longer hold // loop { let buffer_size = self . db . buffer_size ( ) ; if buffer_size < soft_limit { info ! ( db_name buffer_size soft_limit " " ) ; break ; } info ! ( db_name buffer_size soft_limit " " ) ; match candidates . next ( ) { Some ( candidate ) => { let partition = candidate . partition . read ( ) ; match LockablePartition :: chunk ( & partition , candidate . chunk_id ) { Some ( chunk ) => { let chunk = chunk . read ( ) ; if chunk . lifecycle_action ( ) . is_some ( ) { info ! ( db_name chunk_id candidate chunk_id partition " " ) ; continue ; } match candidate . action { FreeAction :: Drop => match chunk . storage ( ) { ChunkStorage :: ReadBuffer | ChunkStorage :: ClosedMutableBuffer => { let tracker = LockablePartition :: drop_chunk ( partition . upgrade ( ) , chunk . upgrade ( ) , ) . expect ( " " ) . with_metadata ( ChunkLifecycleAction :: Dropping ) ; self . trackers . push ( tracker ) ; } storage => warn ! ( db_name chunk_id candidate chunk_id partition storage " " ) , } , FreeAction :: Unload => match chunk . storage ( ) { ChunkStorage :: ReadBufferAndObjectStore => { LockableChunk :: unload_read_buffer ( chunk . upgrade ( ) ) . expect ( " " ) } storage => warn ! ( db_name chunk_id candidate chunk_id partition storage " " ) , } , } } None => info ! ( db_name chunk_id candidate chunk_id partition " " ) , } } None => { warn ! ( db_name soft_limit buffer_size " " ) ; break ; } } } }
Check if database exceeds memory limits and free memory if necessary
The policy will first try to unload persisted chunks in order of creation
time, starting with the oldest.
Check if database exceeds memory limits and free memory if necessary The policy will first try to unload persisted chunks in order of creation time starting with the oldest .
&mut self
db_name
&DatabaseName<'static>
partitions
&[P]
soft_limit
usize
drop_non_persisted
bool
&mut self
db_name
&DatabaseName<'static>
partitions
&[P]
soft_limit
usize
drop_non_persisted
bool