Reduce thread contention

.github/workflows/ci.yml

@@ -14,27 +14,28 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: EmbarkStudios/cargo-deny-action@v2
+        with:
+          rust-version: "1.85.0"
       - uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
           components: rustfmt, clippy
       - run: |
-          cargo clippy --all-targets -F affinity,batching,retry \
+          cargo clippy --all-targets -F affinity,local-batch,retry \
             -- -D warnings $(cat .lints | cut -f1 -d"#" | tr '\n' ' ')
       - run: cargo fmt -- --check
-      - run: cargo doc -F affinity,batching,retry
-      - run: cargo test -F affinity,batching,retry --doc
+      - run: RUSTDOCFLAGS="-D warnings" cargo doc -F affinity,local-batch,retry
+      - run: cargo test -F affinity,local-batch,retry --doc
 
   coverage:
     name: Code coverage
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - uses: EmbarkStudios/cargo-deny-action@v2
       - uses: actions-rust-lang/setup-rust-toolchain@v1
       - name: Install cargo-llvm-cov
         uses: taiki-e/install-action@cargo-llvm-cov
       - name: Generate code coverage
-        run: cargo llvm-cov --lcov --output-path lcov.info -F affinity,batching,retry
+        run: cargo llvm-cov --lcov --output-path lcov.info -F affinity,local-batch,retry
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v3
         with:
@@ -58,7 +59,6 @@ jobs:
           - loole
     steps:
       - uses: actions/checkout@v4
-      - uses: EmbarkStudios/cargo-deny-action@v2
       - uses: actions-rust-lang/setup-rust-toolchain@v1
       - uses: actions-rs/cargo@v1
         with:

CHANGELOG.md

@@ -2,8 +2,36 @@
 
 ## 0.3.0
 
+The general theme of this release is performance improvement by eliminating thread contention due to unnecessary locking of shared state. This required making some breaking changes to the API.
+
+* **Breaking**
+  * `beekeeper::hive::Hive` type signature has changed
+    * Removed the `W: Worker` parameter as it is redundant (can be obtained from `Q::Kind`)
+    * Added `T: TaskQueues`to specify the `TaskQueues` implementation
+  * The `Builder` interface has been re-written to enable maximum flexibility.
+    * `Builder` is now a trait that must be in scope.
+    * `ChannelBuilder` implements the previous builder functionality.
+    * `OpenBuilder` has no type parameters and can be specialized to create a `Hive` with any combination of `Queen` and `TaskQueues`.
+    * `BeeBuilder` and `FullBuilder` are intermediate types that generally should not be instantiated directly.
+  * `beekeeper::bee::Queen::create` now takes `&self` rather than `&mut self`. There is a new type, `beekeeper::bee::QueenMut`, with a `create(&mut self)` method, and needs to be wrapped in a `beekeeper::bee::QueenCell` to implement the `Queen` trait. This enables the `Hive` to create new workers without locking in the case of a `Queen` that does not need mutable state.
+  * `beekeeper::bee::Context` now takes a generic parameter that must be input type of the `Worker`.
+  * `beekeeper::hive::Hive::try_into_husk` now has an `urgent` parameter to indicate whether queued tasks should be abandoned when shutting down the hive (`true`) or if they should be allowed to finish processing (`false`).
+  * The type of `attempt` and `max_retries` has been changed to `u8`. This reduces memory usage and should still allow for the majority of use cases.
+  * The `::of` methods have been removed from stock `Worker`s in favor of implementing `From`.
 * Features
