v2: Propagate panics, add AsyncHandle

.github/workflows/ci.yaml

@@ -1,18 +1,31 @@
 name: CI
-on: [push, pull_request]
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
 jobs:
   build:
     name: Build
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        rust:
+          - stable
+          - beta
+          - nightly
+          - 1.45.0
     steps:
       - name: Check out code
         uses: actions/checkout@v2
       - name: Install Rust toolchain
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: stable
+          toolchain: ${{ matrix.rust }}
           profile: minimal
-          components: clippy, rustfmt
+          components: clippy
           override: true
       - name: Cache Rust dependencies
         uses: Swatinem/rust-cache@v1
@@ -21,11 +34,18 @@ jobs:
         with:
           command: install
           args: cargo-make
-      - name: Build and lint
+      - name: Build
+        if: ${{ matrix.rust != 'stable' }}
         uses: actions-rs/cargo@v1
         with:
           command: make
           args: ci-build
+      - name: Build and lint
+        if: ${{ matrix.rust == 'stable' }}
+        uses: actions-rs/cargo@v1
+        with:
+          command: make
+          args: ci-lint
   format:
     name: Format
     runs-on: ubuntu-latest
@@ -37,7 +57,7 @@ jobs:
         with:
           toolchain: stable
           profile: minimal
-          components: clippy, rustfmt
+          components: rustfmt
           override: true
       - name: Cache Rust dependencies
         uses: Swatinem/rust-cache@v1
@@ -54,15 +74,22 @@ jobs:
   test:
     name: Test
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        rust:
+          - stable
+          - beta
+          - nightly
+          - 1.45.0
     steps:
       - name: Check out code
         uses: actions/checkout@v2
       - name: Install Rust toolchain
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: stable
+          toolchain: ${{ matrix.rust }}
           profile: minimal
-          components: clippy, rustfmt
+          components: ""
           override: true
       - name: Cache Rust dependencies
         uses: Swatinem/rust-cache@v1
@@ -77,6 +104,7 @@ jobs:
           command: make
           args: ci-test
       - name: Upload to codecov.io
+        if: ${{ matrix.rust == 'stable' }}
         uses: codecov/codecov-action@v1
         with:
           fail_ci_if_error: true

CHANGELOG.md

@@ -0,0 +1,16 @@
+# Changelog
+
+## 2.0.0
+
+- Use `std::panic` to propagate panics from the thread pool into the async
+  context, rather than triggering the Rayon panic handler.
+- Add `AsyncRayonHandle` type that implements `Future`, which makes the
+  `async-trait` crate unnecessary.
+- Bypass Tokio `RecvError`. We control the `Sender`, so it should never be
+  dropped too early.
+- Remove `prelude` module.
+- Seal `AsyncThreadPool` trait.
+
+## 1.0.0
+
+- Initial release

Cargo.toml

@@ -1,23 +1,23 @@
 [package]
 name = "tokio-rayon"
-version = "1.0.1-dev"
+version = "2.0.0-dev"
 authors = ["Andy Barron <andrewlbarron@gmail.com>"]
 edition = "2018"
 description = "Mix async code with CPU-heavy thread pools using Tokio + Rayon"
-license-file = "LICENSE.txt"
+license = "MIT"
+readme = "README.md"
 repository = "https://github.com/andybarron/tokio-rayon"
 keywords = ["tokio", "rayon", "async", "thread-pool"]
 categories = ["asynchronous", "concurrency", "rust-patterns"]
 
 [dependencies]
-tokio = { version = "^1.4.0", default-features = false, features = ["sync"] }
-async-trait = "^0.1.48"
-rayon = "^1.5.0"
+tokio = { version = "1.4.0", default-features = false, features = ["sync"] }
+rayon = "1.5.0"
 
 [dev-dependencies]
-tokio-test = "^0.4.1"
+tokio-test = "0.4.1"
 
 [dev-dependencies.tokio]
-version = "^1.4.0"
+version = "1.4.0"
 default-features = false
 features = ["sync", "macros", "rt", "time", "rt-multi-thread"]

LICENSE.txt → LICENSE.md

@@ -1,6 +1,6 @@
-MIT License
+# MIT License
 
-Copyright (c) 2021 Andy Barron
+**Copyright (c) 2021 Andy Barron**
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

Makefile.toml

@@ -19,13 +19,18 @@ args = ["release", "${@}"]
 
 [tasks.ci-build]
 command = "cargo"
-args = ["clippy", "--verbose", "--all-targets"]
+args = ["check", "--verbose", "--all-targets"]
 env = { "RUSTFLAGS" = "-D warnings" }
 
 [tasks.ci-format]
 command = "cargo"
 args = ["fmt", "--", "--check", "--verbose"]
 
+[tasks.ci-lint]
+command = "cargo"
+args = ["clippy", "--verbose", "--all-targets"]
+env = { "RUSTFLAGS" = "-D warnings" }
+
 [tasks.ci-test]
 command = "cargo"
 args = ["tarpaulin", "-v", "--all-features", "--ignore-tests", "--out", "Xml"]

