Sync wrapper for BlockingPermitFuture

.travis.yml

@@ -6,6 +6,11 @@ script:
   - 'cargo test'
   - 'cargo test --features current-thread'
   - 'cargo test --no-default-features --features tokio-oneshot'
+  - 'cargo test --no-default-features --features tokio-threaded,tokio-oneshot'
+  - 'cargo test --no-default-features --features tokio-semaphore,tokio-oneshot'
+  - 'cargo test --no-default-features --features tokio-omnibus'
+  - 'cargo test --features futures-intrusive'
+  - 'cargo test --features tokio-threaded,futures-intrusive'
   - 'cargo test --all-features'
   - 'if [ "$TRAVIS_RUST_VERSION" == "nightly" ]; then cargo build --all-features --all-targets; fi'
 
@@ -17,6 +22,8 @@ before_cache:
   - rm -rf /home/travis/.cargo/registry
 
 env: # important!
+  global: RUSTFLAGS=-Dwarnings
+
 matrix:
   include:
     - rust: 1.39.0

Cargo.toml

@@ -16,29 +16,38 @@ build         = "build.rs"
 [dependencies]
 futures-core        = { version=">=0.3.1, <0.4" }
 futures-channel     = { version=">=0.3.1, <0.4", optional=true }
-lazy_static         = { version=">= 1.3.0, <1.5" }
 log                 = { version=">=0.4.4, <0.5" }
 crossbeam-channel   = { version=">=0.4.0, <0.5" }
 num_cpus            = { version=">=1.11.1, <1.12" }
-futures-intrusive   = { version=">=0.2.2, <0.3" }
+futures-intrusive   = { version=">=0.2.2, <0.3", optional=true }
 tokio               = { version=">=0.2.2, <0.3", optional=true }
 
 [dev-dependencies]
 env_logger          = { version=">=0.7.1, <0.8", default-features=false }
-tempfile            = { version=">=3.1.0, <3.2" }
-rand                = { version=">=0.7.0, <0.8" }
 futures-executor    = { version=">=0.3.1, <0.4" }
 futures-util        = { version=">=0.3.1, <0.4" }
+lazy_static         = { version=">=1.3.0, <1.5" }
+rand                = { version=">=0.7.0, <0.8" }
+tempfile            = { version=">=3.1.0, <3.2" }
 
 [features]
+# Need some oneshot channel for dispatch_rx, default to futures
+# Use default-features=false in combination with tokio-oneshot, if desired.
 default=["futures-channel"]
-tokio-oneshot=[
-    "tokio", "tokio/sync"
-]
-tokio-threaded=[
-    "tokio", "tokio/rt-core", "tokio/rt-threaded", "tokio/blocking"
-]
-current-thread=[] # testing hack
+
+# All available tokio integrations
+tokio-omnibus=["tokio-oneshot", "tokio-threaded", "tokio-semaphore"]
+
+# Instead of (default) futures-channel oneshot
+tokio-oneshot=["tokio", "tokio/sync"]
+
+# Tokio threaded runtime, including blocking support
+tokio-threaded=["tokio", "tokio/rt-core", "tokio/rt-threaded", "tokio/blocking"]
+
+# Instead of futures-intrusive semaphore. (Both are optional)
+tokio-semaphore=["tokio", "tokio/sync"]
+
+current-thread=[] # testing hack only
 
 [profile.bench]
 lto = "thin"

benches/dispatch.rs

@@ -1,3 +1,4 @@
+#![cfg(any(feature = "tokio-semaphore", feature = "futures-intrusive"))]
 #![warn(rust_2018_idioms)]
 
 #![feature(test)]
@@ -19,11 +20,13 @@ use blocking_permit::{
     blocking_permit_future,
     dispatch_rx, DispatchPool,
     deregister_dispatch_pool, register_dispatch_pool,
-    Semaphore
+    Semaphore,
+    Semaphorish,
 };
 
 lazy_static! {
-    static ref TEST_SET: Semaphore = Semaphore::new(true, 20);
+    static ref DEFAULT_SET: Semaphore = Semaphore::default_new(4);
+    static ref SLEEP_SET: Semaphore = Semaphore::default_new(20);
 }
 
 #[cfg(feature="tokio-threaded")]