-  * Added the `batching` feature, which enables worker threads to queue up batches of tasks locally, which can alleviate contention between threads in the pool, especially when there are many short-lived tasks.
+  * Added the `TaskQueues` trait, which enables `Hive` to be specialized for different implementations of global (i.e., sending tasks from the `Hive` to worker threads) and local (i.e., worker thread-specific) queues.
+    * `ChannelTaskQueues` implements the existing behavior, using a channel for sending tasks.
+    * `WorkstealingTaskQueues` has been added to implement the workstealing pattern, based on `crossbeam::dequeue`.
+  * Added the `local-batch` feature, which enables worker threads to queue up batches of tasks locally, which can alleviate contention between threads in the pool, especially when there are many short-lived tasks.
+    * When this feature is enabled, tasks can be optionally weighted (by wrapping each input in `crate::hive::Weighted`) to help evenly distribute tasks with variable processing times.
+    * Enabling this feature should be transparent (i.e., not break existing code), and the `Hive`'s task submission methods support both weighted and unweighted inputs (due to the blanket implementation of `From<T> for Weighted<T>`); however, there are some cases where it is now necessary to specify the input type where before it could be elided.
+  * Added the `Context::submit` method, which enables tasks to submit new tasks to the `Hive`.
+* Other
+  * Switched to using thread-local retry queues for the implementation of the `retry` feature, to reduce thread-contention.
+  * Switched to storing `Outcome`s in the hive using a data structure that does not require locking when inserting, which should reduce thread contention when using `*_store` operations.
+  * Switched to using `crossbeam_channel` for the task input channel in `ChannelTaskQueues`. These are multi-produer, multi-consumer channels (mpmc; as opposed to `std::mpsc`, which is single-consumer), which means it is no longer necessary for worker threads to aquire a Mutex lock on the channel receiver when getting tasks.
+  * Added the `beekeeper::hive::mock` module, which has a `MockTaskRunner` for `apply`ing a worker in a mock context. This is useful for testing your `Worker`.
+  * Updated to `2024` edition and Rust version `1.85`
 
 ## 0.2.1
 

Cargo.toml

@@ -2,51 +2,65 @@
 name = "beekeeper"
 description = "A full-featured worker pool library for parallelizing tasks"
 version = "0.3.0"
-edition = "2021"
-rust-version = "1.80"
+edition = "2024"
+rust-version = "1.85"
 authors = ["John Didion <github@didion.net>"]
 repository = "https://github.com/jdidion/beekeeper"
 license = "MIT OR Apache-2.0"
 
 [dependencies]
+crossbeam-channel = "0.5.13"
 crossbeam-deque = "0.8.6"
+crossbeam-queue = "0.3.12"
 crossbeam-utils = "0.8.20"
+derive_more = { version = "2.0.1", features = ["debug"] }
+nanorand = { version = "0.7.0", default-features = false, features = [
+    "std",
+    "tls",
+] }
 num = "0.4.3"
 num_cpus = "1.16.0"
 parking_lot = "0.12.3"
 paste = "1.0.15"
+simple-mermaid = "0.2.0"
 thiserror = "1.0.63"
 # required with the `affinity` feature
 core_affinity = { version = "0.8.1", optional = true }
-# required with the `batching` feature
-crossbeam-queue = { version = "0.3.12", optional = true }
-# alternate channel implementations that can be enabled with features
-crossbeam-channel = { version = "0.5.13", optional = true }
+# required with alternate outcome channel implementations that can be enabled with features
+# NOTE: these version requirements could be relaxed as we don't actually depend on the
+# functionality of these crates internally (other than in tests)
 flume = { version = "0.11.1", optional = true }
 loole = { version = "0.4.0", optional = true }
 
 [dev-dependencies]
 divan = "0.1.17"
 itertools = "0.14.0"
 serial_test = "3.2.0"
-#rstest = "0.22.0"
+rstest = "0.22.0"
 stacker = "0.1.17"
+aquamarine = "0.6.0"
+simple-mermaid = "0.2.0"
 
 [[bench]]
 name = "perf"
 harness = false
 
 [features]
-default = []
+default = ["local-batch"]
 affinity = ["dep:core_affinity"]
-batching = ["dep:crossbeam-queue"]
+local-batch = []
 retry = []
-crossbeam = ["dep:crossbeam-channel"]
+crossbeam = []
 flume = ["dep:flume"]
 loole = ["dep:loole"]
 
+[lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = [
+    'cfg(coverage,coverage_nightly)',
+] }
+
 [package.metadata.cargo-all-features]
-allowlist = ["affinity", "batching", "retry"]
+allowlist = ["affinity", "local-batch", "retry"]
 
 [profile.release]
 lto = true