disk index: pre-flight disk index grow

bucket_map/src/bucket.rs

@@ -13,6 +13,7 @@ use {
         },
         MaxSearch, RefCount,
     },
+    bv::BitVec,
     rand::{thread_rng, Rng},
     solana_measure::measure::Measure,
     solana_sdk::pubkey::Pubkey,
@@ -435,6 +436,7 @@ impl<'b, T: Clone + Copy + 'static> Bucket<T> {
 
     pub fn grow_index(&self, current_capacity_pow2: u8) {
         if self.index.contents.capacity_pow2() == current_capacity_pow2 {
+            let mut resize_verify_us = 0;
             let mut starting_size_pow2 = self.index.contents.capacity_pow2();
             if self.anticipated_size > 0 {
                 // start the growth at the next pow2 larger than what would be required to hold `anticipated_size`.
@@ -446,6 +448,52 @@ impl<'b, T: Clone + Copy + 'static> Bucket<T> {
             let mut count = 0;
             loop {
                 count += 1;
+                // grow relative to the current capacity
+                // todo: this needs to be page aligned if we do that below so cap is exactly the same
+                let random = thread_rng().gen();
+
+                let mut verify_us = Measure::start("resize_verify_us");
+                let new_capacity = 1 << (starting_size_pow2 + count);
+                if new_capacity > 1_000_000 {
+                    use log::*;error!("allocating new fill: {}, count: {}, max search: {}", new_capacity, count, self.index.max_search);
+                }
+                let mut occupied = BitVec::<u64>::new_fill(false, new_capacity);
+                let mut valid = true;
+                for ix in 0..self.index.capacity() {
+                    if !self.index.is_free(ix) {
+                        let elem: &IndexEntry<T> = self.index.get(ix);
+                        let ix_new = Self::bucket_index_ix(&self.index, &elem.key, random);
+                        valid = false;
+                        for look in 0..(self.index.max_search as u64) {
+                            let look = (ix_new + look) % new_capacity;
+                            if !occupied.get(look) {
+                                occupied.set(look, true);
+                                valid = true;
+                                break;
+                            }
+                        }
+                        if !valid {
+                            if new_capacity > 1_000_000 {
+                                use log::*;error!("invalid: allocating new fill: {}, count: {}, max search: {}, ix_new: {ix_new}, {:?}", new_capacity, count, self.index.max_search,
+                                (0..(self.index.max_search as u64)).map(|i| occupied.get((ix_new + i) % new_capacity)).collect::<Vec<_>>()
+                            );
+                            }
+
+                            break;
+                        }
+                    }
+                }
+                verify_us.stop();
+                resize_verify_us += verify_us.as_us();
+                if !valid {
+                    continue;
+                }
+
+                let nc = Capacity::Pow2(starting_size_pow2 + count);
+                if nc.capacity() > 1_000_000 {
+                    use log::*;error!("{}, size: {}", line!(), nc.capacity());
+                }
+
                 let mut index = BucketStorage::new_with_capacity(
                     Arc::clone(&self.drives),
                     1,
@@ -456,8 +504,6 @@ impl<'b, T: Clone + Copy + 'static> Bucket<T> {
                     Arc::clone(&self.stats.index),
                     Arc::clone(&self.index.count),
                 );
-                let random = thread_rng().gen();
-                let mut valid = true;
                 for ix in 0..self.index.capacity() {
                     if !self.index.is_free(ix) {
                         let elem: &IndexEntry<T> = self.index.get(ix);
@@ -499,6 +545,10 @@ impl<'b, T: Clone + Copy + 'static> Bucket<T> {
                 .index
                 .resize_us
                 .fetch_add(m.as_us(), Ordering::Relaxed);
+            self.stats
+                .index
+                .resize_verify_us
+                .fetch_add(resize_verify_us, Ordering::Relaxed);
         }
     }
 

bucket_map/src/bucket_stats.rs

@@ -9,6 +9,7 @@ pub struct BucketStats {
     pub failed_resizes: AtomicU64,
     pub max_size: AtomicU64,
     pub resize_us: AtomicU64,
+    pub resize_verify_us: AtomicU64,
     pub new_file_us: AtomicU64,
     pub flush_file_us: AtomicU64,
     pub mmap_us: AtomicU64,

bucket_map/src/bucket_storage.rs

@@ -353,6 +353,7 @@ impl<O: BucketOccupied> BucketStorage<O> {
         // the file so that we won't have to resize it later, which may be
         // expensive.
         //debug!("GROWING file {}", capacity * cell_size as u64);
+        //use log::*;error!("growing: {}", bytes);
         data.seek(SeekFrom::Start(bytes - 1)).unwrap();
         data.write_all(&[0]).unwrap();
         data.rewind().unwrap();

runtime/src/bucket_map_holder_stats.rs

@@ -452,6 +452,12 @@ impl BucketMapHolderStats {
                         .unwrap_or_default(),
                     i64
                 ),
+                (
+                    "resize_verify_us",
+                    disk.map(|disk| disk.stats.index.resize_verify_us.swap(0, Ordering::Relaxed))
+                        .unwrap_or_default(),
+                    i64
+                ),
                 (
                     "disk_index_find_index_entry_mut_us",
                     disk.map(|disk| disk