forked from mirror/async-std
Merge branch 'master' into stream_count
Johannes Weissmann
5 years ago
commit
75546ef831
@ -1,46 +1,51 @@
|
||||
use std::mem;
|
||||
use std::pin::Pin;
|
||||
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use super::read_until_internal;
|
||||
use crate::io::{self, BufRead};
|
||||
use crate::stream::Stream;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
|
||||
///
|
||||
/// This stream is created by the [`split`] method on types that implement [`BufRead`].
|
||||
///
|
||||
/// This type is an async version of [`std::io::Split`].
|
||||
///
|
||||
/// [`split`]: trait.BufRead.html#method.lines
|
||||
/// [`BufRead`]: trait.BufRead.html
|
||||
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
|
||||
#[derive(Debug)]
|
||||
pub struct Split<R> {
|
||||
pub(crate) reader: R,
|
||||
pub(crate) buf: Vec<u8>,
|
||||
pub(crate) read: usize,
|
||||
pub(crate) delim: u8,
|
||||
pin_project! {
|
||||
/// A stream over the contents of an instance of [`BufRead`] split on a particular byte.
|
||||
///
|
||||
/// This stream is created by the [`split`] method on types that implement [`BufRead`].
|
||||
///
|
||||
/// This type is an async version of [`std::io::Split`].
|
||||
///
|
||||
/// [`split`]: trait.BufRead.html#method.lines
|
||||
/// [`BufRead`]: trait.BufRead.html
|
||||
/// [`std::io::Split`]: https://doc.rust-lang.org/std/io/struct.Split.html
|
||||
#[derive(Debug)]
|
||||
pub struct Split<R> {
|
||||
#[pin]
|
||||
pub(crate) reader: R,
|
||||
pub(crate) buf: Vec<u8>,
|
||||
pub(crate) read: usize,
|
||||
pub(crate) delim: u8,
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: BufRead> Stream for Split<R> {
|
||||
type Item = io::Result<Vec<u8>>;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let Self {
|
||||
reader,
|
||||
buf,
|
||||
read,
|
||||
delim,
|
||||
} = unsafe { self.get_unchecked_mut() };
|
||||
let reader = unsafe { Pin::new_unchecked(reader) };
|
||||
let n = futures_core::ready!(read_until_internal(reader, cx, *delim, buf, read))?;
|
||||
if n == 0 && buf.is_empty() {
|
||||
let this = self.project();
|
||||
let n = futures_core::ready!(read_until_internal(
|
||||
this.reader,
|
||||
cx,
|
||||
*this.delim,
|
||||
this.buf,
|
||||
this.read
|
||||
))?;
|
||||
if n == 0 && this.buf.is_empty() {
|
||||
return Poll::Ready(None);
|
||||
}
|
||||
if buf[buf.len() - 1] == *delim {
|
||||
buf.pop();
|
||||
if this.buf[this.buf.len() - 1] == *this.delim {
|
||||
this.buf.pop();
|
||||
}
|
||||
Poll::Ready(Some(Ok(mem::replace(buf, vec![]))))
|
||||
Poll::Ready(Some(Ok(mem::replace(this.buf, vec![]))))
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,350 @@
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_std::sync::channel;
|
||||
use async_std::task;
|
||||
use rand::{thread_rng, Rng};
|
||||
|
||||
fn ms(ms: u64) -> Duration {
|
||||
Duration::from_millis(ms)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smoke() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(1);
|
||||
|
||||
s.send(7).await;
|
||||
assert_eq!(r.recv().await, Some(7));
|
||||
|
||||
s.send(8).await;
|
||||
assert_eq!(r.recv().await, Some(8));
|
||||
|
||||
drop(s);
|
||||
assert_eq!(r.recv().await, None);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capacity() {
|
||||
for i in 1..10 {
|
||||
let (s, r) = channel::<()>(i);
|
||||
assert_eq!(s.capacity(), i);
|
||||
assert_eq!(r.capacity(), i);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn len_empty_full() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(2);
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(s.is_empty(), true);
|
||||
assert_eq!(s.is_full(), false);
|
||||
assert_eq!(r.len(), 0);
|
||||
assert_eq!(r.is_empty(), true);
|
||||
assert_eq!(r.is_full(), false);
|
||||
|
||||
s.send(()).await;
|
||||
|
||||
assert_eq!(s.len(), 1);
|
||||
assert_eq!(s.is_empty(), false);
|
||||
assert_eq!(s.is_full(), false);
|
||||
assert_eq!(r.len(), 1);
|
||||
assert_eq!(r.is_empty(), false);
|
||||
assert_eq!(r.is_full(), false);
|
||||
|
||||
s.send(()).await;
|
||||
|
||||
assert_eq!(s.len(), 2);
|
||||
assert_eq!(s.is_empty(), false);
|
||||
assert_eq!(s.is_full(), true);
|
||||
assert_eq!(r.len(), 2);
|
||||
assert_eq!(r.is_empty(), false);
|
||||
assert_eq!(r.is_full(), true);
|
||||
|
||||
r.recv().await;
|
||||
|
||||
assert_eq!(s.len(), 1);
|
||||
assert_eq!(s.is_empty(), false);
|
||||
assert_eq!(s.is_full(), false);
|
||||
assert_eq!(r.len(), 1);
|
||||
assert_eq!(r.is_empty(), false);
|
||||
assert_eq!(r.is_full(), false);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recv() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(100);
|
||||
|
||||
task::spawn(async move {
|
||||
assert_eq!(r.recv().await, Some(7));
|
||||
task::sleep(ms(1000)).await;
|
||||
assert_eq!(r.recv().await, Some(8));
|
||||
task::sleep(ms(1000)).await;
|
||||
assert_eq!(r.recv().await, Some(9));
|
||||
assert_eq!(r.recv().await, None);
|
||||
});
|
||||
|
||||
task::sleep(ms(1500)).await;
|
||||
s.send(7).await;
|
||||
s.send(8).await;
|
||||
s.send(9).await;
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn send() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(1);
|
||||
|
||||
task::spawn(async move {
|
||||
s.send(7).await;
|
||||
task::sleep(ms(1000)).await;
|
||||
s.send(8).await;
|
||||
task::sleep(ms(1000)).await;
|
||||
s.send(9).await;
|
||||
task::sleep(ms(1000)).await;
|
||||
s.send(10).await;
|
||||
});
|
||||
|
||||
task::sleep(ms(1500)).await;
|
||||
assert_eq!(r.recv().await, Some(7));
|
||||
assert_eq!(r.recv().await, Some(8));
|
||||
assert_eq!(r.recv().await, Some(9));
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recv_after_disconnect() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(100);
|
||||
|
||||
s.send(1).await;
|
||||
s.send(2).await;
|
||||
s.send(3).await;
|
||||
|
||||
drop(s);
|
||||
|
||||
assert_eq!(r.recv().await, Some(1));
|
||||
assert_eq!(r.recv().await, Some(2));
|
||||
assert_eq!(r.recv().await, Some(3));
|
||||
assert_eq!(r.recv().await, None);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn len() {
|
||||
const COUNT: usize = 25_000;
|
||||
const CAP: usize = 1000;
|
||||
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(CAP);
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
|
||||
for _ in 0..CAP / 10 {
|
||||
for i in 0..50 {
|
||||
s.send(i).await;
|
||||
assert_eq!(s.len(), i + 1);
|
||||
}
|
||||
|
||||
for i in 0..50 {
|
||||
r.recv().await;
|
||||
assert_eq!(r.len(), 50 - i - 1);
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
|
||||
for i in 0..CAP {
|
||||
s.send(i).await;
|
||||
assert_eq!(s.len(), i + 1);
|
||||
}
|
||||
|
||||
for _ in 0..CAP {
|
||||
r.recv().await.unwrap();
|
||||
}
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
|
||||
let child = task::spawn({
|
||||
let r = r.clone();
|
||||
async move {
|
||||
for i in 0..COUNT {
|
||||
assert_eq!(r.recv().await, Some(i));
|
||||
let len = r.len();
|
||||
assert!(len <= CAP);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
for i in 0..COUNT {
|
||||
s.send(i).await;
|
||||
let len = s.len();
|
||||
assert!(len <= CAP);
|
||||
}
|
||||
|
||||
child.await;
|
||||
|
||||
assert_eq!(s.len(), 0);
|
||||
assert_eq!(r.len(), 0);
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disconnect_wakes_receiver() {
|
||||
task::block_on(async {
|
||||
let (s, r) = channel::<()>(1);
|
||||
|
||||
let child = task::spawn(async move {
|
||||
assert_eq!(r.recv().await, None);
|
||||
});
|
||||
|
||||
task::sleep(ms(1000)).await;
|
||||
drop(s);
|
||||
|
||||
child.await;
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spsc() {
|
||||
const COUNT: usize = 100_000;
|
||||
|
||||
task::block_on(async {
|
||||
let (s, r) = channel(3);
|
||||
|
||||
let child = task::spawn(async move {
|
||||
for i in 0..COUNT {
|
||||
assert_eq!(r.recv().await, Some(i));
|
||||
}
|
||||
assert_eq!(r.recv().await, None);
|
||||
});
|
||||
|
||||
for i in 0..COUNT {
|
||||
s.send(i).await;
|
||||
}
|
||||
drop(s);
|
||||
|
||||
child.await;
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mpmc() {
|
||||
const COUNT: usize = 25_000;
|
||||
const TASKS: usize = 4;
|
||||
|
||||
task::block_on(async {
|
||||
let (s, r) = channel::<usize>(3);
|
||||
let v = (0..COUNT).map(|_| AtomicUsize::new(0)).collect::<Vec<_>>();
|
||||
let v = Arc::new(v);
|
||||
|
||||
let mut tasks = Vec::new();
|
||||
|
||||
for _ in 0..TASKS {
|
||||
let r = r.clone();
|
||||
let v = v.clone();
|
||||
tasks.push(task::spawn(async move {
|
||||
for _ in 0..COUNT {
|
||||
let n = r.recv().await.unwrap();
|
||||
v[n].fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for _ in 0..TASKS {
|
||||
let s = s.clone();
|
||||
tasks.push(task::spawn(async move {
|
||||
for i in 0..COUNT {
|
||||
s.send(i).await;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for t in tasks {
|
||||
t.await;
|
||||
}
|
||||
|
||||
for c in v.iter() {
|
||||
assert_eq!(c.load(Ordering::SeqCst), TASKS);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn oneshot() {
|
||||
const COUNT: usize = 10_000;
|
||||
|
||||
task::block_on(async {
|
||||
for _ in 0..COUNT {
|
||||
let (s, r) = channel(1);
|
||||
|
||||
let c1 = task::spawn(async move { r.recv().await.unwrap() });
|
||||
let c2 = task::spawn(async move { s.send(0).await });
|
||||
|
||||
c1.await;
|
||||
c2.await;
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drops() {
|
||||
const RUNS: usize = 100;
|
||||
|
||||
static DROPS: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
struct DropCounter;
|
||||
|
||||
impl Drop for DropCounter {
|
||||
fn drop(&mut self) {
|
||||
DROPS.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
|
||||
let mut rng = thread_rng();
|
||||
|
||||
for _ in 0..RUNS {
|
||||
task::block_on(async {
|
||||
let steps = rng.gen_range(0, 10_000);
|
||||
let additional = rng.gen_range(0, 50);
|
||||
|
||||
DROPS.store(0, Ordering::SeqCst);
|
||||
let (s, r) = channel::<DropCounter>(50);
|
||||
|
||||
let child = task::spawn({
|
||||
let r = r.clone();
|
||||
async move {
|
||||
for _ in 0..steps {
|
||||
r.recv().await.unwrap();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
for _ in 0..steps {
|
||||
s.send(DropCounter).await;
|
||||
}
|
||||
|
||||
child.await;
|
||||
|
||||
for _ in 0..additional {
|
||||
s.send(DropCounter).await;
|
||||
}
|
||||
|
||||
assert_eq!(DROPS.load(Ordering::SeqCst), steps);
|
||||
drop(s);
|
||||
drop(r);
|
||||
assert_eq!(DROPS.load(Ordering::SeqCst), steps + additional);
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue