forked from rust-lang/cargo
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Replace
std::sync::mpsc with a much simpler queue
We don't need the complexity of most channels since this is not a performance sensitive part of Cargo, nor is it likely to be so any time soon. Coupled with recent bugs (rust-lang#7840) we believe in `std::sync::mpsc`, let's just not use that and use a custom queue type locally which should be amenable to a blocking push soon too.
- Loading branch information
1 parent
43aafb4
commit 458138b
Showing
4 changed files
with
91 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| use std::collections::VecDeque; | ||
| use std::sync::{Condvar, Mutex}; | ||
| use std::time::{Duration, Instant}; | ||
|
|
||
| /// A simple, threadsafe, queue of items of type `T` | ||
| /// | ||
| /// This is a sort of channel where any thread can push to a queue and any | ||
| /// thread can pop from a queue. Currently queues have infinite capacity where | ||
| /// `push` will never block but `pop` will block. | ||
| pub struct Queue<T> { | ||
| state: Mutex<State<T>>, | ||
| condvar: Condvar, | ||
| } | ||
|
|
||
| struct State<T> { | ||
| items: VecDeque<T>, | ||
| } | ||
|
|
||
| impl<T> Queue<T> { | ||
| pub fn new() -> Queue<T> { | ||
| Queue { | ||
| state: Mutex::new(State { | ||
| items: VecDeque::new(), | ||
| }), | ||
| condvar: Condvar::new(), | ||
| } | ||
| } | ||
|
|
||
| pub fn push(&self, item: T) { | ||
| self.state.lock().unwrap().items.push_back(item); | ||
| self.condvar.notify_one(); | ||
| } | ||
|
|
||
| pub fn pop(&self, timeout: Duration) -> Option<T> { | ||
| let mut state = self.state.lock().unwrap(); | ||
| let now = Instant::now(); | ||
| while state.items.is_empty() { | ||
| let elapsed = now.elapsed(); | ||
| if elapsed >= timeout { | ||
| break; | ||
| } | ||
| let (lock, result) = self.condvar.wait_timeout(state, timeout - elapsed).unwrap(); | ||
| state = lock; | ||
| if result.timed_out() { | ||
| break; | ||
| } | ||
| } | ||
| state.items.pop_front() | ||
| } | ||
|
|
||
| pub fn try_pop(&self) -> Option<T> { | ||
| self.state.lock().unwrap().items.pop_front() | ||
| } | ||
| } |