1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
//! # Deadqueue [](https://crates.io/crates/deadqueue) [](https://travis-ci.org/deadpool-rs/deadqueue)
//!
//! Deadqueue is a dead simple async queue with back pressure support.
//!
//! This crate provides three implementations:
//!
//! - Unlimited (`deadqueue::unlimited::Queue`)
//! - Based on `crossbeam_queue::SegQueue`
//! - Has unlimitied capacity and no back pressure on push
//! - Enabled via the `unlimited` feature in your `Cargo.toml`
//!
//! - Resizable (`deadqueue::resizable::Queue`)
//! - Based on `deadqueue::unlimited::Queue`
//! - Has limited capacity with back pressure on push
//! - Supports resizing
//! - Enabled via the `resizable` feature in your `Cargo.toml`
//!
//! - Limited (`deadqueue::limited::Queue`)
//! - Based on `crossbeam_queue::ArrayQueue`
//! - Has limit capacity with back pressure on push
//! - Does not support resizing
//! - Enabled via the `limited` feature in your `Cargo.toml`
//!
//! ## Features
//!
//! | Feature | Description | Extra dependencies | Default |
//! | ------- | ----------- | ------------------ | ------- |
//! | `unlimited` | Enable unlimited queue implementation | – | yes |
//! | `resizable` | Enable resizable queue implementation | `deadqueue/unlimited` | yes |
//! | `limited` | Enable limited queue implementation | – | yes |
//!
//! ## Example
//!
//! ```rust
//! use std::sync::Arc;
//! use tokio::time::{sleep, Duration};
//!
//! const TASK_COUNT: usize = 1000;
//! const WORKER_COUNT: usize = 10;
//!
//! type TaskQueue = deadqueue::limited::Queue<usize>;
//!
//! #[tokio::main]
//! async fn main() {
//! let queue = Arc::new(TaskQueue::new(TASK_COUNT));
//! for i in 0..TASK_COUNT {
//! queue.try_push(i).unwrap();
//! }
//! for worker in 0..WORKER_COUNT {
//! let queue = queue.clone();
//! tokio::spawn(async move {
//! loop {
//! let task = queue.pop().await;
//! println!("worker[{}] processing task[{}] ...", worker, task);
//! }
//! });
//! }
//! println!("Waiting for workers to finish...");
//! queue.wait_empty().await;
//! println!("All tasks done. :-)");
//! }
//! ```
//!
//! ## Reasons for yet another queue
//!
//! Deadqueue is by no means the only queue implementation available. It does things a little different and provides features that other implementations are lacking:
//!
//! - **Resizable queue.** Usually you have to pick between `limited` and `unlimited` queues. This crate features a `resizable` Queue which can be resized as needed. This is probably a big **unique selling point** of this crate.
//!
//! - **Introspection support.** The methods `.len()`, `.capacity()` and `.available()` provide access the current state of the queue.
//!
//! - **Fair scheduling.** Tasks calling `pop` will receive items in a first-come-first-serve fashion. This is mainly due to the use of `tokio::sync::Semaphore` which is fair by nature.
//!
//! - **One struct, not two.** The channels of `tokio`, `async_std` and `futures-intrusive` split the queue in two structs (`Sender` and `Receiver`) which makes the usage sligthly more complicated.
//!
//! - **Bring your own `Arc`.** Since there is no separation between `Sender` and `Receiver` there is also no need for an internal `Arc`. (All implementations that split the channel into a `Sender` and `Receiver` need some kind of `Arc` internally.)
//!
//! - **Fully concurrent access.** No need to wrap the `Receiver` part in a `Mutex`. All methods support concurrent accesswithout the need for an additional synchronization primitive.
//!
//! - **Support for `try__` methods.** The methods `try_push` and `try_pop` can be used to access the queue from non-blocking synchroneous code.
//!
//! - **Support for detecting when the queue becomes empty or full**, using the `wait_empty`, `subscribe_empty`, `wait_full` and `subscribe_full` methods.
//!
//! ## Alternatives
//!
//! | Crate | Limitations | Documentation |
//! | --- | --- | --- |
//! | [`tokio`](https://crates.io/crates/tokio) | No resizable queue. No introspection support. Synchronization of `Receiver` needed. | [`tokio::sync::mpsc::channel`](https://docs.rs/tokio/latest/tokio/sync/mpsc/fn.channel.html), [`tokio::sync::mpsc::unbounded_channel`](https://docs.rs/tokio/latest/tokio/sync/mpsc/fn.unbounded_channel.html) |
//! | [`async-std`](https://crates.io/crates/async-std) | No resizable or unlimited queue. No introspection support. No `try_send` or `try_recv` methods. | [`async_std::sync::channel`](https://docs.rs/async-std/latest/async_std/sync/fn.channel.html) |
//! | [`futures`](https://crates.io/crates/futures) | No resizable queue. No introspection support. | [`futures::channel::mpsc::channel`](https://docs.rs/futures/0.3.1/futures/channel/mpsc/fn.channel.html), [`futures::channel::mpsc::unbounded`](https://docs.rs/futures/0.3.1/futures/channel/mpsc/fn.unbounded.html) |
//!
//! ## License
//!
//! Licensed under either of
//!
//! - Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or <http://www.apache.org/licenses/LICENSE-2.0)>
//! - MIT license ([LICENSE-MIT](LICENSE-MIT) or <http://opensource.org/licenses/MIT)>
//!
//! at your option.
use watch;
/// Private type alias for notify_full and notify_empty
type Notifier = Sender;
/// Public type alias for subscribe_full and subscribe_empty
pub type Receiver = Receiver;
/// Initialize the notify_full sender