Expose the pool, require the ps-st3 that has the wake fixes

.github/workflows/publish.yml

@@ -43,6 +43,29 @@ jobs:
       - if: steps.version.outputs.changed == 'true'
         run: cargo check --no-default-features
 
+      # Checked before the upload rather than during it. This crate has no
+      # repository-level `CARGO_REGISTRY_TOKEN`, and every publish run so far
+      # short-circuited on "version unchanged", so the upload path has never
+      # actually run here. The first real release would be the first time the
+      # token is exercised, and an absent one surfaces as an authentication
+      # error from `cargo publish` after the package is built, which reads like
+      # a registry problem rather than a missing secret.
+      - if: steps.version.outputs.changed == 'true'
+        run: |
+          if [ -z "${CARGO_REGISTRY_TOKEN:-}" ]; then
+            echo "CARGO_REGISTRY_TOKEN is empty."
+            echo
+            echo "Set it on pathscale/nagoya, or at the organisation level with this"
+            echo "repository in scope:"
+            echo "  gh secret set CARGO_REGISTRY_TOKEN --repo pathscale/nagoya"
+            echo
+            echo "Nothing downstream resolves until this crate publishes: worktable"
+            echo "requires nagoya ^0.1.1, and nagoya requires ps-st3 ^0.6.1."
+            exit 1
+          fi
+        env:
+          CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
+
       - if: steps.version.outputs.changed == 'true'
         run: cargo publish --token "$CARGO_REGISTRY_TOKEN"
         env:

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nagoya"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2021"
 rust-version = "1.85"
 license = "MIT OR Apache-2.0"
@@ -14,7 +14,7 @@ categories = ["asynchronous", "concurrency", "no-std"]
 default = ["std"]
 # Off, this crate does not link `std`. What it costs: `Runtime`, which owns
 # threads, and the parker `block_on` uses when there is one to park.
-std = ["ps-st3/host", "dep:futures-io", "futures-io/std", "dep:futures-util", "futures-util/std", "futures-util/io"]
+std = ["ps-st3/host", "async-task/std", "dep:futures-io", "futures-io/std", "dep:futures-util", "futures-util/std", "futures-util/io"]
 
 
 [dependencies]

src/io.rs

