Work in progress, upgraded PyO3 and RocksDb parameters for better per…

.devcontainer/Dockerfile

@@ -0,0 +1 @@
+FROM ghcr.io/insight-platform/manylinux_2_28:v1.1.0 AS builder

.devcontainer/devcontainer.json

@@ -0,0 +1,20 @@
+{
+    "name": "My Dev Container",
+    "image": "ghcr.io/insight-platform/manylinux_2_28:v1.1.0",
+    "features": {},
+    "customizations": {
+        "vscode": {
+            "settings": {},
+            "extensions": [
+                "ms-python.python",
+                "ms-azuretools.vscode-docker",
+                "rust-lang.rust-analyzer",
+                "tamasfe.even-better-toml"
+            ]
+        }
+    },
+    "forwardPorts": [
+        3000
+    ],
+    "postCreateCommand": "echo 'Dev container ready!'"
+}

.github/workflows/rocksq.yml

@@ -27,19 +27,18 @@ jobs:
     strategy:
       matrix:
         include:
-          - docker_file: docker/Dockerfile.manylinux_2_28_ARM64
-            maturin_action_docker_image: ghcr.io/insight-platform/manylinux_2_28_arm64:v0.0.5
+          - docker_file: docker/Dockerfile.manylinux_2_28
+            maturin_action_docker_image: ghcr.io/insight-platform/manylinux_2_28:v1.1.0
             name: manylinux-arm
             arch: linux/arm64
-            runner: ARM64
-          - docker_file: docker/Dockerfile.manylinux_2_28_X64
-            maturin_action_docker_image: ghcr.io/insight-platform/manylinux_2_28_x64:v0.0.5
+            group: native-builders-arm64
+          - docker_file: docker/Dockerfile.manylinux_2_28
+            maturin_action_docker_image: ghcr.io/insight-platform/manylinux_2_28:v1.1.0
             name: manylinux-x86
             arch: linux/amd64
-            runner: X64
+            group: native-builders-x64
     runs-on:
-      - self-hosted
-      - ${{ matrix.runner }}
+      group: ${{ matrix.group }}
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5

Cargo.toml

@@ -7,7 +7,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.3.0"
+version = "0.4.0"
 edition = "2021"
 authors = ["Ivan Kudriavtsev <ivan.a.kudryavtsev@gmail.com>"]
 description = "Rust persistent queue based on RocksDB with Python bindings"
@@ -25,3 +25,19 @@ opt-level = 3
 [profile.bench]
 opt-level = 3
 
+[workspace.dependencies]
+
+anyhow = "1"
+parking_lot = "0.12"
+crossbeam-channel = "0.5"
+bincode = "2.0.1"
+chrono = "0.4"
+rocksdb = { version = "0.24", default-features = false }
+tempfile = "3.21"
+queue_rs = { path = "queue_rs" }
+pyo3 = "0.25"
+serial_test = "3"
+pyo3-build-config = "0.25"
+cbindgen = "0.29"
+criterion = { version = "0.5", features = ["html_reports"] }
+

docker/Dockerfile.manylinux_2_28

@@ -0,0 +1,10 @@
+FROM ghcr.io/insight-platform/manylinux_2_28:v1.1.0 AS builder
+
+WORKDIR /opt
+COPY . .
+ARG PYTHON_INTERPRETER
+RUN bash /opt/docker/build-manylinux.sh
+RUN rm -rf target
+
+FROM alpine:3.18 AS dist
+COPY --from=builder /opt/dist /opt/dist

queue_py/Cargo.toml

@@ -16,17 +16,16 @@ rust-version.workspace = true
 crate-type = ["cdylib", "lib"]
 
 [dependencies]
-queue_rs = { path = "../queue_rs" }
-
-[dependencies.pyo3]
-version = "0.22.2"
+queue_rs = { workspace = true }
+pyo3 = { workspace = true }
+rocksdb = { workspace = true }
 
 [dev-dependencies]
-serial_test = "3"
+serial_test = { workspace = true }
 
 [build-dependencies]
-pyo3-build-config = "0.22.2"
-cbindgen = "0.27.0"
+pyo3-build-config = { workspace = true }
+cbindgen = { workspace = true }
 
 [package.metadata.maturin]
 python-source = "python"
