eater
/
async-std
forked from mirror/async-std
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' into stream-cloned

Browse Source
poc-serde-support
yjh 5 years ago committed by GitHub
parent 8bef812e78 20cdf73bb0
commit e85bbe68e6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 262 additions and 371 deletions
Show Stats Download Patch File Download Diff File

6
src/option/sum.rs
Unescape Escape View File

@ -20,12 +20,10 @@ where
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let words: VecDeque<_> = vec!["have", "a", "great", "day"]
.into_iter()
.collect();
let words = stream::from_iter(vec!["have", "a", "great", "day"]);
let total: Option<usize> = words.map(|w| w.find('a')).sum().await;
assert_eq!(total, Some(5));
#

4
src/result/product.rs
Unescape Escape View File

@ -20,10 +20,10 @@ where
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let v: VecDeque<_> = vec![1, 2, 4].into_iter().collect();
let v = stream::from_iter(vec![1, 2, 4]);
let res: Result<i32, &'static str> = v.map(|x|
if x < 0 {
Err("Negative element found")

4
src/result/sum.rs
Unescape Escape View File

@ -20,10 +20,10 @@ where
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let v: VecDeque<_> = vec![1, 2].into_iter().collect();
let v = stream::from_iter(vec![1, 2]);
let res: Result<i32, &'static str> = v.map(|x|
if x < 0 {
Err("Negative element found")

2
src/stream/from_iter.rs
Unescape Escape View File

@ -12,7 +12,7 @@ pin_project! {
/// See it documentation for more.
///
/// [`from_iter`]: fn.from_iter.html
#[derive(Debug)]
#[derive(Clone, Debug)]
pub struct FromIter<I> {
iter: I,
}

279
src/stream/stream/mod.rs
Unescape Escape View File

@ -282,11 +282,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3, 4].into_iter().collect();
let s = stream::from_iter(vec![1, 2, 3, 4]);
let mut s = s.take_while(|x| x < &3 );
assert_eq!(s.next().await, Some(1));
@ -319,9 +318,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s: VecDeque<_> = vec![0u8, 1, 2, 3, 4].into_iter().collect();
let s = stream::from_iter(vec![0u8, 1, 2, 3, 4]);
let mut stepped = s.step_by(2);
assert_eq!(stepped.next().await, Some(0));
@ -351,10 +350,10 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let first: VecDeque<_> = vec![0u8, 1].into_iter().collect();
let second: VecDeque<_> = vec![2, 3].into_iter().collect();
let first = stream::from_iter(vec![0u8, 1]);
let second = stream::from_iter(vec![2, 3]);
let mut c = first.chain(second);
assert_eq!(c.next().await, Some(0));
@ -421,17 +420,15 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
let v: VecDeque<_> = vec![&1, &2, &3].into_iter().collect();
let mut v_copied = v.copied();
use async_std::stream;
assert_eq!(v_copied.next().await, Some(1));
assert_eq!(v_copied.next().await, Some(2));
assert_eq!(v_copied.next().await, Some(3));
assert_eq!(v_copied.next().await, None);
let s = stream::from_iter(vec![&1, &2, &3]);
let mut s_copied = s.copied();
assert_eq!(s_copied.next().await, Some(1));
assert_eq!(s_copied.next().await, Some(2));
assert_eq!(s_copied.next().await, Some(3));
assert_eq!(s_copied.next().await, None);
#
# }) }
```
@ -457,9 +454,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s: VecDeque<_> = vec!['a', 'b', 'c'].into_iter().collect();
let s = stream::from_iter(vec!['a', 'b', 'c']);
let mut s = s.enumerate();
assert_eq!(s.next().await, Some((0, 'a')));
@ -487,9 +484,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s: VecDeque<_> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1, 2, 3]);
let mut s = s.map(|x| 2 * x);
assert_eq!(s.next().await, Some(2));
@ -521,10 +518,11 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s = stream::from_iter(vec![1, 2, 3, 4, 5]);
let a: VecDeque<_> = vec![1u8, 2, 3, 4, 5].into_iter().collect();
let sum = a
let sum = s
.inspect(|x| println!("about to filter {}", x))
.filter(|x| x % 2 == 0)
.inspect(|x| println!("made it through filter: {}", x))
@ -553,11 +551,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1, 2, 3]);
let last = s.last().await;
assert_eq!(last, Some(3));
@ -569,11 +566,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
use crate::async_std::prelude::*;
let s: VecDeque<usize> = vec![].into_iter().collect();
let s = stream::empty::<()>();
let last = s.last().await;
assert_eq!(last, None);
@ -634,11 +630,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3, 4].into_iter().collect();
let s = stream::from_iter(vec![1, 2, 3, 4]);
let mut s = s.filter(|i| i % 2 == 0);
assert_eq!(s.next().await, Some(2));
@ -666,14 +661,14 @@ extension_trait! {
```
# async_std::task::block_on(async {
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream::IntoStream;
use async_std::stream;
let inner1: VecDeque<u8> = vec![1,2,3].into_iter().collect();
let inner2: VecDeque<u8> = vec![4,5,6].into_iter().collect();
let inner1 = stream::from_iter(vec![1,2,3]);
let inner2 = stream::from_iter(vec![4,5,6]);
let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
let s = stream::from_iter(vec![inner1, inner2]);
let v :Vec<_> = s.flat_map(|s| s.into_stream()).collect().await;
@ -703,12 +698,12 @@ extension_trait! {
```
# async_std::task::block_on(async {
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let inner1: VecDeque<u8> = vec![1,2,3].into_iter().collect();
let inner2: VecDeque<u8> = vec![4,5,6].into_iter().collect();
let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
let inner1 = stream::from_iter(vec![1u8,2,3]);
let inner2 = stream::from_iter(vec![4u8,5,6]);
let s = stream::from_iter(vec![inner1, inner2]);
let v: Vec<_> = s.flatten().collect().await;
@ -736,11 +731,11 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<&str> = vec!["1", "lol", "3", "NaN", "5"].into_iter().collect();
let s = stream::from_iter(vec!["1", "lol", "3", "NaN", "5"]);
let mut parsed = s.filter_map(|a| a.parse::<u32>().ok());
@ -777,16 +772,15 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<i32> = vec![1, 2, -3].into_iter().collect();
let s = stream::from_iter(vec![1isize, 2, -3]);
let min = s.clone().min_by_key(|x| x.abs()).await;
assert_eq!(min, Some(1));
let min = VecDeque::<isize>::new().min_by_key(|x| x.abs()).await;
let min = stream::empty::<isize>().min_by_key(|x| x.abs()).await;
assert_eq!(min, None);
#
# }) }
@ -814,16 +808,15 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<i32> = vec![-1, -2, -3].into_iter().collect();
let s = stream::from_iter(vec![-1isize, -2, -3]);
let max = s.clone().max_by_key(|x| x.abs()).await;
assert_eq!(max, Some(3));
let max = VecDeque::<isize>::new().max_by_key(|x| x.abs()).await;
let max = stream::empty::<isize>().min_by_key(|x| x.abs()).await;
assert_eq!(max, None);
#
# }) }
@ -851,11 +844,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1u8, 2, 3]);
let min = s.clone().min_by(|x, y| x.cmp(y)).await;
assert_eq!(min, Some(1));
@ -863,7 +855,7 @@ extension_trait! {
let min = s.min_by(|x, y| y.cmp(x)).await;
assert_eq!(min, Some(3));
let min = VecDeque::<usize>::new().min_by(|x, y| x.cmp(y)).await;
let min = stream::empty::<u8>().min_by(|x, y| x.cmp(y)).await;
assert_eq!(min, None);
#
# }) }
@ -889,15 +881,15 @@ extension_trait! {
```ignore
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1usize, 2, 3]);
let min = s.clone().min().await;
assert_eq!(min, Some(1));
let min = VecDeque::<usize>::new().min().await;
let min = stream::empty::<usize>().min().await;
assert_eq!(min, None);
#
# }) }
@ -923,11 +915,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1u8, 2, 3]);
let max = s.clone().max_by(|x, y| x.cmp(y)).await;
assert_eq!(max, Some(3));
@ -935,7 +926,7 @@ extension_trait! {
let max = s.max_by(|x, y| y.cmp(x)).await;
assert_eq!(max, Some(1));
let max = VecDeque::<usize>::new().max_by(|x, y| x.cmp(y)).await;
let max = stream::empty::<usize>().max_by(|x, y| x.cmp(y)).await;
assert_eq!(max, None);
#
# }) }
@ -962,11 +953,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let mut s = stream::from_iter(vec![1u8, 2, 3]);
let second = s.nth(1).await;
assert_eq!(second, Some(2));
@ -978,11 +968,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::stream;
use async_std::prelude::*;
let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let mut s = stream::from_iter(vec![1u8, 2, 3]);
let second = s.nth(0).await;
assert_eq!(second, Some(1));
@ -996,11 +985,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let mut s = stream::from_iter(vec![1u8, 2, 3]);
let fourth = s.nth(4).await;
assert_eq!(fourth, None);
@ -1091,9 +1079,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let mut s = stream::from_iter(vec![1u8, 2, 3]);
let res = s.find(|x| *x == 2).await;
assert_eq!(res, Some(2));
#
@ -1106,9 +1094,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let mut s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let mut s= stream::from_iter(vec![1, 2, 3]);
let res = s.find(|x| *x == 2).await;
assert_eq!(res, Some(2));
@ -1136,9 +1124,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let mut s: VecDeque<&str> = vec!["lol", "NaN", "2", "5"].into_iter().collect();
let mut s = stream::from_iter(vec!["lol", "NaN", "2", "5"]);
let first_number = s.find_map(|s| s.parse().ok()).await;
assert_eq!(first_number, Some(2));
@ -1169,9 +1157,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1u8, 2, 3]);
let sum = s.fold(0, |acc, x| acc + x).await;
assert_eq!(sum, 6);
@ -1200,12 +1188,12 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
use std::sync::mpsc::channel;
let (tx, rx) = channel();
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1usize, 2, 3]);
let sum = s.for_each(move |x| tx.clone().send(x).unwrap()).await;
let v: Vec<_> = rx.iter().collect();
@ -1306,11 +1294,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<isize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1isize, 2, 3]);
let mut s = s.scan(1, |state, x| {
*state = *state * x;
Some(-*state)
@ -1347,11 +1334,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let a: VecDeque<_> = vec![-1i32, 0, 1].into_iter().collect();
let a = stream::from_iter(vec![-1i32, 0, 1]);
let mut s = a.skip_while(|x| x.is_negative());
assert_eq!(s.next().await, Some(0));
@ -1377,11 +1363,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1u8, 2, 3]);
let mut skipped = s.skip(2);
assert_eq!(skipped.next().await, Some(3));
@ -1443,9 +1428,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1usize, 2, 3]);
let sum = s.try_fold(0, |acc, v| {
if (acc+v) % 2 == 1 {
Ok(v+3)
@ -1479,13 +1464,13 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use std::sync::mpsc::channel;
use async_std::prelude::*;
use async_std::stream;
let (tx, rx) = channel();
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1u8, 2, 3]);
let s = s.try_for_each(|v| {
if v % 2 == 1 {
tx.clone().send(v).unwrap();
@ -1537,12 +1522,11 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let l: VecDeque<isize> = vec![1, 2, 3].into_iter().collect();
let r: VecDeque<isize> = vec![4, 5, 6, 7].into_iter().collect();
let l = stream::from_iter(vec![1u8, 2, 3]);
let r = stream::from_iter(vec![4u8, 5, 6, 7]);
let mut s = l.zip(r);
assert_eq!(s.next().await, Some((1, 4)));
@ -1672,14 +1656,14 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
use std::cmp::Ordering;
let s1 = VecDeque::from(vec![1]);
let s2 = VecDeque::from(vec![1, 2]);
let s3 = VecDeque::from(vec![1, 2, 3]);
let s4 = VecDeque::from(vec![1, 2, 4]);
let s1 = stream::from_iter(vec![1]);
let s2 = stream::from_iter(vec![1, 2]);
let s3 = stream::from_iter(vec![1, 2, 3]);
let s4 = stream::from_iter(vec![1, 2, 4]);
assert_eq!(s1.clone().partial_cmp(s1.clone()).await, Some(Ordering::Equal));
assert_eq!(s1.clone().partial_cmp(s2.clone()).await, Some(Ordering::Less));
assert_eq!(s2.clone().partial_cmp(s1.clone()).await, Some(Ordering::Greater));
@ -1711,9 +1695,9 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let s: VecDeque<usize> = vec![1, 2, 3].into_iter().collect();
let s = stream::from_iter(vec![1usize, 2, 3]);
let res = s.clone().position(|x| *x == 1).await;
assert_eq!(res, Some(0));
@ -1750,13 +1734,14 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
use std::cmp::Ordering;
let s1 = VecDeque::from(vec![1]);
let s2 = VecDeque::from(vec![1, 2]);
let s3 = VecDeque::from(vec![1, 2, 3]);
let s4 = VecDeque::from(vec![1, 2, 4]);
let s1 = stream::from_iter(vec![1]);
let s2 = stream::from_iter(vec![1, 2]);
let s3 = stream::from_iter(vec![1, 2, 3]);
let s4 = stream::from_iter(vec![1, 2, 4]);
assert_eq!(s1.clone().cmp(s1.clone()).await, Ordering::Equal);
assert_eq!(s1.clone().cmp(s2.clone()).await, Ordering::Less);
assert_eq!(s2.clone().cmp(s1.clone()).await, Ordering::Greater);
@ -1786,11 +1771,13 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
let single: VecDeque<isize> = vec![1].into_iter().collect();
let single_ne: VecDeque<isize> = vec![10].into_iter().collect();
let multi: VecDeque<isize> = vec![1,2].into_iter().collect();
let multi_ne: VecDeque<isize> = vec![1,5].into_iter().collect();
use async_std::stream;
let single = stream::from_iter(vec![1usize]);
let single_ne = stream::from_iter(vec![10usize]);
let multi = stream::from_iter(vec![1usize,2]);
let multi_ne = stream::from_iter(vec![1usize,5]);
assert_eq!(single.clone().ne(single.clone()).await, false);
assert_eq!(single_ne.clone().ne(single.clone()).await, true);
assert_eq!(multi.clone().ne(single_ne.clone()).await, true);
@ -1821,12 +1808,13 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let single = stream::from_iter(vec![1]);
let single_gt = stream::from_iter(vec![10]);
let multi = stream::from_iter(vec![1,2]);
let multi_gt = stream::from_iter(vec![1,5]);
let single: VecDeque<isize> = vec![1].into_iter().collect();
let single_gt: VecDeque<isize> = vec![10].into_iter().collect();
let multi: VecDeque<isize> = vec![1,2].into_iter().collect();
let multi_gt: VecDeque<isize> = vec![1,5].into_iter().collect();
assert_eq!(single.clone().ge(single.clone()).await, true);
assert_eq!(single_gt.clone().ge(single.clone()).await, true);
assert_eq!(multi.clone().ge(single_gt.clone()).await, false);
@ -1857,12 +1845,13 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let single = stream::from_iter(vec![1]);
let single_eq = stream::from_iter(vec![10]);
let multi = stream::from_iter(vec![1,2]);
let multi_eq = stream::from_iter(vec![1,5]);
let single: VecDeque<isize> = vec![1].into_iter().collect();
let single_eq: VecDeque<isize> = vec![10].into_iter().collect();
let multi: VecDeque<isize> = vec![1,2].into_iter().collect();
let multi_eq: VecDeque<isize> = vec![1,5].into_iter().collect();
assert_eq!(single.clone().eq(single.clone()).await, true);
assert_eq!(single_eq.clone().eq(single.clone()).await, false);
assert_eq!(multi.clone().eq(single_eq.clone()).await, false);
@ -1893,12 +1882,13 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let single = stream::from_iter(vec![1]);
let single_gt = stream::from_iter(vec![10]);
let multi = stream::from_iter(vec![1,2]);
let multi_gt = stream::from_iter(vec![1,5]);
let single = VecDeque::from(vec![1]);
let single_gt = VecDeque::from(vec![10]);
let multi = VecDeque::from(vec![1,2]);
let multi_gt = VecDeque::from(vec![1,5]);
assert_eq!(single.clone().gt(single.clone()).await, false);
assert_eq!(single_gt.clone().gt(single.clone()).await, true);
assert_eq!(multi.clone().gt(single_gt.clone()).await, false);
@ -1929,12 +1919,13 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let single = stream::from_iter(vec![1]);
let single_gt = stream::from_iter(vec![10]);
let multi = stream::from_iter(vec![1,2]);
let multi_gt = stream::from_iter(vec![1,5]);
let single = VecDeque::from(vec![1]);
let single_gt = VecDeque::from(vec![10]);
let multi = VecDeque::from(vec![1,2]);
let multi_gt = VecDeque::from(vec![1,5]);
assert_eq!(single.clone().le(single.clone()).await, true);
assert_eq!(single.clone().le(single_gt.clone()).await, true);
assert_eq!(multi.clone().le(single_gt.clone()).await, true);
@ -1965,12 +1956,12 @@ extension_trait! {
# fn main() { async_std::task::block_on(async {
#
use async_std::prelude::*;
use std::collections::VecDeque;
use async_std::stream;
let single = VecDeque::from(vec![1]);
let single_gt = VecDeque::from(vec![10]);
let multi = VecDeque::from(vec![1,2]);
let multi_gt = VecDeque::from(vec![1,5]);
let single = stream::from_iter(vec![1]);
let single_gt = stream::from_iter(vec![10]);
let multi = stream::from_iter(vec![1,2]);
let multi_gt = stream::from_iter(vec![1,5]);
assert_eq!(single.clone().lt(single.clone()).await, false);
assert_eq!(single.clone().lt(single_gt.clone()).await, true);
@ -2012,10 +2003,10 @@ extension_trait! {
```
# fn main() { async_std::task::block_on(async {
#
use std::collections::VecDeque;
use async_std::prelude::*;
use async_std::stream;
let s: VecDeque<_> = vec![0u8, 1, 2, 3, 4].into_iter().collect();
let s = stream::from_iter(vec![0u8, 1, 2, 3, 4]);
let sum: u8 = s.sum().await;
assert_eq!(sum, 10);

12
src/sync/channel.rs
Unescape Escape View File

@ -22,7 +22,7 @@ use crate::sync::WakerSet;
///
/// Senders and receivers can be cloned. When all senders associated with a channel get dropped, it
/// becomes closed. Receive operations on a closed and empty channel return `None` instead of
/// blocking.
/// trying to await a message.
///
/// # Panics
///
@ -44,7 +44,7 @@ use crate::sync::WakerSet;
/// s.send(1).await;
///
/// task::spawn(async move {
/// // This call blocks the current task because the channel is full.
/// // This call will have to wait because the channel is full.
/// // It will be able to complete only after the first message is received.
/// s.send(2).await;
/// });
@ -102,8 +102,7 @@ pub struct Sender<T> {
impl<T> Sender<T> {
/// Sends a message into the channel.
///
/// If the channel is full, this method will block the current task until the send operation
/// can proceed.
/// If the channel is full, this method will wait until there is space in the channel.
///
/// # Examples
///
@ -346,9 +345,8 @@ pub struct Receiver<T> {
impl<T> Receiver<T> {
/// Receives a message from the channel.
///
/// If the channel is empty and it still has senders, this method will block the current task
/// until the receive operation can proceed. If the channel is empty and there are no more
/// senders, this method returns `None`.
/// If the channel is empty and still has senders, this method will wait until a message is
/// sent into the channel or until all senders get dropped.
///
/// # Examples
///

290
src/sync/rwlock.rs
Unescape Escape View File

@ -1,26 +1,21 @@
use std::cell::UnsafeCell;
use std::fmt;
use std::isize;
use std::ops::{Deref, DerefMut};
use std::pin::Pin;
use std::process;
use std::sync::atomic::{AtomicUsize, Ordering};
use slab::Slab;
use crate::future::Future;
use crate::task::{Context, Poll, Waker};
use crate::sync::WakerSet;
use crate::task::{Context, Poll};
/// Set if a write lock is held.
#[allow(clippy::identity_op)]
const WRITE_LOCK: usize = 1 << 0;
/// Set if there are read operations blocked on the lock.
const BLOCKED_READS: usize = 1 << 1;
/// Set if there are write operations blocked on the lock.
const BLOCKED_WRITES: usize = 1 << 2;
/// The value of a single blocked read contributing to the read count.
const ONE_READ: usize = 1 << 3;
const ONE_READ: usize = 1 << 1;
/// The bits in which the read count is stored.
const READ_COUNT_MASK: usize = !(ONE_READ - 1);
@ -56,8 +51,8 @@ const READ_COUNT_MASK: usize = !(ONE_READ - 1);
/// ```
pub struct RwLock<T> {
state: AtomicUsize,
reads: std::sync::Mutex<Slab<Option<Waker>>>,
writes: std::sync::Mutex<Slab<Option<Waker>>>,
read_wakers: WakerSet,
write_wakers: WakerSet,
value: UnsafeCell<T>,
}
@ -77,8 +72,8 @@ impl<T> RwLock<T> {
pub fn new(t: T) -> RwLock<T> {
RwLock {
state: AtomicUsize::new(0),
reads: std::sync::Mutex::new(Slab::new()),
writes: std::sync::Mutex::new(Slab::new()),
read_wakers: WakerSet::new(),
write_wakers: WakerSet::new(),
value: UnsafeCell::new(t),
}
}
@ -104,100 +99,61 @@ impl<T> RwLock<T> {
/// # })
/// ```
pub async fn read(&self) -> RwLockReadGuard<'_, T> {
pub struct LockFuture<'a, T> {
pub struct ReadFuture<'a, T> {
lock: &'a RwLock<T>,
opt_key: Option<usize>,
acquired: bool,
}
impl<'a, T> Future for LockFuture<'a, T> {
impl<'a, T> Future for ReadFuture<'a, T> {
type Output = RwLockReadGuard<'a, T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.lock.try_read() {
Some(guard) => {
self.acquired = true;
Poll::Ready(guard)
}
let poll = match self.lock.try_read() {
Some(guard) => Poll::Ready(guard),
None => {
let mut reads = self.lock.reads.lock().unwrap();
// Register the current task.
// Insert this lock operation.
match self.opt_key {
None => {
// Insert a new entry into the list of blocked reads.
let w = cx.waker().clone();
let key = reads.insert(Some(w));
self.opt_key = Some(key);
if reads.len() == 1 {
self.lock.state.fetch_or(BLOCKED_READS, Ordering::Relaxed);
}
}
Some(key) => {
// There is already an entry in the list of blocked reads. Just
// reset the waker if it was removed.
if reads[key].is_none() {
let w = cx.waker().clone();
reads[key] = Some(w);
}
}
None => self.opt_key = Some(self.lock.read_wakers.insert(cx)),
Some(key) => self.lock.read_wakers.update(key, cx),
}
// Try locking again because it's possible the lock got unlocked just
// before the current task was registered as a blocked task.
// before the current task was inserted into the waker set.
match self.lock.try_read() {
Some(guard) => {
self.acquired = true;
Poll::Ready(guard)
}
Some(guard) => Poll::Ready(guard),
None => Poll::Pending,
}
}
};
if poll.is_ready() {
// If the current task is in the set, remove it.
if let Some(key) = self.opt_key.take() {
self.lock.read_wakers.complete(key);
}
}
poll
}
}
impl<T> Drop for LockFuture<'_, T> {
impl<T> Drop for ReadFuture<'_, T> {
fn drop(&mut self) {
// If the current task is still in the set, that means it is being cancelled now.
if let Some(key) = self.opt_key {
let mut reads = self.lock.reads.lock().unwrap();
let opt_waker = reads.remove(key);
if reads.is_empty() {
self.lock.state.fetch_and(!BLOCKED_READS, Ordering::Relaxed);
}
self.lock.read_wakers.cancel(key);
if opt_waker.is_none() {
// We were awoken. Wake up another blocked read.
if let Some((_, opt_waker)) = reads.iter_mut().next() {
if let Some(w) = opt_waker.take() {
w.wake();
return;
}
}
drop(reads);
if !self.acquired {
// We didn't acquire the lock and didn't wake another blocked read.
// Wake a blocked write instead.
let mut writes = self.lock.writes.lock().unwrap();
if let Some((_, opt_waker)) = writes.iter_mut().next() {
if let Some(w) = opt_waker.take() {
w.wake();
return;
}
}
}
// If there are no active readers, wake one of the writers.
if self.lock.state.load(Ordering::SeqCst) & READ_COUNT_MASK == 0 {
self.lock.write_wakers.notify_one();
}
}
}
}
LockFuture {
ReadFuture {
lock: self,
opt_key: None,
acquired: false,
}
.await
}
@ -226,7 +182,7 @@ impl<T> RwLock<T> {
/// # })
/// ```
pub fn try_read(&self) -> Option<RwLockReadGuard<'_, T>> {
let mut state = self.state.load(Ordering::Acquire);
let mut state = self.state.load(Ordering::SeqCst);
loop {
// If a write lock is currently held, then a read lock cannot be acquired.
@ -234,12 +190,17 @@ impl<T> RwLock<T> {
return None;
}
// Make sure the number of readers doesn't overflow.
if state > isize::MAX as usize {
process::abort();
}
// Increment the number of active reads.
match self.state.compare_exchange_weak(
state,
state + ONE_READ,
Ordering::AcqRel,
Ordering::Acquire,
Ordering::SeqCst,
Ordering::SeqCst,
) {
Ok(_) => return Some(RwLockReadGuard(self)),
Err(s) => state = s,
@ -268,99 +229,59 @@ impl<T> RwLock<T> {
/// # })
/// ```
pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
pub struct LockFuture<'a, T> {
pub struct WriteFuture<'a, T> {
lock: &'a RwLock<T>,
opt_key: Option<usize>,
acquired: bool,
}
impl<'a, T> Future for LockFuture<'a, T> {
impl<'a, T> Future for WriteFuture<'a, T> {
type Output = RwLockWriteGuard<'a, T>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.lock.try_write() {
Some(guard) => {
self.acquired = true;
Poll::Ready(guard)
}
let poll = match self.lock.try_write() {
Some(guard) => Poll::Ready(guard),
None => {
let mut writes = self.lock.writes.lock().unwrap();
// Register the current task.
// Insert this lock operation.
match self.opt_key {
None => {
// Insert a new entry into the list of blocked writes.
let w = cx.waker().clone();
let key = writes.insert(Some(w));
self.opt_key = Some(key);
if writes.len() == 1 {
self.lock.state.fetch_or(BLOCKED_WRITES, Ordering::Relaxed);
}
}
Some(key) => {
// There is already an entry in the list of blocked writes. Just
// reset the waker if it was removed.
if writes[key].is_none() {
let w = cx.waker().clone();
writes[key] = Some(w);
}
}
None => self.opt_key = Some(self.lock.write_wakers.insert(cx)),
Some(key) => self.lock.write_wakers.update(key, cx),
}
// Try locking again because it's possible the lock got unlocked just
// before the current task was registered as a blocked task.
// before the current task was inserted into the waker set.
match self.lock.try_write() {
Some(guard) => {
self.acquired = true;
Poll::Ready(guard)
}
Some(guard) => Poll::Ready(guard),
None => Poll::Pending,
}
}
};
if poll.is_ready() {
// If the current task is in the set, remove it.
if let Some(key) = self.opt_key.take() {
self.lock.write_wakers.complete(key);
}
}
poll
}
}
impl<T> Drop for LockFuture<'_, T> {
impl<T> Drop for WriteFuture<'_, T> {
fn drop(&mut self) {
// If the current task is still in the set, that means it is being cancelled now.
if let Some(key) = self.opt_key {
let mut writes = self.lock.writes.lock().unwrap();
let opt_waker = writes.remove(key);
if writes.is_empty() {
self.lock
.state
.fetch_and(!BLOCKED_WRITES, Ordering::Relaxed);
}
if opt_waker.is_none() && !self.acquired {
// We were awoken but didn't acquire the lock. Wake up another write.
if let Some((_, opt_waker)) = writes.iter_mut().next() {
if let Some(w) = opt_waker.take() {
w.wake();
return;
}
}
drop(writes);
// There are no blocked writes. Wake a blocked read instead.
let mut reads = self.lock.reads.lock().unwrap();
if let Some((_, opt_waker)) = reads.iter_mut().next() {
if let Some(w) = opt_waker.take() {
w.wake();
return;
}
}
if !self.lock.write_wakers.cancel(key) {
// If no other blocked reader was notified, notify all readers.
self.lock.read_wakers.notify_all();
}
}
}
}
LockFuture {
WriteFuture {
lock: self,
opt_key: None,
acquired: false,
}
.await
}
@ -389,24 +310,10 @@ impl<T> RwLock<T> {
/// # })
/// ```
pub fn try_write(&self) -> Option<RwLockWriteGuard<'_, T>> {
let mut state = self.state.load(Ordering::Acquire);
loop {
// If any kind of lock is currently held, then a write lock cannot be acquired.
if state & (WRITE_LOCK | READ_COUNT_MASK) != 0 {
return None;
}
// Set the write lock.
match self.state.compare_exchange_weak(
state,
state | WRITE_LOCK,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => return Some(RwLockWriteGuard(self)),
Err(s) => state = s,
}
if self.state.compare_and_swap(0, WRITE_LOCK, Ordering::SeqCst) == 0 {
Some(RwLockWriteGuard(self))
} else {
None
}
}
@ -449,18 +356,15 @@ impl<T> RwLock<T> {
impl<T: fmt::Debug> fmt::Debug for RwLock<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.try_read() {
None => {
struct LockedPlaceholder;
impl fmt::Debug for LockedPlaceholder {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("<locked>")
}
}
f.debug_struct("RwLock")
.field("data", &LockedPlaceholder)
.finish()
struct Locked;
impl fmt::Debug for Locked {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("<locked>")
}
}
match self.try_read() {
None => f.debug_struct("RwLock").field("data", &Locked).finish(),
Some(guard) => f.debug_struct("RwLock").field("data", &&*guard).finish(),
}
}
@ -486,18 +390,11 @@ unsafe impl<T: Sync> Sync for RwLockReadGuard<'_, T> {}
impl<T> Drop for RwLockReadGuard<'_, T> {
fn drop(&mut self) {
let state = self.0.state.fetch_sub(ONE_READ, Ordering::AcqRel);
// If this was the last read and there are blocked writes, wake one of them up.
if (state & READ_COUNT_MASK) == ONE_READ && state & BLOCKED_WRITES != 0 {
let mut writes = self.0.writes.lock().unwrap();
let state = self.0.state.fetch_sub(ONE_READ, Ordering::SeqCst);
if let Some((_, opt_waker)) = writes.iter_mut().next() {
// If there is no waker in this entry, that means it was already woken.
if let Some(w) = opt_waker.take() {
w.wake();
}
}
// If this was the last read, wake one of the writers.
if state & READ_COUNT_MASK == ONE_READ {
self.0.write_wakers.notify_one();
}
}
}
@ -530,25 +427,12 @@ unsafe impl<T: Sync> Sync for RwLockWriteGuard<'_, T> {}
impl<T> Drop for RwLockWriteGuard<'_, T> {
fn drop(&mut self) {
let state = self.0.state.fetch_and(!WRITE_LOCK, Ordering::AcqRel);
let mut guard = None;
// Check if there are any blocked reads or writes.
if state & BLOCKED_READS != 0 {
guard = Some(self.0.reads.lock().unwrap());
} else if state & BLOCKED_WRITES != 0 {
guard = Some(self.0.writes.lock().unwrap());
}
self.0.state.store(0, Ordering::SeqCst);
// Wake up a single blocked task.
if let Some(mut guard) = guard {
if let Some((_, opt_waker)) = guard.iter_mut().next() {
// If there is no waker in this entry, that means it was already woken.
if let Some(w) = opt_waker.take() {
w.wake();
}
}
// Notify all blocked readers.
if !self.0.read_wakers.notify_all() {
// If there were no blocked readers, notify a blocked writer.
self.0.write_wakers.notify_one();
}
}
}

36
src/sync/waker_set.rs
Unescape Escape View File

@ -95,8 +95,11 @@ impl WakerSet {
}
/// Removes the waker of a cancelled operation.
pub fn cancel(&self, key: usize) {
///
/// Returns `true` if another blocked operation from the set was notified.
pub fn cancel(&self, key: usize) -> bool {
let mut inner = self.lock();
if inner.entries.remove(key).is_none() {
inner.none_count -= 1;
@ -107,33 +110,45 @@ impl WakerSet {
w.wake();
inner.none_count += 1;
}
return true;
}
}
false
}
/// Notifies one blocked operation.
///
/// Returns `true` if an operation was notified.
#[inline]
pub fn notify_one(&self) {
pub fn notify_one(&self) -> bool {
// Use `SeqCst` ordering to synchronize with `Lock::drop()`.
if self.flag.load(Ordering::SeqCst) & NOTIFY_ONE != 0 {
self.notify(false);
self.notify(false)
} else {
false
}
}
/// Notifies all blocked operations.
// TODO: Delete this attribute when `crate::sync::channel()` is stabilized.
#[cfg(feature = "unstable")]
///
/// Returns `true` if at least one operation was notified.
#[inline]
pub fn notify_all(&self) {
pub fn notify_all(&self) -> bool {
// Use `SeqCst` ordering to synchronize with `Lock::drop()`.
if self.flag.load(Ordering::SeqCst) & NOTIFY_ALL != 0 {
self.notify(true);
self.notify(true)
} else {
false
}
}
/// Notifies blocked operations, either one or all of them.
fn notify(&self, all: bool) {
///
/// Returns `true` if at least one operation was notified.
fn notify(&self, all: bool) -> bool {
let mut inner = &mut *self.lock();
let mut notified = false;
for (_, opt_waker) in inner.entries.iter_mut() {
// If there is no waker in this entry, that means it was already woken.
@ -141,10 +156,15 @@ impl WakerSet {
w.wake();
inner.none_count += 1;
}
notified = true;
if !all {
break;
}
}
notified
}
/// Locks the list of entries.

Write Preview
Loading…
Cancel
Save