@@ -34,7 +37,8 @@ fn noop_threaded_dispatch_rx(b: &mut Bencher) {
         .create();
 
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4)
         .threaded_scheduler()
         .on_thread_start(move || {
             register_dispatch_pool(pool.clone());
@@ -48,7 +52,7 @@ fn noop_threaded_dispatch_rx(b: &mut Bencher) {
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -70,15 +74,16 @@ fn noop_threaded_dispatch_rx(b: &mut Bencher) {
 #[bench]
 fn noop_threaded_spawn_blocking(b: &mut Bencher) {
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4+4+1)
         .threaded_scheduler()
         .build()
         .unwrap();
 
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -100,15 +105,17 @@ fn noop_threaded_spawn_blocking(b: &mut Bencher) {
 #[bench]
 fn noop_threaded_in_place(b: &mut Bencher) {
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4+4+1)
         .threaded_scheduler()
         .build()
         .unwrap();
 
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
+                    .make_sync()
                     .await
                     .unwrap();
                 let r = p.run(|| 41);
@@ -134,7 +141,7 @@ fn noop_local_dispatch_rx(b: &mut Bencher) {
         let sp = pool.spawner();
         for _ in 0..100 {
             sp.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -157,7 +164,8 @@ fn r_expensive_threaded_dispatch_rx(b: &mut Bencher) {
         .create();
 
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4)
         .threaded_scheduler()
         .on_thread_start(move || {
             register_dispatch_pool(pool.clone());
@@ -171,7 +179,7 @@ fn r_expensive_threaded_dispatch_rx(b: &mut Bencher) {
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -198,15 +206,16 @@ fn r_expensive_threaded_dispatch_rx(b: &mut Bencher) {
 #[bench]
 fn r_expensive_threaded_spawn_blocking(b: &mut Bencher) {
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4+4+1)
         .threaded_scheduler()
         .build()
         .unwrap();
 
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -228,15 +237,17 @@ fn r_expensive_threaded_spawn_blocking(b: &mut Bencher) {
 #[bench]
 fn r_expensive_threaded_in_place(b: &mut Bencher) {
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4+4+1)
         .threaded_scheduler()
         .build()
         .unwrap();
 
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
+                    .make_sync()
                     .await
                     .unwrap();
                 let r = p.run(|| expensive_comp());
@@ -262,7 +273,7 @@ fn r_expensive_local_dispatch_rx(b: &mut Bencher) {
         let sp = pool.spawner();
         for _ in 0..100 {
             sp.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&DEFAULT_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -285,7 +296,8 @@ fn sleep_threaded_dispatch_rx(b: &mut Bencher) {
         .create();
 
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4)
         .threaded_scheduler()
         .on_thread_start(move || {
             register_dispatch_pool(pool.clone());
@@ -299,7 +311,7 @@ fn sleep_threaded_dispatch_rx(b: &mut Bencher) {
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&SLEEP_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -321,15 +333,16 @@ fn sleep_threaded_dispatch_rx(b: &mut Bencher) {
 #[bench]
 fn sleep_threaded_spawn_blocking(b: &mut Bencher) {
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(4)
+        .core_threads(4)
+        .max_threads(4+20+1)
         .threaded_scheduler()
         .build()
         .unwrap();
 
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&SLEEP_SET)
                     .await
                     .unwrap();
                 p.enter();
@@ -351,15 +364,17 @@ fn sleep_threaded_spawn_blocking(b: &mut Bencher) {
 #[bench]
 fn sleep_threaded_in_place(b: &mut Bencher) {
     let mut rt = tokio::runtime::Builder::new()
-        .num_threads(20)
+        .core_threads(4)
+        .max_threads(4+20+1)
         .threaded_scheduler()
         .build()
         .unwrap();
 
     b.iter(|| {
         let futures: FuturesUnordered<_> = (0..100).map(|_| {
             rt.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&SLEEP_SET)
+                    .make_sync()
                     .await
                     .unwrap();
                 let r = p.run(|| random_sleep());
@@ -385,7 +400,7 @@ fn sleep_local_dispatch_rx(b: &mut Bencher) {
         let sp = pool.spawner();
         for _ in 0..100 {
             sp.spawn(async {
-                let p = blocking_permit_future(&TEST_SET)
+                let p = blocking_permit_future(&SLEEP_SET)
                     .await
                     .unwrap();
                 p.enter();

src/fs.rs

@@ -1,15 +1,16 @@
-// TODO: This is currently just a subset of potential usage from tokio_fs
+// TODO: This is currently just a subset of potential usage from tokio_fs,
+// retained for testing.
 
 use std::fs;
 use std::io;
 use std::path::Path;
 
 use lazy_static::lazy_static;
 
-use crate::{dispatch_or_permit, Semaphore};
+use crate::{dispatch_or_permit, Semaphore, Semaphorish};
 
 lazy_static! {
-    pub static ref BLOCKING_SET: Semaphore = Semaphore::new(true, 1);
+    static ref BLOCKING_SET: Semaphore = Semaphore::default_new(1);
 }
 
 /// Creates a new, empty directory at the provided path
@@ -71,7 +72,8 @@ mod tests {
 
         {
             let mut rt = tokio::runtime::Builder::new()
-                .num_threads(2)
+                .core_threads(2)
+                .max_threads(2)
                 .threaded_scheduler()
                 .build()
                 .unwrap();