Implement the BucketQueueView (#2449)

* Implement the BucketQueueView which takes the size as a template parameter.
* Put random tests and unit tests.
* Put benchmark and in particular BucketQueueView vs QueueView.
Co-authored-by: Andreas Fackler <[email protected]>

Author: MathieuDutSik

linera-views/Cargo.toml

@@ -84,3 +84,7 @@ harness = false
 [[bench]]
 name = "stores"
 harness = false
+
+[[bench]]
+name = "queue_view"
+harness = false

linera-views/benches/queue_view.rs

@@ -0,0 +1,226 @@
+// Copyright (c) Zefchain Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::{
+    fmt::Debug,
+    time::{Duration, Instant},
+};
+
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+#[cfg(with_dynamodb)]
+use linera_views::dynamo_db::create_dynamo_db_test_store;
+#[cfg(with_rocksdb)]
+use linera_views::rocks_db::create_rocks_db_test_store;
+#[cfg(with_scylladb)]
+use linera_views::scylla_db::create_scylla_db_test_store;
+use linera_views::{
+    backends::memory::create_test_memory_store,
+    bucket_queue_view::BucketQueueView,
+    common::KeyValueStore,
+    context::ViewContext,
+    queue_view::QueueView,
+    test_utils::{make_deterministic_rng, DeterministicRng},
+    views::{CryptoHashRootView, RootView, View},
+};
+use rand::Rng;
+use tokio::runtime::Runtime;
+
+/// The number of operations
+const N_OPERATIONS: usize = 1000;
+
+enum Operations {
+    Save,
+    DeleteFront,
+    PushBack(u8),
+}
+
+fn generate_test_case(n_operation: usize, rng: &mut DeterministicRng) -> Vec<Operations> {
+    let mut operations = Vec::new();
+    let mut total_length = 0;
+    for _ in 0..n_operation {
+        let choice = rng.gen_range(0..10);
+        if choice == 0 {
+            operations.push(Operations::Save);
+        } else if choice < 3 && total_length > 0 {
+            operations.push(Operations::DeleteFront);
+            total_length -= 1;
+        } else {
+            let val = rng.gen::<u8>();
+            operations.push(Operations::PushBack(val));
+            total_length += 1;
+        }
+    }
+    operations
+}
+
+#[derive(CryptoHashRootView)]
+pub struct QueueStateView<C> {
+    pub queue: QueueView<C, u8>,
+}
+
+pub async fn performance_queue_view<S: KeyValueStore + Clone + Sync + 'static>(
+    store: S,
+    iterations: u64,
+) -> Duration
+where
+    S::Error: Debug + Send + Sync + 'static,
+{
+    let context = ViewContext::<(), S>::create_root_context(store, ())
+        .await
+        .unwrap();
+    let mut total_time = Duration::ZERO;
+    let mut rng = make_deterministic_rng();
+    for _ in 0..iterations {
+        let operations = generate_test_case(N_OPERATIONS, &mut rng);
+        let mut view = QueueStateView::load(context.clone()).await.unwrap();
+        let measurement = Instant::now();
+        for operation in operations {
+            match operation {
+                Operations::Save => {
+                    view.save().await.unwrap();
+                }
+                Operations::DeleteFront => {
+                    view.queue.delete_front();
+                }
+                Operations::PushBack(val) => {
+                    view.queue.push_back(val);
+                }
+            }
+            black_box(view.queue.front().await.unwrap());
+        }
+        view.clear();
+        view.save().await.unwrap();
+        total_time += measurement.elapsed();
+    }
+
+    total_time
+}
+
+fn bench_queue_view(criterion: &mut Criterion) {
+    criterion.bench_function("memory_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_test_memory_store();
+                performance_queue_view(store, iterations).await
+            })
+    });
+
+    #[cfg(with_rocksdb)]
+    criterion.bench_function("rocksdb_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_rocks_db_test_store().await;
+                performance_queue_view(store, iterations).await
+            })
+    });
+
+    #[cfg(with_dynamodb)]
+    criterion.bench_function("dynamodb_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_dynamo_db_test_store().await;
+                performance_queue_view(store, iterations).await
+            })
+    });
+
+    #[cfg(with_scylladb)]
+    criterion.bench_function("scylladb_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_scylla_db_test_store().await;
+                performance_queue_view(store, iterations).await
+            })
+    });
+}
+
+#[derive(CryptoHashRootView)]
+pub struct BucketQueueStateView<C> {
+    pub queue: BucketQueueView<C, u8, 100>,
+}
+
+pub async fn performance_bucket_queue_view<S: KeyValueStore + Clone + Sync + 'static>(
+    store: S,
+    iterations: u64,
+) -> Duration
+where
+    S::Error: Debug + Send + Sync + 'static,
+{
+    let context = ViewContext::<(), S>::create_root_context(store, ())
+        .await
+        .unwrap();
+    let mut total_time = Duration::ZERO;
+    let mut rng = make_deterministic_rng();
+    for _ in 0..iterations {
+        let operations = generate_test_case(N_OPERATIONS, &mut rng);
+        let mut view = BucketQueueStateView::load(context.clone()).await.unwrap();
+        //
+        let measurement = Instant::now();
+        for operation in operations {
+            match operation {
+                Operations::Save => {
+                    view.save().await.unwrap();
+                }
+                Operations::DeleteFront => {
+                    view.queue.delete_front().await.unwrap();
+                }
+                Operations::PushBack(val) => {
+                    view.queue.push_back(val);
+                }
+            }
+            black_box(view.queue.front());
+        }
+        view.clear();
+        view.save().await.unwrap();
+        total_time += measurement.elapsed();
+    }
+
+    total_time
+}
+
+fn bench_bucket_queue_view(criterion: &mut Criterion) {
+    criterion.bench_function("memory_bucket_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_test_memory_store();
+                performance_bucket_queue_view(store, iterations).await
+            })
+    });
+
+    #[cfg(with_rocksdb)]
+    criterion.bench_function("rocksdb_bucket_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_rocks_db_test_store().await;
+                performance_bucket_queue_view(store, iterations).await
+            })
+    });
+
+    #[cfg(with_dynamodb)]
+    criterion.bench_function("dynamodb_bucket_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_dynamo_db_test_store().await;
+                performance_bucket_queue_view(store, iterations).await
+            })
+    });
+
+    #[cfg(with_scylladb)]
+    criterion.bench_function("scylladb_bucket_queue_view", |bencher| {
+        bencher
+            .to_async(Runtime::new().expect("Failed to create Tokio runtime"))
+            .iter_custom(|iterations| async move {
+                let store = create_scylla_db_test_store().await;
+                performance_bucket_queue_view(store, iterations).await
+            })
+    });
+}
+
+criterion_group!(benches, bench_queue_view, bench_bucket_queue_view);
+criterion_main!(benches);

linera-views/src/lib.rs

@@ -97,8 +97,8 @@ pub use backends::rocks_db;
 pub use backends::scylla_db;
 pub use backends::{journaling, lru_caching, memory, value_splitting};
 pub use views::{
-    collection_view, hashable_wrapper, key_value_store_view, log_view, map_view, queue_view,
-    reentrant_collection_view, register_view, set_view,
+    bucket_queue_view, collection_view, hashable_wrapper, key_value_store_view, log_view, map_view,
+    queue_view, reentrant_collection_view, register_view, set_view,
 };
 /// Re-exports used by the derive macros of this library.
 #[doc(hidden)]