ci: install openssl for the Windows build (solana-labs#33356)

Switch confirmed_unrooted_slots from Vec<_> to HashSet<_> (solana-labs#33311)

The container is only used to check for inclusion of slots with
the .contains() method. This method is O(n) on a Vec<_> but O(1) on a
HashSet<_>.

add Restart structs for disk index (solana-labs#33361)

use bytemuck for disk bucket restart (solana-labs#33371)

disk bucket: init restart path (solana-labs#33375)

simple cleanup in bucket map (solana-labs#33376)

add disk bucket get_restart_file (solana-labs#33373)

* add disk bucket get_restart_file

* add get_restartable_buckets

pass RestartableBucket through disk index (solana-labs#33377)

data bucket holds RestartableBucket (solana-labs#33381)

Feature - better error codes for tx lamport check (solana-labs#33343)

Replaces `TransactionError::InstructionError(0, InstructionError::UnbalancedInstruction)` with `TransactionError::UnbalancedTransaction`.

Co-authored-by: Alexander Meißner <AlexanderMeissner@gmx.net>

DiskIdx: reuse disk bucket file if possible (solana-labs#33379)

diskidx: stats for created vs reused (solana-labs#33385)

solana-program - altbn128: add g1 & g2 compression

still fixing tests for point of infinity

feat: proof compression syscall working

add rust test to ci

remove prints

added c test

added sycall pricing

fixed ci checks

refactored altbn128 and compression

.github/workflows/release-artifacts.yml

@@ -47,7 +47,8 @@ jobs:
         id: build
         shell: bash
         run: |
-          export OPENSSL_DIR="C:\Program Files\OpenSSL"
+          choco install openssl
+          export OPENSSL_DIR="C:\Program Files\OpenSSL-Win64"
           choco install protoc
           export PROTOC="C:\ProgramData\chocolatey\lib\protoc\tools\bin\protoc.exe"
           source /tmp/env.sh

accounts-db/src/bucket_map_holder.rs

@@ -204,7 +204,15 @@ impl<T: IndexValue, U: DiskIndexValue + From<T> + Into<T>> BucketMapHolder<T, U>
             .unwrap_or(DEFAULT_AGE_TO_STAY_IN_CACHE);
 
         let mut bucket_config = BucketMapConfig::new(bins);
-        bucket_config.drives = config.as_ref().and_then(|config| config.drives.clone());
+        bucket_config.drives = config.as_ref().and_then(|config| {
+            bucket_config.restart_config_file = config.drives.as_ref().and_then(|drives| {
+                drives
+                    .first()
+                    .map(|drive| drive.join("accounts_index_restart"))
+            });
+            config.drives.clone()
+        });
+
         let mem_budget_mb = match config
             .as_ref()
             .map(|config| &config.index_limit_mb)

accounts-db/src/bucket_map_holder_stats.rs

@@ -226,6 +226,34 @@ impl BucketMapHolderStats {
         // sum of elapsed time in each thread
         let mut thread_time_elapsed_ms = elapsed_ms * storage.threads as u64;
         if disk.is_some() {
+            if startup || was_startup {
+                // these stats only apply at startup
+                datapoint_info!(
+                    "accounts_index_startup",
+                    (
+                        "entries_created",
+                        disk.map(|disk| disk
+                            .stats
+                            .index
+                            .startup
+                            .entries_created
+                            .swap(0, Ordering::Relaxed))
+                            .unwrap_or_default(),
+                        i64
+                    ),
+                    (
+                        "entries_reused",
+                        disk.map(|disk| disk
+                            .stats
+                            .index
+                            .startup
+                            .entries_reused
+                            .swap(0, Ordering::Relaxed))
+                            .unwrap_or_default(),
+                        i64
+                    ),
+                );
+            }
             datapoint_info!(
                 if startup || was_startup {
                     thread_time_elapsed_ms *= 2; // more threads are allocated during startup

bucket_map/Cargo.toml

@@ -12,6 +12,7 @@ edition = { workspace = true }
 
 [dependencies]
 bv = { workspace = true, features = ["serde"] }
+bytemuck = { workspace = true, features = ["derive"] }
 log = { workspace = true }
 memmap2 = { workspace = true }
 modular-bitfield = { workspace = true }

bucket_map/src/bucket.rs

@@ -11,14 +11,17 @@ use {
             DataBucket, IndexBucket, IndexEntry, IndexEntryPlaceInBucket, MultipleSlots,
             OccupiedEnum,
         },
+        restart::RestartableBucket,
         MaxSearch, RefCount,
     },
     rand::{thread_rng, Rng},
     solana_measure::measure::Measure,
     solana_sdk::pubkey::Pubkey,
     std::{
         collections::hash_map::DefaultHasher,
+        fs,
         hash::{Hash, Hasher},
+        num::NonZeroU64,
         ops::RangeBounds,
         path::PathBuf,
         sync::{
@@ -107,34 +110,66 @@ pub struct Bucket<T: Copy + PartialEq + 'static> {
     /// set to true once any entries have been deleted from the index.
     /// Deletes indicate that there can be free slots and that the full search range must be searched for an entry.
     at_least_one_entry_deleted: bool,
+
+    /// keep track of which index file this bucket is using so on restart we can try to reuse it
+    restartable_bucket: RestartableBucket,
 }
 
 impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
-    pub fn new(
+    pub(crate) fn new(
         drives: Arc<Vec<PathBuf>>,
         max_search: MaxSearch,
         stats: Arc<BucketMapStats>,
         count: Arc<AtomicU64>,
+        mut restartable_bucket: RestartableBucket,
     ) -> Self {
-        let (index, _file_name) = BucketStorage::new(
-            Arc::clone(&drives),
-            1,
-            std::mem::size_of::<IndexEntry<T>>() as u64,
-            max_search,
-            Arc::clone(&stats.index),
-            count,
-        );
-        stats.index.resize_grow(0, index.capacity_bytes());
+        let reuse_path = std::mem::take(&mut restartable_bucket.path);
+        let elem_size = NonZeroU64::new(std::mem::size_of::<IndexEntry<T>>() as u64).unwrap();
+        let (index, random) = reuse_path
+            .and_then(|path| {
+                // try to re-use the file this bucket was using last time we were running
+                restartable_bucket.get().and_then(|(_file_name, random)| {
+                    let result = BucketStorage::load_on_restart(
+                        path.clone(),
+                        elem_size,
+                        max_search,
+                        Arc::clone(&stats.index),
+                        count.clone(),
+                    )
+                    .map(|index| (index, random));
+                    if result.is_none() {
+                        // we couldn't re-use it, so delete it
+                        _ = fs::remove_file(path);
+                    }
+                    result
+                })
+            })
+            .unwrap_or_else(|| {
+                // no file to re-use, so create a new file
+                let (index, file_name) = BucketStorage::new(
+                    Arc::clone(&drives),
+                    1,
+                    elem_size.into(),
+                    max_search,
+                    Arc::clone(&stats.index),
+                    count,
+                );
+                stats.index.resize_grow(0, index.capacity_bytes());
+                let random = thread_rng().gen();
+                restartable_bucket.set_file(file_name, random);
+                (index, random)
+            });
 
         Self {
-            random: thread_rng().gen(),
+            random,
             drives,
             index,
             data: vec![],
             stats,
             reallocated: Reallocated::default(),
             anticipated_size: 0,
             at_least_one_entry_deleted: false,
+            restartable_bucket,
         }
     }
 
@@ -303,6 +338,7 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
         self.set_anticipated_count((anticipated).saturating_add(current_len));
         let mut entries = Self::index_entries(items, self.random);
         let mut duplicates = Vec::default();
+        let mut entries_created_on_disk = 0;
         // insert, but resizes may be necessary
         loop {
             let cap = self.index.capacity();
@@ -316,6 +352,7 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
                 &self.data,
                 items,
                 &mut entries,
+                &mut entries_created_on_disk,
                 &mut duplicates,
             );
             match result {
@@ -326,6 +363,18 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
                         items.len().saturating_sub(duplicates.len()) as u64,
                         Ordering::Relaxed,
                     );
+                    self.index.stats.startup.entries_reused.fetch_add(
+                        items
+                            .len()
+                            .saturating_sub(duplicates.len())
+                            .saturating_sub(entries_created_on_disk) as u64,
+                        Ordering::Relaxed,
+                    );
+                    self.index
+                        .stats
+                        .startup
+                        .entries_created
+                        .fetch_add(entries_created_on_disk as u64, Ordering::Relaxed);
                     return duplicates;
                 }
                 Err(error) => {
@@ -342,13 +391,18 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
     /// insert as much of `entries` as possible into `index`.
     /// return an error if the index needs to resize.
     /// for every entry that already exists in `index`, add it (and the value already in the index) to `duplicates`
+    /// `reverse_sorted_entries` is (raw index (range = U64::MAX) in hash map, index in `items`)
     pub fn batch_insert_non_duplicates_internal(
         index: &mut BucketStorage<IndexBucket<T>>,
         data_buckets: &[BucketStorage<DataBucket>],
         items: &[(Pubkey, T)],
         reverse_sorted_entries: &mut Vec<(u64, usize)>,
+        entries_created_on_disk: &mut usize,
         duplicates: &mut Vec<(usize, T)>,
     ) -> Result<(), BucketMapError> {
+        if reverse_sorted_entries.is_empty() {
+            return Ok(());
+        }
         let max_search = index.max_search();
         let cap = index.capacity();
         let search_end = max_search.min(cap);
@@ -362,6 +416,7 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
                 let ix_index = (ix_entry + search) % cap;
                 let elem = IndexEntryPlaceInBucket::new(ix_index);
                 if index.try_lock(ix_index) {
+                    *entries_created_on_disk += 1;
                     // found free element and occupied it
                     // These fields will be overwritten after allocation by callers.
                     // Since this part of the mmapped file could have previously been used by someone else, there can be garbage here.
@@ -566,7 +621,7 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
                 count += 1;
                 // grow relative to the current capacity
                 let new_capacity = (current_capacity * 110 / 100).max(anticipated_size);
-                let (mut index, _file_name) = BucketStorage::new_with_capacity(
+                let (mut index, file_name) = BucketStorage::new_with_capacity(
                     Arc::clone(&self.drives),
                     1,
                     std::mem::size_of::<IndexEntry<T>>() as u64,
@@ -592,20 +647,14 @@ impl<'b, T: Clone + Copy + PartialEq + std::fmt::Debug + 'static> Bucket<T> {
                         let new_elem: &mut IndexEntry<T> = index.get_mut(new_ix);
                         *new_elem = *elem;
                         index.copying_entry(new_ix, &self.index, ix);
-                        /*
-                        let dbg_elem: IndexEntry = *new_elem;
-                        assert_eq!(
-                            Self::bucket_find_index_entry(&index, &elem.key, random).unwrap(),
-                            (&dbg_elem, new_ix)
-                        );
-                        */
                     }
                 }
                 if valid {
                     self.stats.index.update_max_size(index.capacity());
                     let mut items = self.reallocated.items.lock().unwrap();
                     items.index = Some(index);
                     self.reallocated.add_reallocation();
+                    self.restartable_bucket.set_file(file_name, self.random);
                     break;
                 }
             }
@@ -816,12 +865,14 @@ mod tests {
 
                 let mut index = create_test_index(None);
 
+                let mut entries_created = 0;
                 let mut duplicates = Vec::default();
                 assert!(Bucket::<u64>::batch_insert_non_duplicates_internal(
                     &mut index,
                     &Vec::default(),
                     &raw,
                     &mut hashed,
+                    &mut entries_created,
                     &mut duplicates,
                 )
                 .is_ok());
@@ -865,11 +916,13 @@ mod tests {
                 let mut index = create_test_index(None);
 
                 let mut duplicates = Vec::default();
+                let mut entries_created = 0;
                 assert!(Bucket::<u64>::batch_insert_non_duplicates_internal(
                     &mut index,
                     &Vec::default(),
                     &raw,
                     &mut hashed,
+                    &mut entries_created,
                     &mut duplicates,
                 )
                 .is_ok());
@@ -913,11 +966,13 @@ mod tests {
                     let mut index = create_test_index(Some(max_search as u8));
 
                     let mut duplicates = Vec::default();
+                    let mut entries_created = 0;
                     let result = Bucket::<u64>::batch_insert_non_duplicates_internal(
                         &mut index,
                         &Vec::default(),
                         &raw,
                         &mut hashed,
+                        &mut entries_created,
                         &mut duplicates,
                     );
 
@@ -1071,7 +1126,13 @@ mod tests {
         let paths: Vec<PathBuf> = vec![tmpdir.path().to_path_buf()];
         assert!(!paths.is_empty());
         let max_search = 2;
-        let mut bucket = Bucket::new(Arc::new(paths), max_search, Arc::default(), Arc::default());
+        let mut bucket = Bucket::new(
+            Arc::new(paths),
+            max_search,
+            Arc::default(),
+            Arc::default(),
+            RestartableBucket::default(),
+        );
 
         let key = Pubkey::new_unique();
         assert_eq!(bucket.read_value(&key), None);

bucket_map/src/bucket_api.rs

@@ -1,7 +1,7 @@
 use {
     crate::{
         bucket::Bucket, bucket_item::BucketItem, bucket_map::BucketMapError,
-        bucket_stats::BucketMapStats, MaxSearch, RefCount,
+        bucket_stats::BucketMapStats, restart::RestartableBucket, MaxSearch, RefCount,
     },
     solana_sdk::pubkey::Pubkey,
     std::{
@@ -23,20 +23,27 @@ pub struct BucketApi<T: Clone + Copy + PartialEq + 'static> {
 
     bucket: LockedBucket<T>,
     count: Arc<AtomicU64>,
+
+    /// keeps track of which index file this bucket is currently using
+    /// or at startup, which bucket file this bucket should initially use
+    #[allow(dead_code)]
+    restartable_bucket: RestartableBucket,
 }
 
 impl<T: Clone + Copy + PartialEq + std::fmt::Debug> BucketApi<T> {
-    pub fn new(
+    pub(crate) fn new(
         drives: Arc<Vec<PathBuf>>,
         max_search: MaxSearch,
         stats: Arc<BucketMapStats>,
+        restartable_bucket: RestartableBucket,
     ) -> Self {
         Self {
             drives,
             max_search,
             stats,
             bucket: RwLock::default(),
             count: Arc::default(),
+            restartable_bucket,
         }
     }
 
@@ -90,6 +97,7 @@ impl<T: Clone + Copy + PartialEq + std::fmt::Debug> BucketApi<T> {
                 self.max_search,
                 Arc::clone(&self.stats),
                 Arc::clone(&self.count),
+                self.restartable_bucket.clone(),
             ));
         }
     }