From 861230edc9bd47e7d19c08d4c69b1db118f332bd Mon Sep 17 00:00:00 2001 From: "Ivan A. Kudriavtsev" Date: Tue, 26 Aug 2025 15:56:42 +0700 Subject: [PATCH] Work in progress, upgraded PyO3 and RocksDb parameters for better performance. --- .devcontainer/Dockerfile | 1 + .devcontainer/devcontainer.json | 20 ++++++++++++++ .github/workflows/rocksq.yml | 15 +++++------ Cargo.toml | 18 ++++++++++++- docker/Dockerfile.manylinux_2_28 | 10 +++++++ docker/Dockerfile.manylinux_2_28_ARM64 | 19 ------------- docker/Dockerfile.manylinux_2_28_X64 | 19 ------------- queue_py/Cargo.toml | 16 +++++------ queue_py/src/blocking.rs | 4 +-- queue_py/src/lib.rs | 2 +- queue_py/src/nonblocking.rs | 4 +-- queue_rs/Cargo.toml | 26 +++++++++++------- queue_rs/benches/throughput_mpmc_sync.rs | 34 +++++++++++++----------- queue_rs/benches/throughput_sync.rs | 34 +++++++++++++----------- queue_rs/src/lib.rs | 1 + queue_rs/src/mpmc.rs | 1 + 16 files changed, 120 insertions(+), 104 deletions(-) create mode 100644 .devcontainer/Dockerfile create mode 100644 .devcontainer/devcontainer.json create mode 100644 docker/Dockerfile.manylinux_2_28 delete mode 100644 docker/Dockerfile.manylinux_2_28_ARM64 delete mode 100644 docker/Dockerfile.manylinux_2_28_X64 diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile new file mode 100644 index 0000000..c655a57 --- /dev/null +++ b/.devcontainer/Dockerfile @@ -0,0 +1 @@ +FROM ghcr.io/insight-platform/manylinux_2_28:v1.1.0 AS builder diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json new file mode 100644 index 0000000..e0bfdba --- /dev/null +++ b/.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!'" +} diff --git a/.github/workflows/rocksq.yml b/.github/workflows/rocksq.yml index f2b4e46..8c461ce 100644 --- a/.github/workflows/rocksq.yml +++ b/.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 diff --git a/Cargo.toml b/Cargo.toml index b411b8e..9aa2c90 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "2" [workspace.package] -version = "0.3.0" +version = "0.4.0" edition = "2021" authors = ["Ivan Kudriavtsev "] 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"] } + diff --git a/docker/Dockerfile.manylinux_2_28 b/docker/Dockerfile.manylinux_2_28 new file mode 100644 index 0000000..32ba448 --- /dev/null +++ b/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 diff --git a/docker/Dockerfile.manylinux_2_28_ARM64 b/docker/Dockerfile.manylinux_2_28_ARM64 deleted file mode 100644 index b33bb26..0000000 --- a/docker/Dockerfile.manylinux_2_28_ARM64 +++ /dev/null @@ -1,19 +0,0 @@ -FROM ghcr.io/insight-platform/manylinux_2_28_arm64:v0.0.4 as chef - -FROM chef as planner - -WORKDIR /opt -COPY . . -RUN cargo chef prepare --recipe-path recipe.json - -FROM chef AS builder -WORKDIR /opt -COPY --from=planner /opt/recipe.json recipe.json -RUN cargo chef cook --release --recipe-path recipe.json -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 diff --git a/docker/Dockerfile.manylinux_2_28_X64 b/docker/Dockerfile.manylinux_2_28_X64 deleted file mode 100644 index 3ca3d3c..0000000 --- a/docker/Dockerfile.manylinux_2_28_X64 +++ /dev/null @@ -1,19 +0,0 @@ -FROM ghcr.io/insight-platform/manylinux_2_28_x64:v0.0.4 as chef - -FROM chef as planner - -WORKDIR /opt -COPY . . -RUN cargo chef prepare --recipe-path recipe.json - -FROM chef AS builder -WORKDIR /opt -COPY --from=planner /opt/recipe.json recipe.json -RUN cargo chef cook --release --recipe-path recipe.json -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 diff --git a/queue_py/Cargo.toml b/queue_py/Cargo.toml index 262b9f1..1a8ef74 100644 --- a/queue_py/Cargo.toml +++ b/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 diff --git a/queue_py/src/blocking.rs b/queue_py/src/blocking.rs index 3564c22..6979ef8 100644 --- a/queue_py/src/blocking.rs +++ b/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(()) }) diff --git a/queue_py/src/lib.rs b/queue_py/src/lib.rs index 7088ecf..762d0ce 100644 --- a/queue_py/src/lib.rs +++ b/queue_py/src/lib.rs @@ -92,7 +92,7 @@ fn rocksq(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; - 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::()?; diff --git a/queue_py/src/nonblocking.rs b/queue_py/src/nonblocking.rs index 1591a3a..1923621 100644 --- a/queue_py/src/nonblocking.rs +++ b/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(()) }) diff --git a/queue_rs/Cargo.toml b/queue_rs/Cargo.toml index e99bc93..339ec14 100644 --- a/queue_rs/Cargo.toml +++ b/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 diff --git a/queue_rs/benches/throughput_mpmc_sync.rs b/queue_rs/benches/throughput_mpmc_sync.rs index 0247a0b..c91f93e 100644 --- a/queue_rs/benches/throughput_mpmc_sync.rs +++ b/queue_rs/benches/throughput_mpmc_sync.rs @@ -1,21 +1,18 @@ -#![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 { @@ -23,16 +20,17 @@ fn rw_mixed(b: &mut Bencher) { 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); diff --git a/queue_rs/benches/throughput_sync.rs b/queue_rs/benches/throughput_sync.rs index 277b243..bd2348b 100644 --- a/queue_rs/benches/throughput_sync.rs +++ b/queue_rs/benches/throughput_sync.rs @@ -1,19 +1,16 @@ -#![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 { @@ -21,16 +18,17 @@ fn rw_mixed(b: &mut Bencher) { 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); diff --git a/queue_rs/src/lib.rs b/queue_rs/src/lib.rs index be67694..ba2bd9e 100644 --- a/queue_rs/src/lib.rs +++ b/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)?; diff --git a/queue_rs/src/mpmc.rs b/queue_rs/src/mpmc.rs index f34709e..20a869b 100644 --- a/queue_rs/src/mpmc.rs +++ b/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();