@@ -35,6 +34,3 @@ python-source = "python"
 extension-module = ["pyo3/extension-module"]
 default = ["extension-module"]
 
-[dependencies.rocksdb]
-version = "0.22"
-default-features = false

queue_py/src/blocking.rs

@@ -111,7 +111,7 @@ impl PersistentQueueWithCapacity {
                 results
                     .into_iter()
                     .map(|r| {
-                        PyBytes::new_bound_with(py, r.len(), |b: &mut [u8]| {
+                        PyBytes::new_with(py, r.len(), |b: &mut [u8]| {
                             b.copy_from_slice(&r);
                             Ok(())
                         })
@@ -306,7 +306,7 @@ impl MpmcQueue {
                     .0
                     .into_iter()
                     .map(|r| {
-                        PyBytes::new_bound_with(py, r.len(), |b: &mut [u8]| {
+                        PyBytes::new_with(py, r.len(), |b: &mut [u8]| {
                             b.copy_from_slice(&r);
                             Ok(())
                         })

queue_py/src/lib.rs

@@ -92,7 +92,7 @@ fn rocksq(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
 
     m.add_class::<StartPosition>()?;
 
-    let sys = PyModule::import_bound(py, "sys")?;
+    let sys = PyModule::import(py, "sys")?;
     let attr_modules = sys.getattr("modules")?;
     let sys_modules: &Bound<'_, PyDict> = attr_modules.downcast::<PyDict>()?;
 

queue_py/src/nonblocking.rs

@@ -38,7 +38,7 @@ impl ResponseVariant {
                         results
                             .iter()
                             .map(|r| {
-                                PyBytes::new_bound_with(py, r.len(), |b: &mut [u8]| {
+                                PyBytes::new_with(py, r.len(), |b: &mut [u8]| {
                                     b.copy_from_slice(r);
                                     Ok(())
                                 })
@@ -382,7 +382,7 @@ impl MpmcResponseVariant {
                             .0
                             .iter()
                             .map(|r| {
-                                PyBytes::new_bound_with(py, r.len(), |b: &mut [u8]| {
+                                PyBytes::new_with(py, r.len(), |b: &mut [u8]| {
                                     b.copy_from_slice(r);
                                     Ok(())
                                 })

queue_rs/Cargo.toml

@@ -18,15 +18,21 @@ crate-type = ["cdylib", "lib"]
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-anyhow = "1"
-parking_lot = "0.12"
-crossbeam-channel = "0.5.13"
-bincode = "2.0.0-rc.3"
-chrono = "0.4.38"
-
-[dependencies.rocksdb]
-version = "0.22"
-default-features = false
+anyhow = { workspace = true }
+parking_lot = { workspace = true }
+crossbeam-channel = { workspace = true }
+bincode = { workspace = true }
+chrono = { workspace = true }
+rocksdb = { workspace = true }
 
 [dev-dependencies]
-tempfile = "3.12.0"
+tempfile = { workspace = true }
+criterion = { workspace = true }
+
+[[bench]]
+name = "throughput_mpmc_sync"
+harness = false
+
+[[bench]]
+name = "throughput_sync"
+harness = false

queue_rs/benches/throughput_mpmc_sync.rs

@@ -1,38 +1,36 @@
-#![feature(test)]
-
-extern crate test;
-
+use criterion::{criterion_group, criterion_main, Criterion};
 use queue_rs::blocking::MpmcQueue;
 use queue_rs::mpmc::StartPosition;
 use std::time::Duration;
-use test::Bencher;
 
-const COUNT: usize = 10;
+const COUNT: usize = 100;
 const LABEL: &str = "label";
+const BLOCK_SIZE: usize = 64 * 1024;
 
-#[bench]
-fn rw_mixed(b: &mut Bencher) {
-    let block = vec![0u8; 256 * 1024];
+fn rw_mixed(c: &mut Criterion) {
+    let block = vec![0u8; BLOCK_SIZE];
     let path = "/tmp/test_mpmc_b1".to_string();
     _ = MpmcQueue::remove_db(&path);
-    {
+
+    c.bench_function("mpmc_rw_mixed", |b| {
         let db = MpmcQueue::new(&path, Duration::from_secs(60)).unwrap();
         b.iter(|| {
             for _ in 0..COUNT {
                 db.add(&[&block]).unwrap();
                 db.next(1, LABEL, StartPosition::Oldest).unwrap();
             }
         });
-    }
+    });
+
     MpmcQueue::remove_db(&path).unwrap();
 }
 
-#[bench]
-fn write_read(b: &mut Bencher) {
-    let block = vec![0u8; 256 * 1024];
+fn write_read(c: &mut Criterion) {
+    let block = vec![0u8; BLOCK_SIZE];
     let path = "/tmp/test_mpmc_b2".to_string();
     _ = MpmcQueue::remove_db(&path);
-    {
+
+    c.bench_function("mpmc_write_read", |b| {
         let db = MpmcQueue::new(&path, Duration::from_secs(60)).unwrap();
         b.iter(|| {
             for _ in 0..COUNT {
@@ -42,6 +40,10 @@ fn write_read(b: &mut Bencher) {
                 db.next(1, LABEL, StartPosition::Oldest).unwrap();
             }
         });
-    }
+    });
+
     MpmcQueue::remove_db(&path).unwrap();
 }
+
+criterion_group!(benches, rw_mixed, write_read);
+criterion_main!(benches);

queue_rs/benches/throughput_sync.rs

@@ -1,36 +1,34 @@
-#![feature(test)]
-
-extern crate test;
-
+use criterion::{criterion_group, criterion_main, Criterion};
 use queue_rs::blocking::PersistentQueueWithCapacity;
 use rocksdb::Options;
-use test::Bencher;
 
-const COUNT: usize = 10;
+const COUNT: usize = 100;
+const BLOCK_SIZE: usize = 64 * 1024;
 
-#[bench]
-fn rw_mixed(b: &mut Bencher) {
-    let block = vec![0u8; 256 * 1024];
+fn rw_mixed(c: &mut Criterion) {
+    let block = vec![0u8; BLOCK_SIZE];
     let path = "/tmp/test_b1".to_string();
     _ = PersistentQueueWithCapacity::remove_db(&path);
-    {
+
+    c.bench_function("sync_rw_mixed", |b| {
         let db = PersistentQueueWithCapacity::new(&path, COUNT, Options::default()).unwrap();
         b.iter(|| {
             for _ in 0..COUNT {
                 db.push(&[&block]).unwrap();
                 db.pop(1).unwrap();
             }
         });
-    }
+    });
+
     PersistentQueueWithCapacity::remove_db(&path).unwrap();
 }
 
-#[bench]
-fn write_read(b: &mut Bencher) {
-    let block = vec![0u8; 256 * 1024];
+fn write_read(c: &mut Criterion) {
+    let block = vec![0u8; BLOCK_SIZE];
     let path = "/tmp/test_b2".to_string();
     _ = PersistentQueueWithCapacity::remove_db(&path);
-    {
+
+    c.bench_function("sync_write_read", |b| {
         let db = PersistentQueueWithCapacity::new(&path, COUNT, Options::default()).unwrap();
         b.iter(|| {
             for _ in 0..COUNT {
@@ -40,6 +38,10 @@ fn write_read(b: &mut Bencher) {
                 db.pop(1).unwrap();
             }
         });
-    }
+    });
+
     PersistentQueueWithCapacity::remove_db(&path).unwrap();
 }
+
+criterion_group!(benches, rw_mixed, write_read);
+criterion_main!(benches);

queue_rs/src/lib.rs

@@ -48,6 +48,7 @@ impl PersistentQueueWithCapacity {
         }
         db_opts.create_if_missing(true);
         db_opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(U64_BYTE_LEN));
+        db_opts.set_compaction_style(rocksdb::DBCompactionStyle::Fifo);
 
         let db = DB::open(&db_opts, path)?;
 

queue_rs/src/mpmc.rs

@@ -60,6 +60,7 @@ impl MpmcQueue {
         let mut cf_opts = Options::default();
         cf_opts.create_if_missing(true);
         cf_opts.set_prefix_extractor(SliceTransform::create_fixed_prefix(crate::U64_BYTE_LEN));
+        cf_opts.set_compaction_style(rocksdb::DBCompactionStyle::Fifo);
         let data_cf = ColumnFamilyDescriptor::new(DATA_CF, cf_opts);
 
         let mut cf_opts = Options::default();