Confused about Send, Arc, and Mutex when passing dyn trait function parameters

help
1

For a while now I've been able to skirt the intricacies of multithread programming in Rust by simply moving data into Tokio tasks as they spawn and/or letting the compiler auto-implement Send (presumably). Today, however, I hit a snag while trying to pass a boxed dyn trait to an async API in the context of a Tokio spawned thread that requires its futures to be Send.

error: future cannot be sent between threads safely
   --> src\main.rs:153:11
    |
153 |     tasks.spawn(async move {
    |           ^^^^^ future created by async block is not `Send`
    |
    = help: within `{async block@src\main.rs:153:17: 153:27}`, the trait `Send` is not implemented for `NonNull<rclite::rc::RcInner<RefCell<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>>`
note: future is not `Send` as this value is used across an await
   --> src\main.rs:188:10
    |
159 |         let task_progess_subscription = task_progress_observer.clone().subscribe(move |v: f64| {
    |             ------------------------- has type `SubjectSubscription<MutRc<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>, LocalScheduler, CellRc<SubscriptionState>>` which is not `Send`
...
188 |         .await
    |          ^^^^^ await occurs here, with `task_progess_subscription` maybe used later
note: required by a bound in `JoinSet::<T>::spawn`
   --> ...tokio-1.48.0\src\task\join_set.rs:145:12
    |
142 |     pub fn spawn<F>(&mut self, task: F) -> AbortHandle
    |            ----- required by a bound in this associated function
...
145 |         F: Send + 'static,
    |            ^^^^ required by this bound in `JoinSet::<T>::spawn`

The idea is to pass a payload to a remote address alongside an rxrust Subject which will act as an Observer and receive Observer::next() calls with percentage progress updates, but even though everything is local to the task and inside the async move block, I get the error above. Here's the first iteration of the code:

use rxrust::prelude::*;
use tokio::task;
...
pub type ProgressPercentageObserver = Box<dyn Observer<f64, std::convert::Infallible>>;
...
let mut tasks = task::JoinSet::new();
tasks.spawn(async move {
    let addr = String::from("test_ip");
    let payload = Vec::new();
    let task_progress_observer = Local::subject();
    let subscription_addr = addr.clone();
    let task_progess_subscription = task_progress_observer.clone().subscribe(move |v: f64| {
        tracing::info!("Progress for {}: {:.2}", subscription_addr, v);
    });
    let boxed_task_progress_observer: ProgressPercentageObserver =
        Box::new(task_progress_observer);

    let res = match upload(
        &addr,
        &payload,
        &mut Some(boxed_task_progress_observer),
    )
    .await
    {
        Ok(()) => Ok(format!(
            "Successfully uploaded {} bytes to address {}.",
            &payload.len(),
            &addr
        )),
        Err(e) => anyhow::bail!(e),
    };
    task_progess_subscription.unsubscribe();
    res
});

I figured I didn't need to use the rxrust Shared context because everything was essentially local to the current task, wherever said task might be running. I tried using Shared::subject() instead of Local::subject() in the above just in case it helped, and got the same error anyway, which is really puzzling. Evidently the issue is about my ProgressPercentageObserver type, which is really a boxed dyn Observer trait that expects f64 events. I tried just adding + Send to its declaration as Box<dyn Observer<f64, std::convert::Infallible> + Send>, but that didn't seem to change anything on its own.

Meanwhile, if I use a bespoke callback handler instead of an rxrust dynamically implemented Observer, everything seems fine:

use tokio::task;
...
pub trait ProgressHandling {
    fn handle_progress_event();
}
#[derive(Default)]
pub struct ProgressHandler {}
impl ProgressHandling for ProgressHandler {
    fn handle_progress_event(addr: &String, progress: f64) {
       tracing::info!("Progress for {}: {:.2}", addr, progress);
    }
}
let progress = ProgressHandler::default();
...
let mut tasks = task::JoinSet::new();
tasks.spawn(async move {
    let addr = String::from("test_ip");
    let payload = Vec::new();

    let res = match upload(
        &addr,
        &payload,
        &mut Some(progress),
    )
    .await
    {
        Ok(()) => Ok(format!(
            "Successfully uploaded {} bytes to address {}.",
            &payload.len(),
            &addr
        )),
        Err(e) => anyhow::bail!(e),
    };
    res
});

This approach is apparently fine, with the modified async upload() expecting an Option<impl ProgressHandling>. I also found that the compiler seems happy with a ProgressPercentageObserver that specifies + Send AND is actually used in the rxrust Shared context; either of those modifications without the other frustratingly results in the same error from above. Finally, wrapping the ProgressPercentageObserver in a Mutex wrapped in an Arc and modifying the upload() API to expect a mutable reference to the underlying ProgressPercentageObserver (to avoid attempts to move the underlying Box out of the Mutex, which it doesn't allow) also seems to work without having to modify the ProgressPercentageObserver type declaration or using rxrust Shared context.

So after that journey, I'm left with several questions:

  1. What does the compiler actually mean by note: future is not Send as this value is used across an await? Is the across an await part referring to the fact that the async move block could theoretically access the ProgressPercentageObserver again after upload().await? Why isn't that a problem when ProgressPercentageObserver is Send via + Send in its declaration plus using Shared::subject(), or Send implicitly via Tokio Mutex? It still could theoretically be used after the await.
  2. Why is Tokio spawn() happy with String and Vector and impl trait, but not dyn trait? Does it have something to do with limitations on the compiler auto-generating Send for dyn traits?
  3. Why does adding + Send to the dyn trait declaration combined with using rxrust's Shared context satisfy the Sendable future constraint? Why does locking the dyn trait behind a Mutex satisfy the Sendable future constaint? Why can't the future be Send implicitly simply because all its data is effectively thread and task local because it all lives inside the spawned task's async move block?
  4. What is the conceptual difference between Arc and Mutex with respect to Send? I think the main difference comes down to lifetime (made thread-safe by Arc) vs. access (made thread-safe by Mutex) and to be fully thread safe you need both?
  5. In terms of API design, is it preferred to enforce that the dyn trait parameter of a library function is Send via + Send in its declaration so that it can be used in multithreaded contexts or to leave the responsibility of making the dyn trait Sendable when necessary up to the calling client? I'm leaning towards the latter since nothing about the library function implies multithread vs. single thread usage, but the former has slightly cleaner syntax.
2

I don’t use async enough to speak to some of this, but I can say this much: dyn Trait (where Trait isn’t Send or Sync) does not implement Send or Sync. This isn’t some unfortunate compiler limitation; it’s because somebody could genuinely stuff a !Send + !Sync type into a dyn Trait, while that’s not possible for dyn Trait + Send + Sync.

impl Trait, conversely, is actually a single concrete type inferred by the compiler. (And, yes, it’s sort of hacky that impl Trait leaks information about whether the supposedly “opaque type” implements Send and Sync.)

Note that values that are held across an await point are, AFAIK, part of the future’s state. Like, a concrete compiler-generated state struct (or enum). If a value held across an await point is !Send, then the future is !Send. AFAIK.

Unfortunately, it doesn’t seem like there’s currently a good way to be “generic over trait bounds” in some form.

As for Arc and Mutex… no, I wouldn’t say you need both. If you need threadsafe shared & access to some expensive-to-clone thing, Arc is good enough. Mutex only provides a way to go from shared & access to exclusive &mut access. You could share a &Mutex<T> across threads and achieve much the same as Arc<Mutex<T>>; but, then, you wouldn’t usually satisfy a 'static bound. Thread-related stuff requiring 'static isn’t intrinsic to threadsafety; it’s related to how the threads are created and joined/destroyed. Some thread APIs (like scoped threads) actually don’t require a 'static bound, meaning that &Mutex<T> could suffice in those cases (at least).

3

"local" is a misleading terminology here. an async task would be suspended at any await-ing point and might be polled again from a different thread, thus what's seems "local" in the sense of lexical scopes is not "local" in the sense of the execution threads.

if you can't modify the code to use a Send type, then you can use a single threaded execution environment that explicitly does NOT require the Send bound for the spawn() primitive. one such example is tokio::spawn_local().

I don't know what is the Observable type, but the compiler diagnostics seems indicate there's some non-Send types involed, such as MutRc, LocalScheduler. I think the api is intentionally not thread safe. maybe there's some thread safe variant api from the library you use?

4

Well, whatever the type involved,[1] checking if the type implements Send or not is the same as any other trait check, like "are you Clone or not". So we're really talking about when Send is automatically implemented or not.

Your typical struct/enum types, like String and Vector, automatically implement Send if all their fields do. (Or you can override the automatic behavior with explicit implementations.) spawn is "happy with these" because they definitely meet the required bound.

For each distinct -> impl Trait, there's a single hidden type underneath. The compiler still knows that hidden type, and allows the auto-trait details to leak through, even if you don't always require them (-> impl Send + Trait). When spawn is "happy with these", it's because the hidden type meets the required bound. You could work up examples where they do not, though (if you didn't require it with -> Send + ...).

In contrast, for dyn Trait, there's an open set of possible erased types underneath, which is not tracked at compile time. If dyn Trait automatically implemented Send, and you coerced an implementor which was not Send to dyn Trait, that would be unsound. But pretty much the entire point of dyn Trait is that you can coerce to dyn Trait if you meet the Trait bound. So dyn Trait does not implement Send automatically unless the trait requires all implementors to be Send (trait Trait: Send). It's not a limitation so much as a design choice.

Thus spawn is "not happy with dyn Trait" because dyn Trait in your case definitely does not meet the required bound. dyn Trait + Send definitely does meet the bound -- and this is sound because you can only coerce to dyn Trait + Send if you meet both the Trait and the Send bound.

Could the Send of the erased type be tracked in, say, the vtable? Yes, but but that wouldn't help with trait bounds like those on spawn, which are conditions on types and not particular values. Instead you would probably end up with some sort of fallible conversion from dyn Trait to dyn Trait + Send... but if you always need the Send anyway, what's the point.

I'm not familiar with rxrust, but my guess is that it's something like Local::subject() is never Send, and Shared::subject() is Send if some parameter is Send. (Like maybe we're talking about Shared<R>? Shared<R> is Send if R: Send, which is typical when R is the type of one of your fields.)

That's more puzzling. Perhaps you changed things enough you avoided holding state across an await.

Trait bounds check properties of types, not of values -- and certainly not of how values are used in some particular code block.

Also, it's probably not true that all the data is effectively thread and task local. If you do have some state held at an await point, that await point is an opportunity for the async runtime to send the task (state) to some other thread (e.g. for "work stealing") and resume execution in the middle of the syntactic block. (This point goes to your question (1) as well.)

The Send bound on spawn is present because this may happen. There are other runtimes that don't do this and don't need a Send bound.

I don't think it'll really help here, but anyway. Prerequisite for understanding: U: Sync if and only if &U: Send.

Arc<T> can act like a &T (usually) or a Box<T> which can move out of (if it's the only remaining owner). So Arc<T>: Send if T: Send + Sync.

Mutex<T> doesn't act like a &T, but you can move out of it. So Mutex<T>: Send if T: Send.

Lifetimes aren't a consideration here. The 'static requirement on spawn is independent of a Send requirement. Arc<T>: 'static if and only if T: 'static and similar for Mutex<T>. Those bounds are also type requirements, and not requirements on how long values actually last.[2]

  1. dyn Trait and dyn Trait + Send are (two distinct) concrete types, as are String and Vector and every distinct -> impl Trait ↩︎

  2. "lifetime" is an overloaded and confusing term in Rust. ↩︎

5

There is -- that's rxrust's Shared context. However, if I just use the Shared context intended for multithreaded usage, the compiler still complained. I had to add an explicit + Send to the actual ProgressPercentageObserver type declaration and also use rxrust's Shared context to get the compiler to give it a pass. I think the scenario was essentially that adding + Send told the compiler that the ProgressPercentageObserver should be Sendable, and using the rxrust Shared context allowed ProgressPercentageObserver to actually be Sendable since the rxrust context wraps the Observer type and only the Shared context supports thread-safety. Notably, before I had + Send on ProgressPercentageObserver the compiler error called out tasks.spawn() and a future that was not Send as the problem origin (and gave the error in my original post) whereas if I tried to do

pub type ProgressPercentageObserver = Box<dyn Observer<f64, std::convert::Infallible> + Send>;
let task_progress_observer = Local::subject();
...
let boxed_task_progress_observer: ProgressPercentageObserver = Box::new(task_progress_observer);

I get the following error citing the call to Box::new()

error[E0277]: `NonNull<rclite::rc::RcInner<RefCell<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>>` cannot be sent between threads safely
   --> src\main.rs:167:13
    |
167 |             Box::new(task_progress_observer);
    |             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `NonNull<rclite::rc::RcInner<RefCell<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>>` cannot be sent between threads safely
    |
    = help: within `LocalCtx<Subject<MutRc<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>, LocalScheduler>`, the trait `Send` is not implemented for `NonNull<rclite::rc::RcInner<RefCell<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>>`
note: required because it appears within the type `rclite::rc::Rc<RefCell<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>`
   --> ...rclite-0.4.1\src\rc.rs:27:12
    |
 27 | pub struct Rc<T> {
    |            ^^
note: required because it appears within the type `MutRc<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>`
   --> ...rxrust-1.0.0-rc.5\src\rc.rs:69:12
    |
 69 | pub struct MutRc<T>(Rc<RefCell<T>>);
    |            ^^^^^
note: required because it appears within the type `Subject<MutRc<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>`
   --> ...rxrust-1.0.0-rc.5\src\subject\subject_core.rs:337:12
    |
337 | pub struct Subject<P> {
    |            ^^^^^^^
note: required because it appears within the type `LocalCtx<Subject<MutRc<Subscribers<Box<dyn DynObserver<f64, Infallible>>>>>, LocalScheduler>`
   --> ...rxrust-1.0.0-rc.5\src\context.rs:170:12
    |
170 | pub struct LocalCtx<T, S> {
    |            ^^^^^^^^
    = note: required for the cast from `Box<LocalCtx<Subject<MutRc<Subscribers<Box<...>>>>, ...>>` to `Box<(dyn rxrust::Observer<f64, Infallible> + Send + 'static)>`
    = note: consider using `--verbose` to print the full type name to the console

which is mostly incomprehensible, but generally appears to say "you're using the thread-unsafe context as a vehicle for your Observer that needs to be Sendable, you dingus" XD

6

That brings up another question I had that maybe should be its own post depending on the complexity of the answer -- does the 'static lifetime as a type constraint necessarily mean that an actual value will be held in memory for the lifetime of the program? e.g. all the futures generated by calls to tasks.spawn() just hang around filling up memory until the program exits? Based on common-rust-lifetime-misconceptions by pretzelhammer, it seems the answer is not necessarily (at least if the type is an owned type), but I don't really trust my understanding here.

7

What does it mean to implement Send and/or Sync, though? In the docs, they have no member functions like typical traits. When I add + Send to a type, does that alone do anything or do I have to involve a sync primitive like Mutex somewhere in the usage of that type?

8

Turns out I was wrong and simply using a Mutex (Tokio or std) to wrap the Observer didn't actually help; I thought that would be sufficient to make the value "effectively Send" regardless of the type declaration by mutex locking it, but as you explained earlier the constraint checks are based on type rather than value so the premise doesn't make sense.

9

It does not. That would mean String never gets deallocated, for example, because String: 'static.

It does mean it would be sound to hold an actual value in memory for the entire duration of the program.

Rust lifetimes -- 'a things -- generally denote the duration of borrows. They don't directly represent liveness of a value. And a value can contain a borrow that lasts longer than the value itself. Otherwise it would be impossible to store a &'static str (or Cow<'static, str>, etc) in a local variable.

An approximation of what T: 'x means is "any borrows in T are at least as long as 'x". So for 'static specifically, T: 'static means "any borrows in T are valid for the rest of the of the program." And an approximation of that is "T holds no borrows", as in some sense, borrows that last forever might as well not be borrows at all.

These are approximations as the lifetime bounds are checked syntactically. So a dyn Trait<&'a str> + 'static does not meet a 'static bound for example, due to the presence of 'a.[1] It has to work this way for various parts of the type system to be sound.[2]

Borrow checking is a pass-or-fail test that doesn't change the semantics of your program. It's impossible to use Rust lifetime annotations to force a value to live longer. And again, Rust lifetimes don't even denote how long values live at all -- they are a type level property.

So this isn't the case. The 'static bound is required because the call to spawn returns immediately, and the caller is free to keep executing -- including dropping or mutating local values, say. But the futures still exist in the runtime and will generally get executed at some point. If those futures held borrows of the local values that got dropped or mutated, that would be unsound (executing the futures could result in use-after-free, data races, all that junk).

Thus spawn forbids the futures from holding any borrows -- or at least, any borrows that aren't valid "forever".

Spawning threads and immediately letting the caller continue executing has the same type of concerns (independently of async considerations), so the closure you pass it must meet a 'static bound too. In contrast, scope allows spawning threads that do borrow locals (and thus the closures do not have to be 'static), because scope doesn't return until all the threads have stopped executing.

Send is a type-system level assertion that says "it is always sound to send this values of this type to another thread". It has no methods or runtime effect; it exists so we can have compile-time guarantees that we're not going to end up with a data races and the like in multi-threaded programs. The compiler doesn't intrinsically know what it means; APIs that are going to send things to another thread (like spawn) have Send bounds on their generics in order to participate in this compile-time soundness guarantee system.

T: Sync means &T is Send.

Here's my favorite exploration of the topic. It has some concrete examples of types which are and are not Send/Sync and why.

The shorter answer is that it doesn't do anything like silently adding a Mutex somewhere. It changes some types and some type level guarantees which are checked at compile time. I think the only places you can add + Send to types are dyn Trait and -> impl Trait, so let's look at each of those more closely.

These are four distinct types:

dyn Trait               // Erased type implements `Trait`
dyn Trait + Send        // Erased type implements `Trait` and `Send`
dyn Trait + Sync        // Erased type implements `Trait` and `Sync`
dyn Trait + Send + Sync // Erased type implements `Trait`, `Send`, and `Sync`

And generally speaking, dyn Trait + Send will implement Send, but dyn Trait will not. So in this case, when you add + Send, you're choosing a different type, like choosing Arc<str> instead of Rc<str> or such.

There's no extra Mutex or such introduced by adding + Send. There's a difference in what types can coerce to dyn Trait + ... (you have to meet all the bounds), and there's a difference in what traits are implemented automatically. dyn Trait + Send can implement Send automatically because you cannot coerce a value to dyn Trait + Send unless the value's type is Send. It's all part of the "guaranteeing soundness at compile time using the type system" approach.

When you add + Send to a -> impl Trait type...

fn f0<T>() -> impl Trait { ... }
fn f1<T>() -> impl Trait + Send { ... }
fn f2<T>() -> impl Trait + Sync { ... }
fn f3<T>() -> impl Trait + Send + Sync { ... }

you're guaranteeing to the caller that the hidden return type always implements Send (for f1 and f3). (There's no change in the type; no implicit Mutex etc.)

If you don't make that guarantee, it may still be the case that the hidden return type always implements Send. Or it may be the case that it never implements Send. Or it may be the case that it conditionally implements Send based on what the generic T type ends up being.

Whether you make the guarantee or not, if the hidden type implements Send (or any other auto-trait), the caller is allowed to utilize that (e.g. by passing returned values to spawn). This is sort of like how if you don't explicitly implement Send for your struct, users of the struct can still make use of its automatic Send implementation (assuming it has one).[3]

A big reason it works this way -- auto traits of the hidden type leaking when there is no guarantee -- is that the conditional case is very common, but we have no syntax to convey a conditional guarantee. Like from earlier in this topic I mentioned a Shared<T> which is Send if (and only if) T is Send.[4] So you might have something like...

// Can't use `-> impl Trait + Send` here because whether or not
// the return type is `Send` is conditional on `T`.
fn g0<T>(t: T) -> impl Trait { Shared(t) }

// Works, but now we've limited which `T` you can pass in
// (which might be fine sometimes, but not always).
fn g1<T: Send>(t: T) -> impl Trait + Send { Shared(t) }

// More ideal, but doesn't exist in the language.
fn we_wish<T>(t: T) -> impl Trait + if<T: Send then Send> { Shared(t) }

Because the actual type of each -> impl Trait is hidden from the caller, the return type of g0 and g1 act like distinct types (e.g. you couldn't put them both in the same Vec<_>[5]), so providing both functions isn't a complete workaround for the "conditionally Send" scenario.

And also I suppose, the conditions aren't always so straight-forward, it would be a lot of boilerplate to cover all the auto-trait combinations, etc. Downsides which do sometimes come up with dyn Trait + ..., where there is no leakage.

(A downside of the leakage is that it's easier to accidentally make a breaking change by returning a different type that doesn't implement Send in the same cases -- similar to structs without explicit auto trait implementations.)

  1. assuming 'a does not equal 'static ↩︎

  2. at least as things are now, i.e., without significant changes to the type system ↩︎

  3. Part of the reason for -> impl Trait is so that you can change the actual type returned, so it's not exactly the same, but anyway. ↩︎

  4. This is almost always how things end up working for some GenericStruct<T>, hence why the conditional case is so common. ↩︎

  5. without type erasure ↩︎

10

I tackled dyn Trait + Send and -> impl Trait + Send separately, but I guess there is a correlation in some sense -- if we consider the erased type to be an input to some

// Can't return `+ Send` because it's conditional on `T`
fn i0<T: Trait>(t: &T) -> &impl Trait { t }
fn d0<T: Trait>(t: &T) -> &dyn Trait { t }

// Can return `+ Send` but we've limited the input types
fn i1<T: Trait + Send>(t: &T) -> &(impl Trait + Send) { t }
fn d1<T: Trait + Send>(t: &T) -> &(dyn Trait + Send) { t }

A key difference being that i0's return type will still satisfy Send when the input T was Send due to auto-trait leakage, but d0's return type does not (and cannot, at the type level).

(Every distinct input T to i0 results in a distinct output type; the output type is parameterized by T. But dyn Trait is not parameterized by the erased type T.)