Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

sync: add more oneshot examples #4153

Merged
merged 3 commits into from Oct 8, 2021
Merged

sync: add more oneshot examples #4153

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a060bd5
sync: add more oneshot examples
Darksonn Oct 5, 2021
14cdcc1
Update examples
Darksonn Oct 7, 2021
cee4fad
Fix test
Darksonn Oct 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
179 changes: 179 additions & 0 deletions tokio/src/sync/oneshot.rs
View file
Open in desktop
Expand Up @@ -51,6 +51,70 @@
//! }
//! }
//! ```
//!
//! To use a oneshot channel in a `tokio::select!` loop, add `&mut` in front of
//! the channel.
//!
//! ```
//! use tokio::sync::oneshot;
//! use tokio::time::{interval, sleep, Duration};
//!
//! #[tokio::main]
//! # async fn _doc() {}
//! # #[tokio::main(flavor = "current_thread", start_paused = true)]
//! async fn main() {
//! let (send, mut recv) = oneshot::channel();
//! let mut interval = interval(Duration::from_millis(100));
//!
//! # let handle =
//! tokio::spawn(async move {
//! sleep(Duration::from_secs(1)).await;
//! send.send("shut down").unwrap();
//! });
//!
//! loop {
//! tokio::select! {
//! _ = interval.tick() => println!("Another 100ms"),
//! msg = &mut recv => {
//! println!("Got message: {}", msg.unwrap());
//! break;
//! }
//! }
//! }
//! # handle.await.unwrap();
//! }
//! ```
//!
//! To use a `Sender` from a destructor, put it in an [`Option`] and call
//! [`Option::take`].
//!
//! ```
//! use tokio::sync::oneshot;
//!
//! struct SendOnDrop {
//! sender: Option<oneshot::Sender<&'static str>>,
//! }
//! impl Drop for SendOnDrop {
//! fn drop(&mut self) {
//! if let Some(sender) = self.sender.take() {
//! // Using `let _ =` to ignore send errors.
//! let _ = sender.send("I got dropped!");
//! }
//! }
//! }
//!
//! #[tokio::main]
//! # async fn _doc() {}
//! # #[tokio::main(flavor = "current_thread")]
//! async fn main() {
//! let (send, recv) = oneshot::channel();
//!
//! let send_on_drop = SendOnDrop { sender: Some(send) };
//! drop(send_on_drop);
//!
//! assert_eq!(recv.await, Ok("I got dropped!"));
//! }
//! ```

use crate::loom::cell::UnsafeCell;
use crate::loom::sync::atomic::AtomicUsize;
Expand All @@ -68,6 +132,83 @@ use std::task::{Context, Poll, Waker};
///
/// A pair of both a [`Sender`] and a [`Receiver`] are created by the
/// [`channel`](fn@channel) function.
///
/// # Examples
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, rx) = oneshot::channel();
///
/// tokio::spawn(async move {
/// if let Err(_) = tx.send(3) {
/// println!("the receiver dropped");
/// }
/// });
///
/// match rx.await {
/// Ok(v) => println!("got = {:?}", v),
/// Err(_) => println!("the sender dropped"),
/// }
/// }
/// ```
///
/// If the sender is dropped without sending, the receiver will fail with
/// [`error::RecvError`]:
///
/// ```
/// use tokio::sync::oneshot;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx, rx) = oneshot::channel::<u32>();
///
/// tokio::spawn(async move {
/// drop(tx);
/// });
///
/// match rx.await {
/// Ok(_) => panic!("This doesn't happen"),
/// Err(_) => println!("the sender dropped"),
/// }
/// }
/// ```
///
/// To use a `Sender` from a destructor, put it in an [`Option`] and call
/// [`Option::take`].
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can I ask why it's recommended to wrap Sender in an Option here? I can't seem to find a rationale!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you try removing the option, it should quickly become clear: it wont compile without it. Calling send on an oneshot channel destroys the sender.

///
/// ```
/// use tokio::sync::oneshot;
///
/// struct SendOnDrop {
/// sender: Option<oneshot::Sender<&'static str>>,
/// }
/// impl Drop for SendOnDrop {
/// fn drop(&mut self) {
/// if let Some(sender) = self.sender.take() {
/// // Using `let _ =` to ignore send errors.
/// let _ = sender.send("I got dropped!");
/// }
/// }
/// }
///
/// #[tokio::main]
/// # async fn _doc() {}
/// # #[tokio::main(flavor = "current_thread")]
/// async fn main() {
/// let (send, recv) = oneshot::channel();
///
/// let send_on_drop = SendOnDrop { sender: Some(send) };
/// drop(send_on_drop);
///
/// assert_eq!(recv.await, Ok("I got dropped!"));
/// }
/// ```
///
/// [`Option`]: std::option::Option
/// [`Option::take`]: std::option::Option::take
#[derive(Debug)]
pub struct Sender<T> {
inner: Option<Arc<Inner<T>>>,
Expand All @@ -78,6 +219,11 @@ pub struct Sender<T> {
/// A pair of both a [`Sender`] and a [`Receiver`] are created by the
/// [`channel`](fn@channel) function.
///
/// This channel has no `recv` method because the receiver itself implements the
/// [`Future`] trait. To receive a value, `.await` the `Receiver` object directly.
///
/// [`Future`]: trait@std::future::Future
///
/// # Examples
///
/// ```
Expand Down Expand Up @@ -120,6 +266,39 @@ pub struct Sender<T> {
/// }
/// }
/// ```
///
/// To use a `Receiver` in a `tokio::select!` loop, add `&mut` in front of the
/// channel.
///
/// ```
/// use tokio::sync::oneshot;
/// use tokio::time::{interval, sleep, Duration};
///
/// #[tokio::main]
/// # async fn _doc() {}
/// # #[tokio::main(flavor = "current_thread", start_paused = true)]
/// async fn main() {
/// let (send, mut recv) = oneshot::channel();
/// let mut interval = interval(Duration::from_millis(100));
///
/// # let handle =
/// tokio::spawn(async move {
/// sleep(Duration::from_secs(1)).await;
/// send.send("shut down").unwrap();
/// });
///
/// loop {
/// tokio::select! {
/// _ = interval.tick() => println!("Another 100ms"),
/// msg = &mut recv => {
/// println!("Got message: {}", msg.unwrap());
/// break;
/// }
/// }
/// }
/// # handle.await.unwrap();
/// }
/// ```
#[derive(Debug)]
pub struct Receiver<T> {
inner: Option<Arc<Inner<T>>>,
Expand Down