From ddbbbfc32af5c8572303086145a3932518fc5f74 Mon Sep 17 00:00:00 2001 From: nasa Date: Sun, 3 Nov 2019 21:40:51 +0900 Subject: [PATCH 1/3] Replace `VecDeque` with `stream::from_iter` in examples (#447) --- src/option/sum.rs | 6 +- src/result/product.rs | 4 +- src/result/sum.rs | 4 +- src/stream/from_iter.rs | 2 +- src/stream/stream/mod.rs | 280 +++++++++++++++++++-------------------- 5 files changed, 143 insertions(+), 153 deletions(-) diff --git a/src/option/sum.rs b/src/option/sum.rs index 25dc920..5c154f4 100644 --- a/src/option/sum.rs +++ b/src/option/sum.rs @@ -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 = words.map(|w| w.find('a')).sum().await; assert_eq!(total, Some(5)); # diff --git a/src/result/product.rs b/src/result/product.rs index 17afa94..fd24216 100644 --- a/src/result/product.rs +++ b/src/result/product.rs @@ -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 = v.map(|x| if x < 0 { Err("Negative element found") diff --git a/src/result/sum.rs b/src/result/sum.rs index caca4f6..dd68772 100644 --- a/src/result/sum.rs +++ b/src/result/sum.rs @@ -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 = v.map(|x| if x < 0 { Err("Negative element found") diff --git a/src/stream/from_iter.rs b/src/stream/from_iter.rs index 43bc961..5fd216d 100644 --- a/src/stream/from_iter.rs +++ b/src/stream/from_iter.rs @@ -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 { iter: I, } diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index e182c03..923c52b 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -280,11 +280,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![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)); @@ -317,9 +316,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)); @@ -349,10 +348,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)); @@ -385,18 +384,17 @@ extension_trait! { # fn main() { async_std::task::block_on(async { # use async_std::prelude::*; - use std::collections::VecDeque; + use async_std::stream; - let v: VecDeque<_> = vec![&1, &2, &3].into_iter().collect(); - - let mut v_copied = v.copied(); + let s = stream::from_iter(vec![&1, &2, &3]); + let second = stream::from_iter(vec![2, 3]); - 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 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); # # }) } ``` @@ -422,9 +420,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'))); @@ -452,9 +450,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)); @@ -486,10 +484,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)) @@ -518,11 +517,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![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)); @@ -534,11 +532,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 = vec![].into_iter().collect(); + let s = stream::empty::<()>(); let last = s.last().await; assert_eq!(last, None); @@ -599,11 +596,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![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)); @@ -631,14 +627,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 = vec![1,2,3].into_iter().collect(); - let inner2: VecDeque = 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; @@ -668,12 +664,12 @@ extension_trait! { ``` # async_std::task::block_on(async { - use std::collections::VecDeque; use async_std::prelude::*; + use async_std::stream; - let inner1: VecDeque = vec![1,2,3].into_iter().collect(); - let inner2: VecDeque = 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; @@ -701,11 +697,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::().ok()); @@ -742,16 +738,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 = 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::::new().min_by_key(|x| x.abs()).await; + let min = stream::empty::().min_by_key(|x| x.abs()).await; assert_eq!(min, None); # # }) } @@ -779,16 +774,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 = 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::::new().max_by_key(|x| x.abs()).await; + let max = stream::empty::().min_by_key(|x| x.abs()).await; assert_eq!(max, None); # # }) } @@ -816,11 +810,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![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)); @@ -828,7 +821,7 @@ extension_trait! { let min = s.min_by(|x, y| y.cmp(x)).await; assert_eq!(min, Some(3)); - let min = VecDeque::::new().min_by(|x, y| x.cmp(y)).await; + let min = stream::empty::().min_by(|x, y| x.cmp(y)).await; assert_eq!(min, None); # # }) } @@ -854,15 +847,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 = 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::::new().min().await; + let min = stream::empty::().min().await; assert_eq!(min, None); # # }) } @@ -888,11 +881,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![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)); @@ -900,7 +892,7 @@ extension_trait! { let max = s.max_by(|x, y| y.cmp(x)).await; assert_eq!(max, Some(1)); - let max = VecDeque::::new().max_by(|x, y| x.cmp(y)).await; + let max = stream::empty::().max_by(|x, y| x.cmp(y)).await; assert_eq!(max, None); # # }) } @@ -927,11 +919,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 = 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)); @@ -943,11 +934,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 = 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)); @@ -961,11 +951,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 = 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); @@ -1056,9 +1045,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 = 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)); # @@ -1071,9 +1060,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 = 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)); @@ -1101,9 +1090,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)); @@ -1134,9 +1123,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![1u8, 2, 3]); let sum = s.fold(0, |acc, x| acc + x).await; assert_eq!(sum, 6); @@ -1165,12 +1154,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 = 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(); @@ -1271,11 +1260,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![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) @@ -1312,11 +1300,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)); @@ -1342,11 +1329,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![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)); @@ -1408,9 +1394,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![1usize, 2, 3]); let sum = s.try_fold(0, |acc, v| { if (acc+v) % 2 == 1 { Ok(v+3) @@ -1444,13 +1430,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 = 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(); @@ -1502,12 +1488,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 = vec![1, 2, 3].into_iter().collect(); - let r: VecDeque = 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))); @@ -1637,14 +1622,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)); @@ -1676,9 +1661,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![1usize, 2, 3]); let res = s.clone().position(|x| *x == 1).await; assert_eq!(res, Some(0)); @@ -1715,13 +1700,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); @@ -1751,11 +1737,13 @@ extension_trait! { # fn main() { async_std::task::block_on(async { # use async_std::prelude::*; - use std::collections::VecDeque; - let single: VecDeque = vec![1].into_iter().collect(); - let single_ne: VecDeque = vec![10].into_iter().collect(); - let multi: VecDeque = vec![1,2].into_iter().collect(); - let multi_ne: VecDeque = 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); @@ -1786,12 +1774,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 = vec![1].into_iter().collect(); - let single_gt: VecDeque = vec![10].into_iter().collect(); - let multi: VecDeque = vec![1,2].into_iter().collect(); - let multi_gt: VecDeque = 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); @@ -1822,12 +1811,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 = vec![1].into_iter().collect(); - let single_eq: VecDeque = vec![10].into_iter().collect(); - let multi: VecDeque = vec![1,2].into_iter().collect(); - let multi_eq: VecDeque = 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); @@ -1858,12 +1848,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); @@ -1894,12 +1885,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); @@ -1930,12 +1922,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); @@ -1977,10 +1969,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); From 78614c6c1d4ddc70ca7b83f9d902ff85da1eb677 Mon Sep 17 00:00:00 2001 From: Stjepan Glavina Date: Sun, 3 Nov 2019 22:19:04 +0100 Subject: [PATCH 2/3] Clarify blocking in channel docs (#448) --- src/sync/channel.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/sync/channel.rs b/src/sync/channel.rs index 403bee7..d4b6423 100644 --- a/src/sync/channel.rs +++ b/src/sync/channel.rs @@ -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 { impl Sender { /// 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 { impl Receiver { /// 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 /// From 20cdf73bb053d1e6a9468775d96950327bd252e6 Mon Sep 17 00:00:00 2001 From: Stjepan Glavina Date: Mon, 4 Nov 2019 02:40:55 +0100 Subject: [PATCH 3/3] Simplify RwLock using WakerSet (#440) --- src/sync/rwlock.rs | 290 +++++++++++++----------------------------- src/sync/waker_set.rs | 36 ++++-- 2 files changed, 115 insertions(+), 211 deletions(-) diff --git a/src/sync/rwlock.rs b/src/sync/rwlock.rs index a0d0f07..81b0735 100644 --- a/src/sync/rwlock.rs +++ b/src/sync/rwlock.rs @@ -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 { state: AtomicUsize, - reads: std::sync::Mutex>>, - writes: std::sync::Mutex>>, + read_wakers: WakerSet, + write_wakers: WakerSet, value: UnsafeCell, } @@ -77,8 +72,8 @@ impl RwLock { pub fn new(t: T) -> RwLock { 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 RwLock { /// # }) /// ``` pub async fn read(&self) -> RwLockReadGuard<'_, T> { - pub struct LockFuture<'a, T> { + pub struct ReadFuture<'a, T> { lock: &'a RwLock, opt_key: Option, - 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 { - 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 Drop for LockFuture<'_, T> { + impl 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 RwLock { /// # }) /// ``` pub fn try_read(&self) -> Option> { - 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 RwLock { 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 RwLock { /// # }) /// ``` pub async fn write(&self) -> RwLockWriteGuard<'_, T> { - pub struct LockFuture<'a, T> { + pub struct WriteFuture<'a, T> { lock: &'a RwLock, opt_key: Option, - 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 { - 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 Drop for LockFuture<'_, T> { + impl 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 RwLock { /// # }) /// ``` pub fn try_write(&self) -> Option> { - 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 RwLock { impl fmt::Debug for RwLock { 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("") - } - } - 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("") } + } + + 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 Sync for RwLockReadGuard<'_, T> {} impl 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 Sync for RwLockWriteGuard<'_, T> {} impl 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(); } } } diff --git a/src/sync/waker_set.rs b/src/sync/waker_set.rs index 8fd1b62..eb44a67 100644 --- a/src/sync/waker_set.rs +++ b/src/sync/waker_set.rs @@ -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.