Benchmarking PreferDistinct strategy changes

Cargo.toml

@@ -1,15 +1,15 @@
 [workspace]
 
-# Resolver 3 is default in the 2024 Edition, but the workspace doesn't know that
-resolver = "3"
+# Resolver 2 is default in the 2021 Edition, but the workspace doesn't know that
+resolver = "2"
 
 members = [
     "opsqueue/",
     "libs/opsqueue_python",
 ]
 
 [workspace.package]
-version = "0.36.0"
+version = "0.35.5"
 
 [workspace.lints.clippy]
 cargo =  { level = "warn", priority = -1 }
@@ -27,10 +27,6 @@ restriction = { level = "allow", priority = -1 }
 style = { level = "warn", priority = -1 }
 suspicious = { level = "warn", priority = -1 }
 
-# Prevent duplicate versions of crates from being used in the workspace. If an easy fix is not
-# possible, you can add the crate to the `allowed-duplicate-crates` list in `clippy.toml`.
-multiple_crate_versions = "deny"
-
 [profile.release]
 # Full fat LTO makes a big difference in performance, at the cost of a large
 # compile time hit. You should only use release builds when performance matters.

default.nix

@@ -63,6 +63,8 @@ let
         # For the shell, libpython needs to be in the search path.
         pythonEnv
       ];
+      # Fixes: libstdc++.so.6: cannot open shared object file: No such file or directory
+      LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath [ pkgs.stdenv.cc.cc.lib ];
     };
   };
 in

justfile

@@ -53,7 +53,7 @@ test-integration *TEST_ARGS: build-bin build-python
   cd libs/opsqueue_python
   source "./.setup_local_venv.sh"
 
-  timeout {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}}
+  pytest --timeout {{pytest_timeout_seconds}} --color=yes {{TEST_ARGS}}
 
 # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest
 [group('nix')]
@@ -68,7 +68,19 @@ nix-test-integration *TEST_ARGS: nix-build
   export OPSQUEUE_BIN="${nix_build_bin_dir}/bin/opsqueue"
   export RUST_LOG="opsqueue=debug"
 
-  timeout {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}}
+  pytest --timeout {{pytest_timeout_seconds}} --color=yes {{TEST_ARGS}}
+
+# Benchmark query cost and render a graph.
+[group('bench')]
+bench-chunks-select:
+  #!/usr/bin/env bash
+  set -euo pipefail
+  cargo bench --bench chunks_select
+
+  cd libs/opsqueue_python
+  source "./.setup_local_venv.sh"
+  cd -
+  python opsqueue/benches/plot_chunks_select.py
 
 # Run all linters, fast and slow
 [group('lint')]

libs/opsqueue_python/pyproject.toml

@@ -41,9 +41,10 @@ include = ["**/py.typed"]
 # because otherwise `pytest` won't be able to see the locally build python package!
 [project.optional-dependencies]
 test = [
+    "matplotlib==3.11.1",
+    "multiprocess==0.70.19",
     "pytest==9.0.3",
     "pytest-xdist==3.8.0",
-    "multiprocess==0.70.19",
     "pytest-timeout==2.4.0",
 ]
 

libs/opsqueue_python/tests/test_roundtrip.py

@@ -667,3 +667,32 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
             )
         assert exc.type is TooManyMatchingSubmissionsError
         assert exc.value.max_submissions == max_
+
+
+def test_prefer_distinct_strategy_fairness(opsqueue: OpsqueueProcess) -> None:
+    """Test the PreferDistinct strategy fairly interleaves chunks from different
+    submissions based on their strategic metadata.
+
+    """
+    url = "file:///tmp/opsqueue/test_prefer_distinct_fairness"
+    producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
+    consumer_client = ConsumerClient(f"localhost:{opsqueue.port}", url)
+    company_ids = [1, 2, 3]
+    chunks_per_company = 4
+    company_id_per_submission = {}
+    for company_id in company_ids:
+        sub_id = producer_client.insert_submission(
+            [company_id] * chunks_per_company,
+            chunk_size=1,
+            strategic_metadata={"company_id": company_id},
+        )
+        company_id_per_submission[sub_id] = company_id
+    strategy = strategy_from_description(("PreferDistinct", "company_id", "Oldest"))
+    reserved_company_order = []
+    # Fetch 1 chunk at a time. Because we don't complete chunks, opsqueue's
+    # metastate tracks them as reserved, increasing the busy count for that
+    # company.
+    for _ in range(len(company_ids) * chunks_per_company):
+        [chunk] = consumer_client.reserve_chunks(strategy=strategy)
+        reserved_company_order.append(company_id_per_submission[chunk.submission_id])
+    assert reserved_company_order == [1, 2, 3] * chunks_per_company

opsqueue/Cargo.toml

@@ -33,6 +33,7 @@ ux = "0.1.6"
 anyhow = "1.0.102"
 # Database:
 sqlx = { version = "0.9.0", features = ["sqlite", "runtime-tokio", "chrono"], optional = true }
+libsqlite3-sys = { version = "0.30.1", optional = true }
 # Serialization:
 serde = { version = "1.0.203", features = ["derive"] }
 serde_json = "1.0.149"
@@ -83,13 +84,14 @@ sqlformat = "0.5.0"
 workspace = true
 
 [dev-dependencies]
+assert_matches = { version = "1.5.0" }
 criterion = {version = "0.8", features = ["async_tokio"]}
 insta = { version = "1.47.2" }
-assert_matches = { version = "1.5.0" }
 
-# [[bench]]
-# name = "chunks_select"
-# harness = false
+[[bench]]
+name = "chunks_select"
+harness = false
+required-features = ["server-logic"]
 
 # [[bench]]
 # name = "submissions_insert"
@@ -114,4 +116,4 @@ client-logic = [
     "dep:tokio-tungstenite",
     "dep:arc-swap",
 ]
-default = ["server-logic"]
+default = ["server-logic", "dep:libsqlite3-sys"]