README.md

@@ -5,24 +5,34 @@ _Mix async code with CPU-heavy thread pools using [Tokio][tokio-url] + [Rayon][r
 [tokio-url]: https://docs.rs/tokio
 [rayon-url]: https://docs.rs/rayon
 
-[![crates.io][crates-badge]][crates-url]
 [![Documentation][docs-badge]][docs-url]
 [![Build status][build-badge]][build-url]
 [![Test coverage][coverage-badge]][coverage-url]
+<br />
+[![crates.io][crates-badge]][crates-url]
+[![Downloads][downloads-badge]][crates-url]
+[![Rust version][rust-version-badge]][rust-version-link]
+<br />
 [![MIT license][license-badge]][license-url]
 
-[crates-badge]: https://img.shields.io/crates/v/tokio-rayon.svg
-[crates-url]: https://crates.io/crates/tokio-rayon
-[docs-badge]: https://docs.rs/tokio-rayon/badge.svg
-[docs-url]: https://docs.rs/tokio-rayon
-[license-badge]: https://img.shields.io/badge/license-MIT-blue.svg
-[license-url]: https://github.com/andybarron/tokio-rayon/blob/master/LICENSE
-[build-badge]: https://github.com/andybarron/tokio-rayon/actions/workflows/ci.yaml/badge.svg
+[build-badge]: https://img.shields.io/github/workflow/status/andybarron/tokio-rayon/CI?labelColor=112&logo=github&logoColor=fff&style=flat-square
 [build-url]: https://github.com/andybarron/tokio-rayon/actions
-[coverage-badge]: https://codecov.io/gh/andybarron/tokio-rayon/branch/main/graph/badge.svg
+[coverage-badge]: https://img.shields.io/codecov/c/gh/andybarron/tokio-rayon?labelColor=112&logo=codecov&logoColor=fff&style=flat-square
 [coverage-url]: https://codecov.io/gh/andybarron/tokio-rayon
+[crates-badge]: https://img.shields.io/crates/v/tokio-rayon?labelColor=112&logo=rust&logoColor=fff&style=flat-square
+[crates-url]: https://crates.io/crates/tokio-rayon
+[docs-badge]: https://img.shields.io/docsrs/tokio-rayon?labelColor=112&logo=read-the-docs&logoColor=fff&style=flat-square
+[docs-url]: https://docs.rs/tokio-rayon
+[downloads-badge]: https://img.shields.io/crates/d/tokio-rayon?labelColor=112&color=informational&style=flat-square
+[license-badge]: https://img.shields.io/crates/l/tokio-rayon?labelColor=112&style=flat-square
+[license-url]: https://github.com/andybarron/tokio-rayon/blob/main/LICENSE.txt
+[rust-version-badge]: https://img.shields.io/badge/rustc-1.45+-informational?logo=rust&logoColor=fff&labelColor=112&style=flat-square
+[rust-version-link]: https://www.rust-lang.org
+
+## Resources
 
-## [API documentation][docs-url]
+- [**Documentation**][docs-url]
+- [crates.io][crates-url]
 
 ## TL;DR
 

codecov.yml

@@ -0,0 +1,8 @@
+coverage:
+  status:
+    project:
+      default:
+        target: 85%
+        threshold: 100%
+    patch: off
+    changes: off

src/async_handle.rs

@@ -0,0 +1,61 @@
+use std::future::Future;
+use std::panic::resume_unwind;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::thread;
+use tokio::sync::oneshot::Receiver;
+
+/// Async handle for a blocking task running in a Rayon thread pool.
+///
+/// If the spawned task panics, `poll()` will propagate the panic.
+#[must_use]
+#[derive(Debug)]
+pub struct AsyncRayonHandle<T> {
+    pub(crate) rx: Receiver<thread::Result<T>>,
+}
+
+impl<T> Future for AsyncRayonHandle<T> {
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let rx = Pin::new(&mut self.rx);
+        rx.poll(cx).map(|result| {
+            result
+                .expect("Unreachable error: Tokio channel closed")
+                .unwrap_or_else(|err| resume_unwind(err))
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test::init;
+    use std::panic::catch_unwind;
+    use std::thread;
+    use tokio::sync::oneshot::channel;
+
+    #[tokio::test]
+    #[should_panic(expected = "Task failed successfully")]
+    async fn test_poll_propagates_panic() {
+        init();
+        let panic_err = catch_unwind(|| {
+            panic!("Task failed successfully");
+        })
+        .unwrap_err();
+
+        let (tx, rx) = channel::<thread::Result<()>>();
+        let handle = AsyncRayonHandle { rx };
+        tx.send(Err(panic_err)).unwrap();
+        handle.await;
+    }
+
+    #[tokio::test]
+    #[should_panic(expected = "Unreachable error: Tokio channel closed")]
+    async fn test_unreachable_channel_closed() {
+        init();
+        let (_, rx) = channel::<thread::Result<()>>();
+        let handle = AsyncRayonHandle { rx };
+        handle.await;
+    }
+}