@@ -79,9 +79,11 @@ pub trait Read {
         async move {
             let mut filled = 0;
             while filled < buffer.len() {
-                match self.read(&mut buffer[filled..]).await? {
-                    0 => return Err(Error::new(ErrorKind::UnexpectedEof)),
-                    n => filled += n,
+                match self.read(&mut buffer[filled..]).await {
+                    Ok(0) => return Err(Error::new(ErrorKind::UnexpectedEof)),
+                    Ok(n) => filled += n,
+                    Err(error) if error.kind() == ErrorKind::Interrupted => continue,
+                    Err(error) => return Err(error),
                 }
             }
             Ok(())
@@ -114,6 +116,9 @@ pub trait Read {
                         return Ok(filled - start);
                     }
                     Ok(n) => buffer.truncate(filled + n),
+                    Err(error) if error.kind() == ErrorKind::Interrupted => {
+                        buffer.truncate(filled);
+                    }
                     Err(error) => {
                         buffer.truncate(filled);
                         return Err(error);
@@ -144,11 +149,13 @@ pub trait Write {
         async move {
             let mut written = 0;
             while written < buffer.len() {
-                match self.write(&buffer[written..]).await? {
+                match self.write(&buffer[written..]).await {
                     // Nothing taken and no error: the destination will not
                     // accept more, and looping would spin forever.
-                    0 => return Err(Error::new(ErrorKind::WriteZero)),
-                    n => written += n,
+                    Ok(0) => return Err(Error::new(ErrorKind::WriteZero)),
+                    Ok(n) => written += n,
+                    Err(error) if error.kind() == ErrorKind::Interrupted => continue,
+                    Err(error) => return Err(error),
                 }
             }
             Ok(())

src/lib.rs

@@ -62,15 +62,25 @@ mod yield_now;
 
 pub use block_on::block_on;
 pub use par::{par_for, Cancel, ParFor};
+/// The idle policy a pool runs with, from the queues underneath.
+///
+/// Re-exported because selecting one is a nagoya-level decision and every
+/// caller that makes it would otherwise have to name `ps-st3` as a direct
+/// dependency to spell the type, for one struct, in a graph that already
+/// contains it transitively.
+pub use st3::fanout::Tuning;
 pub use time::{
     now_ns, poll as poll_timers, set_clock, sleep, sleep_until, timeout, Elapsed, Sleep, Timeout,
 };
 pub use yield_now::{yield_now, YieldNow};
 
 /// A handle to a spawned task's output.
 ///
-/// Awaiting it yields `Some(output)`, or `None` if the task was cancelled by
-/// dropping the handle before it finished.
+/// Awaiting yields `Some(output)`, or `None` if its runnable was canceled
+/// because the pool was destroyed before it could run. With `std` and
+/// unwinding enabled, a task panic is rethrown when its handle is awaited,
+/// rather than terminating the worker. Without `std`, panic handling belongs
+/// to the host. Abort-on-panic builds cannot isolate a panic in either mode.
 ///
 /// **Dropping this detaches the task; it does not cancel it.** That is Tokio's
 /// behaviour and it was this crate's before, and it is worth stating because
@@ -80,11 +90,11 @@ pub use yield_now::{yield_now, YieldNow};
 /// [`cancel`](JoinHandle::cancel) is how to ask for the other thing.
 pub struct JoinHandle<T> {
     /// `None` only between `Drop` taking it and the handle going away.
-    task: Option<async_task::Task<T>>,
+    task: Option<async_task::FallibleTask<T>>,
 }
 
 impl<T> JoinHandle<T> {
-    fn task(&mut self) -> &mut async_task::Task<T> {
+    fn task(&mut self) -> &mut async_task::FallibleTask<T> {
         self.task
             .as_mut()
             .expect("the task is taken only by `Drop`")
@@ -122,7 +132,7 @@ impl<T> core::fmt::Debug for JoinHandle<T> {
                 &self
                     .task
                     .as_ref()
-                    .is_some_and(async_task::Task::is_finished),
+                    .is_some_and(async_task::FallibleTask::is_finished),
             )
             .finish()
     }
@@ -140,16 +150,33 @@ impl<T> Future for JoinHandle<T> {
     type Output = Option<T>;
 
     fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
-        Pin::new(self.get_mut().task()).poll(context).map(Some)
+        Pin::new(self.get_mut().task()).poll(context)
     }
 }
 
+/// Host-supplied identity of the execution lane currently running a pool worker.
+///
+/// Implementations must return an identity for the calling lane, not for the
+/// lane that originally created the executor. Return `None` off a worker or
+/// for a different pool. Invalid worker indices fall back to shared routing.
+/// Keep this lookup bounded and nonblocking: it runs on every task wake.
+/// Context implementations must not strongly own their pool; use a weak
+/// reference or a host-owned identity to avoid a queue ownership cycle.
+///
+/// This interface requires only `core`/`alloc`. A no_std host can use its
+/// scheduler's execution-local storage without acquiring OS-thread ownership.
+pub trait WorkerContext: Send + Sync {
+    /// The calling lane's worker index for `pool`, if it is currently running it.
+    fn current_worker(&self, pool: &Pool) -> Option<usize>;
+}
+
 /// A pool with a place to put the next task.
 ///
 /// Work goes to one injector that every worker drains, so there is nothing for
 /// a spawn to choose: the pool decides who runs it.
 pub struct Executor {
     pool: Arc<Pool>,
+    context: Option<Arc<dyn WorkerContext>>,
 }
 
 impl Executor {
@@ -159,7 +186,32 @@ impl Executor {
     /// which is what keeps this free of an operating system.
     #[must_use]
     pub fn new(pool: Arc<Pool>) -> Self {
-        Self { pool }
+        Self {
+            pool,
+            context: None,
+        }
+    }
+
+    /// Use host-provided execution-local identity for locality scheduling.
+    ///
+    /// Unlike `Runtime`, this does not create threads or require `std`.
+    #[must_use]
+    pub fn with_worker_context(pool: Arc<Pool>, context: Arc<dyn WorkerContext>) -> Self {
+        Self {
+            pool,
+            context: Some(context),
+        }
+    }
+
+    /// Run a caller-owned worker thread with nagoya's thread-local identity.
+    ///
+    /// The runner must belong to this executor's pool. Returns `false` if the
+    /// worker is already running, as `Pool::run` does. For a custom/no_std host,
+    /// use `with_worker_context` and drive the pool directly.
+    #[cfg(feature = "std")]
+    pub fn run_worker(&self, runner: st3::fanout::Runner) -> bool {
+        let _current = task::mark_current(&self.pool, runner.id());
+        self.pool.run(runner)
     }
 
     /// Another handle to the same pool, for a task that spawns.
@@ -168,7 +220,10 @@ impl Executor {
     /// is one: another handle to the same pool.
     #[must_use]
     pub fn clone_handle(&self) -> Self {
-        Self::new(self.pool.clone())
+        Self {
+            pool: self.pool.clone(),
+            context: self.context.clone(),
+        }
     }
 
     /// The pool this executor spawns onto.
@@ -184,7 +239,7 @@ impl Executor {
         F::Output: Send + 'static,
     {
         JoinHandle {
-            task: Some(task::spawn(future, self.pool.clone())),
+            task: Some(task::spawn(future, self.pool.clone(), self.context.clone()).fallible()),
         }
     }
 }