/// Checks for candidate partitions to compact and spawns tokio tasks to compact as many
/// as the configuration will allow. Once those are done it rechecks the catalog for the
/// next top partitions to compact.
Checks for candidate partitions to compact and spawns tokio tasks to compact as many
as the configuration will allow. Once those are done it rechecks the catalog for the
next top partitions to compact.
Checks for candidate partitions to compact and spawns tokio tasks to compact as many as the configuration will allow . Once those are done it rechecks the catalog for the next top partitions to compact .
async fn run_compactor(compactor: Arc<Compactor>, shutdown: CancellationToken) {
    while !shutdown.is_cancelled() {
        let candidates = Backoff::new(&compactor.backoff_config)
            .retry_all_errors("partitions_to_compact", || async {
                compactor.partitions_to_compact().await
            })
            .await
            .expect("retry forever");
        let n_candidates = candidates.len();
        debug!(n_candidates, "found compaction candidates");
        let mut used_size = 0;
        let max_size = compactor.config.max_concurrent_compaction_size_bytes();
        let max_file_size = compactor.config.compaction_max_size_bytes();
        let max_file_count = compactor.config.compaction_max_file_count();
        let mut handles = vec![];
        for c in candidates {
            let compactor = Arc::clone(&compactor);
            let compact_and_upgrade = compactor
                .groups_to_compact_and_files_to_upgrade(
                    c.partition_id,
                    max_file_size,
                    max_file_count,
                )
                .await;
            match compact_and_upgrade {
                Err(e) => {
                    warn!(
                        "groups file to compact and upgrade on partition {} failed with: {:?}",
                        c.partition_id, e
                    );
                }
                Ok(compact_and_upgrade) => {
                    if compact_and_upgrade.compactable() {
                        let handle = tokio::task::spawn(async move {
                            debug!(candidate=?c, "compacting candidate");
                            let res = compactor
                                .compact_partition(c.partition_id, compact_and_upgrade)
                                .await;
                            if let Err(e) = res {
                                warn!(
                                    "compaction on partition {} failed with: {:?}",
                                    c.partition_id, e
                                );
                            }
                            debug!(candidate=?c, "compaction complete");
                        });
                        used_size += c.file_size_bytes;
                        handles.push(handle);
                        if used_size > max_size {
                            debug!(%max_size, %used_size, n_compactions=%handles.len(), "reached maximum concurrent compaction size limit");
                            break;
                        }
                    }
                }
            }
        }
        let compactions_run = handles.len();
        debug!(
            ?compactions_run,
            "Number of concurrent partitions are being compacted in this cycle"
        );
        let _ = futures::future::join_all(handles).await;
        // if all candidate partitions have been compacted, wait a bit
        // before checking again, waking early if cancel arrives
        if compactions_run == n_candidates {
            select! {
                () = tokio::time::sleep(Duration::from_secs(5)).fuse() => {},
                () = shutdown.cancelled().fuse() => {}
            }
        }
    }
}
async fn run_compactor ( compactor : Arc < Compactor > , shutdown : CancellationToken ) { while ! shutdown . is_cancelled ( ) { let candidates = Backoff :: new ( & compactor . backoff_config ) . retry_all_errors ( " " , | | async { compactor . partitions_to_compact ( ) . await } ) . await . expect ( " " ) ; let n_candidates = candidates . len ( ) ; debug ! ( n_candidates " " ) ; let mut used_size = 0 ; let max_size = compactor . config . max_concurrent_compaction_size_bytes ( ) ; let max_file_size = compactor . config . compaction_max_size_bytes ( ) ; let max_file_count = compactor . config . compaction_max_file_count ( ) ; let mut handles = vec ! [ ] ; for c in candidates { let compactor = Arc :: clone ( & compactor ) ; let compact_and_upgrade = compactor . groups_to_compact_and_files_to_upgrade ( c . partition_id , max_file_size , max_file_count , ) . await ; match compact_and_upgrade { Err ( e ) => { warn ! ( " " c partition_id e ) ; } Ok ( compact_and_upgrade ) => { if compact_and_upgrade . compactable ( ) { let handle = tokio :: task :: spawn ( async move { debug ! ( candidate c " " ) ; let res = compactor . compact_partition ( c . partition_id , compact_and_upgrade ) . await ; if let Err ( e ) = res { warn ! ( " " c partition_id e ) ; } debug ! ( candidate c " " ) ; } ) ; used_size += c . file_size_bytes ; handles . push ( handle ) ; if used_size > max_size { debug ! ( max_size used_size n_compactions handles len ( ) " " ) ; break ; } } } } } let compactions_run = handles . len ( ) ; debug ! ( compactions_run " " ) ; let _ = futures :: future :: join_all ( handles ) . await ; // if all candidate partitions have been compacted, wait a bit // before checking again, waking early if cancel arrives if compactions_run == n_candidates { select ! { ( ) tokio time sleep ( Duration from_secs ( 5 ) ) fuse ( ) { } ( ) shutdown cancelled ( ) fuse ( ) { } } } } }
Checks for candidate partitions to compact and spawns tokio tasks to compact as many
as the configuration will allow.
Checks for candidate partitions to compact and spawns tokio tasks to compact as many as the configuration will allow .
compactor
Arc<Compactor>
shutdown
CancellationToken