Fix compaction buckets priority

Author: Guilucand

crates/api/src/lib.rs

@@ -5,6 +5,7 @@ use colors::colors_manager::ColorsManager;
 use colors::{
     bundles::multifile_building::ColorBundleMultifileBuilding, non_colored::NonColoredManager,
 };
+use config::{MAX_BUCKET_CHUNK_SIZE, MIN_BUCKET_CHUNK_SIZE};
 pub use ggcat_logging::MessageLevel;
 use ggcat_logging::UnrecoverableErrorLogging;
 use io::concurrent::structured_sequences::fasta::FastaWriterWrapper;
@@ -272,7 +273,11 @@ impl GGCATInstance {
                 None
             } else {
                 // Heuristic for chunks used for maximum disk usage
-                Some((estimated_bases_count as u64) / (disk_optimization_level as u64 + 1))
+                Some(
+                    ((estimated_bases_count as u64) / (disk_optimization_level as u64 + 1))
+                        .min(MAX_BUCKET_CHUNK_SIZE)
+                        .max(MIN_BUCKET_CHUNK_SIZE),
+                )
             }
         };
 

crates/config/src/lib.rs

@@ -80,6 +80,9 @@ pub const PRIORITY_SCHEDULING_HIGH: usize = 0;
 pub const PRIORITY_SCHEDULING_BASE: usize = 1;
 pub const PRIORITY_SCHEDULING_LOW: usize = 2;
 
+pub const MIN_BUCKET_CHUNK_SIZE: u64 = 1024 * 1024 * 8;
+pub const MAX_BUCKET_CHUNK_SIZE: u64 = 1024 * 1024 * 1024;
+
 pub struct SwapPriority {}
 #[allow(non_upper_case_globals)]
 impl SwapPriority {

crates/minimizer_bucketing/src/compactor.rs

@@ -170,20 +170,31 @@ impl<E: MinimizerBucketingExecutorFactory + Sync + Send + 'static> AsyncExecutor
                 buckets[bucket_index].chunks.sort_by_cached_key(|c| {
                     let file_size = MemoryFs::get_file_size(c).unwrap();
                     total_size += file_size;
-                    Reverse(file_size)
+                    let is_compacted = c
+                        .file_name()
+                        .unwrap()
+                        .to_str()
+                        .unwrap()
+                        .contains("compacted");
+                    Reverse((is_compacted, file_size))
                 });
 
                 let mut chosen_size = 0;
 
                 // Choose the buckets to compact, taking all the buckets that strictly do not exceed half of the total buckets size.
                 // this allows to keep a linear time complexity
 
+                // println!(
+                //     "Buckets tot size: {} tbc: {:?}",
+                //     total_size, buckets[bucket_index].chunks
+                // );
+
                 let mut last = buckets[bucket_index].chunks.pop().unwrap();
                 let mut last_size = MemoryFs::get_file_size(&last).unwrap();
 
                 // Choose buckets until one of two conditions is met:
                 // 1. The next bucket would add up to a size greater than half ot the total size
-                // 2. Two buckets were already selected and the number of sequences is greater than the maximum amount
+                // 2. Four buckets were already selected and the number of sequences is greater than the maximum amount
                 // The second condition is checked below, after the processing of each bucket
                 while chosen_size + last_size < total_size / 2 {
                     chosen_size += last_size;
@@ -201,6 +212,9 @@ impl<E: MinimizerBucketingExecutorFactory + Sync + Send + 'static> AsyncExecutor
                     buckets[bucket_index]
                         .chunks
                         .extend(chosen_buckets.drain(..));
+                }
+
+                if chosen_buckets.is_empty() {
                     continue;
                 }
 
@@ -295,7 +309,7 @@ impl<E: MinimizerBucketingExecutorFactory + Sync + Send + 'static> AsyncExecutor
 
                     // Do not process more buckets if it will increase the maximum number of allowed sequences
                     if !global_params.common.is_active.load(Ordering::Relaxed)
-                        || processed_buckets >= 2 && total_sequences > MAXIMUM_SEQUENCES
+                        || processed_buckets >= 4 && total_sequences > MAXIMUM_SEQUENCES
                     {
                         break;
                     }