Skip to content

Commit

Permalink
Add multi-producer, multi-consumer channel (mpmc)
Browse files Browse the repository at this point in the history
  • Loading branch information
obeis committed Sep 30, 2024
1 parent 0245b0c commit 04721ba
Show file tree
Hide file tree
Showing 8 changed files with 1,759 additions and 53 deletions.
4 changes: 3 additions & 1 deletion library/std/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@
//! the [`io`], [`fs`], and [`net`] modules.
//!
//! The [`thread`] module contains Rust's threading abstractions. [`sync`]
//! contains further primitive shared memory types, including [`atomic`] and
//! contains further primitive shared memory types, including [`atomic`], [`mpmc`] and
//! [`mpsc`], which contains the channel types for message passing.
//!
//! # Use before and after `main()`
Expand All @@ -177,6 +177,7 @@
//! - after-main use of thread-locals, which also affects additional features:
//! - [`thread::current()`]
//! - [`thread::scope()`]
//! - [`sync::mpmc`]
//! - [`sync::mpsc`]
//! - before-main stdio file descriptors are not guaranteed to be open on unix platforms
//!
Expand All @@ -202,6 +203,7 @@
//! [`atomic`]: sync::atomic
//! [`for`]: ../book/ch03-05-control-flow.html#looping-through-a-collection-with-for
//! [`str`]: prim@str
//! [`mpmc`]: sync::mpmc
//! [`mpsc`]: sync::mpsc
//! [`std::cmp`]: cmp
//! [`std::slice`]: mod@slice
Expand Down
11 changes: 9 additions & 2 deletions library/std/src/sync/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,11 @@
//! inter-thread synchronisation mechanism, at the cost of some
//! extra memory.
//!
//! - [`mpmc`]: Multi-producer, multi-consumer queues, used for
//! message-based communication. Can provide a lightweight
//! inter-thread synchronisation mechanism, at the cost of some
//! extra memory.
//!
//! - [`Mutex`]: Mutual Exclusion mechanism, which ensures that at
//! most one thread at a time is able to access some data.
//!
Expand All @@ -153,6 +158,7 @@
//! [`Arc`]: crate::sync::Arc
//! [`Barrier`]: crate::sync::Barrier
//! [`Condvar`]: crate::sync::Condvar
//! [`mpmc`]: crate::sync::mpmc
//! [`mpsc`]: crate::sync::mpsc
//! [`Mutex`]: crate::sync::Mutex
//! [`Once`]: crate::sync::Once
Expand Down Expand Up @@ -184,7 +190,7 @@ pub use self::mutex::{Mutex, MutexGuard};
pub use self::once::{ONCE_INIT, Once, OnceState};
#[stable(feature = "once_cell", since = "1.70.0")]
pub use self::once_lock::OnceLock;
#[stable(feature = "rust1", since = "1.0.0")]
#[stable(feature = "once_cell", since = "1.70.0")]
pub use self::poison::{LockResult, PoisonError, TryLockError, TryLockResult};
#[unstable(feature = "reentrant_lock", issue = "121440")]
pub use self::reentrant_lock::{ReentrantLock, ReentrantLockGuard};
Expand All @@ -193,12 +199,13 @@ pub use self::rwlock::{MappedRwLockReadGuard, MappedRwLockWriteGuard};
#[stable(feature = "rust1", since = "1.0.0")]
pub use self::rwlock::{RwLock, RwLockReadGuard, RwLockWriteGuard};

#[unstable(feature = "mpmc_channel", issue = "126840")]
pub mod mpmc;
pub mod mpsc;

mod barrier;
mod condvar;
mod lazy_lock;
mod mpmc;
mod mutex;
pub(crate) mod once;
mod once_lock;
Expand Down
5 changes: 5 additions & 0 deletions library/std/src/sync/mpmc/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ use crate::{error, fmt};
///
/// [`send_timeout`]: super::Sender::send_timeout
#[derive(PartialEq, Eq, Clone, Copy)]
#[unstable(feature = "mpmc_channel", issue = "126840")]
pub enum SendTimeoutError<T> {
/// The message could not be sent because the channel is full and the operation timed out.
///
Expand All @@ -18,12 +19,14 @@ pub enum SendTimeoutError<T> {
Disconnected(T),
}

#[unstable(feature = "mpmc_channel", issue = "126840")]
impl<T> fmt::Debug for SendTimeoutError<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
"SendTimeoutError(..)".fmt(f)
}
}

#[unstable(feature = "mpmc_channel", issue = "126840")]
impl<T> fmt::Display for SendTimeoutError<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
Expand All @@ -33,8 +36,10 @@ impl<T> fmt::Display for SendTimeoutError<T> {
}
}

#[unstable(feature = "mpmc_channel", issue = "126840")]
impl<T> error::Error for SendTimeoutError<T> {}

#[unstable(feature = "mpmc_channel", issue = "126840")]
impl<T> From<SendError<T>> for SendTimeoutError<T> {
fn from(err: SendError<T>) -> SendTimeoutError<T> {
match err {
Expand Down
Loading

0 comments on commit 04721ba

Please sign in to comment.