oneshot Channel

[connortsui20]

Tracking Issue: #143674

This PR adds an experimental oneshot module.

Before talking about the API itself, I would prefer to get some of these questions below out of the way first. And as discussed in the ACP it would be

Unresolved Questions

• Why exactly is it okay for Sender to be Sync? Or basically, how do we boil down the discussion in Implement Sync for mpsc::Sender #111087 into a comment for the unsafe impl<T: Send> Sync for Sender<T> {}?
• Why is mpsc::Receiver !Sync but mpmc::Receiver is Sync? Should oneshot::Receiver be Sync or not?
• Should this PR try to add an is_ready method as proposed in the tracking issue? If so, then the surface of this PR would likely need to increase to add a pub(crate) fn is_disconnected method to mpmc (might even be a good idea to add that to all 3 channel flavors).
• In a similar vein to the previous question, should the first internal implementation simply be a wrapper around mpmc, or should it be a wrapper around the internal crossbeam implementation?
• Should the Sender and Receiver operations be methods or associated methods? So sender.send(msg) or Sender::send(sender, msg)? The method syntax is more consistent with the rest of the ecosystem (namely tokio)

[rustbot]

r? @tgross35

rustbot has assigned @tgross35.
They will have a look at your PR within the next two weeks and either review your PR or reassign to another reviewer.

Use r? to explicitly pick a reviewer

[tgross35]

Maybe want to ask the unresolved questions at the ACP thread? To make sure it's seen by those already involved.

(I'll take a closer look later)

[RustyYato]

Why exactly is it okay for Sender to be Sync? Or basically, how do we boil down the discussion in #111087 into a comment for the unsafe impl<T: Send> Sync for Sender {}?

Why is mpsc::Receiver !Sync but mpmc::Receiver is Sync? Should oneshot::Receiver be Sync or not?

Because mpsc is multiple producer single consumer, and does the receiver operation via &self, so it would be incorrect to allow recv to be shared across threads because that would allow two threads to recv at the same time, which isn't allowed for a mpsc.

But mpmc is multiple producer multiple consumer, so it doesn't matter how many threads recv at the same time.

For this one, they can both be unconditionally Sync, since there is no API to go from &Sender<T> or &Reciever<T> to &T. (even via traits like Debug or Clone). In fact it looks like &Sender<T> and &Receiver<T> are entirely useless (only a Debug impl, which doesn't provide any information), which is just like Exclusive.. This is more evidence that both Sender<T> and Receiver<T> can be unconditionally Sync

[connortsui20]

Thank you for the answer!

Please let me know if I'm interpreting this right: If all methods where synchronization must occur consume self, we can say that both Sender and Receiver are Sync. For example, if threads share a &oneshot::Receiver, then none of them can call recv anyways. And this would still be compatible with the proposed is_ready method discussed in the tracking issue.

[tgross35]

I haven't gotten a chance to look into this unfortunately

r? libs

[connortsui20]

r? libs (based on this comment)

[bors]

☔ The latest upstream changes (presumably #147928) made this pull request unmergeable. Please resolve the merge conflicts.

[tgross35]

It's been a little while so
r? @joboet

Feel free to reroll if you want but you've reviewed a lot of sync module things.

[JonathanBrouwer]

@bors r-
#150537 (comment)

[tgross35]

^ also the last commit here should be squashed

[connortsui20]

Does anyone have an idea of why the test fails (under Miri)?

The issue is a doctest failure, this is the example:

#![feature(oneshot_channel)]
use std::sync::oneshot::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;

let (sender, receiver) = oneshot::channel();

thread::spawn(move || {
    // Simulate a long computation that takes longer than our timeout.
    thread::sleep(Duration::from_millis(500));
    sender.send("Too late!".to_string()).unwrap();
});

// Try to receive the message with a short timeout.
match receiver.recv_timeout(Duration::from_millis(100)) {
    Ok(msg) => println!("Received: {}", msg),
    Err(RecvTimeoutError::Timeout(rx)) => {
        println!("Timed out waiting for message!");
        // You can reuse the receiver if needed.
        drop(rx);
    }
    Err(RecvTimeoutError::Disconnected) => println!("Sender dropped!"),
}

This is the error (taken from here):

---- library/std/src/sync/oneshot.rs - sync::oneshot::RecvTimeoutError (line 358) stdout ----
Test executable failed (exit status: 1).
stdout:
Timed out waiting for message!
stderr:
error: the main thread terminated without waiting for all remaining threads
note: set `MIRIFLAGS=-Zmiri-ignore-leaks` to disable this check
error: aborting due to 1 previous error

As far as I can tell, there are plenty of other doc tests similar to this that do not join the spawned thread and run perfectly fine under Miri. Is there some sort of limit to how long a thread can sleep for in Miri doc tests?

[tgross35]

All of our tests should probably be joining correctly: it is technically a detach with resource leak https://doc.rust-lang.org/std/thread/fn.spawn.html, and I know this error shows up occasionally. Not sure why Miri only flags this sometimes but I'm sure @RalfJung does.

[connortsui20]

We could probably get away with assigning the thread handle to a variable in a # line in the doc comment and then join that later? And maybe we can do that for every doctest in a separate PR

[RalfJung]

When the main thread is done, Miri yields a couple times to give other threads the chance to finish... but with such a huge timeout, that's not enough. This is why it works for other tests (where the background threads can pretty much finish immediately when the main thread is done) but not for this one (where the background thread can take 0.4s to finish).

So it's not necessary to do that for all doctests, but it'd be better to do that for those tests where the background thread is waiting on a timeout that's bigger than the time it takes the main thread to finish.

[connortsui20]

For reference I just changed the doctest to this:

#![feature(oneshot_channel)]
use std::sync::oneshot::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;

let (sender, receiver) = oneshot::channel();

let send_failure = thread::spawn(move || {
    // Simulate a long computation that takes longer than our timeout.
    thread::sleep(Duration::from_millis(250));

    // This will likely fail to send because we drop the receiver in the main thread.
    sender.send("Goodbye!".to_string()).unwrap();
});

// Try to receive the message with a short timeout.
match receiver.recv_timeout(Duration::from_millis(10)) {
    Ok(msg) => println!("Received: {}", msg),
    Err(RecvTimeoutError::Timeout(rx)) => {
        println!("Timed out waiting for message!");

        // Note that you can reuse the receiver without dropping it.
        drop(rx);
    },
    Err(RecvTimeoutError::Disconnected) => println!("Sender dropped!"),
}

send_failure.join().unwrap_err();

@rustbot ready

[joboet]

Somethings gone wrong here, this issue is back...

[connortsui20]

argh I accidentally dropped the commit instead of squashing it... Thank you for catching that

[rustbot]

This PR was rebased onto a different main commit. Here's a range-diff highlighting what actually changed.

Rebasing is a normal part of keeping PRs up to date, so no action is needed—this note is just to help reviewers.

[connortsui20]

@rustbot ready