summaryrefslogtreecommitdiffstats
path: root/third_party/rust/futures-0.1.31/tests/support
diff options
context:
space:
mode:
Diffstat (limited to 'third_party/rust/futures-0.1.31/tests/support')
-rw-r--r--third_party/rust/futures-0.1.31/tests/support/local_executor.rs164
-rw-r--r--third_party/rust/futures-0.1.31/tests/support/mod.rs134
2 files changed, 298 insertions, 0 deletions
diff --git a/third_party/rust/futures-0.1.31/tests/support/local_executor.rs b/third_party/rust/futures-0.1.31/tests/support/local_executor.rs
new file mode 100644
index 0000000000..cf89e8152f
--- /dev/null
+++ b/third_party/rust/futures-0.1.31/tests/support/local_executor.rs
@@ -0,0 +1,164 @@
+//! Execution of futures on a single thread
+//!
+//! This module has no special handling of any blocking operations other than
+//! futures-aware inter-thread communications, and is not intended to be used to
+//! manage I/O. For futures that do I/O you'll likely want to use `tokio-core`.
+
+#![allow(bare_trait_objects, unknown_lints)]
+
+use std::cell::{Cell, RefCell};
+use std::sync::{Arc, Mutex, mpsc};
+
+use futures::executor::{self, Spawn, Notify};
+use futures::future::{Executor, ExecuteError};
+use futures::{Future, Async};
+
+/// Main loop object
+pub struct Core {
+ tx: mpsc::Sender<usize>,
+ rx: mpsc::Receiver<usize>,
+ notify: Arc<MyNotify>,
+
+ // Slab of running futures used to track what's running and what slots are
+ // empty. Slot indexes are then sent along tx/rx above to indicate which
+ // future is ready to get polled.
+ tasks: RefCell<Vec<Slot>>,
+ next_vacant: Cell<usize>,
+}
+
+enum Slot {
+ Vacant { next_vacant: usize },
+ Running(Option<Spawn<Box<Future<Item = (), Error = ()>>>>),
+}
+
+impl Core {
+ /// Create a new `Core`.
+ pub fn new() -> Self {
+ let (tx, rx) = mpsc::channel();
+ Core {
+ notify: Arc::new(MyNotify {
+ tx: Mutex::new(tx.clone()),
+ }),
+ tx: tx,
+ rx: rx,
+ next_vacant: Cell::new(0),
+ tasks: RefCell::new(Vec::new()),
+ }
+ }
+
+ /// Spawn a future to be executed by a future call to `run`.
+ ///
+ /// The future `f` provided will not be executed until `run` is called
+ /// below. While futures passed to `run` are executing, the future provided
+ /// here will be executed concurrently as well.
+ pub fn spawn<F>(&self, f: F)
+ where F: Future<Item=(), Error=()> + 'static
+ {
+ let idx = self.next_vacant.get();
+ let mut tasks = self.tasks.borrow_mut();
+ match tasks.get_mut(idx) {
+ Some(&mut Slot::Vacant { next_vacant }) => {
+ self.next_vacant.set(next_vacant);
+ }
+ Some(&mut Slot::Running (_)) => {
+ panic!("vacant points to running future")
+ }
+ None => {
+ assert_eq!(idx, tasks.len());
+ tasks.push(Slot::Vacant { next_vacant: 0 });
+ self.next_vacant.set(idx + 1);
+ }
+ }
+ tasks[idx] = Slot::Running(Some(executor::spawn(Box::new(f))));
+ self.tx.send(idx).unwrap();
+ }
+
+ /// Run the loop until the future `f` completes.
+ ///
+ /// This method will block the current thread until the future `f` has
+ /// resolved. While waiting on `f` to finish it will also execute any
+ /// futures spawned via `spawn` above.
+ pub fn run<F>(&self, f: F) -> Result<F::Item, F::Error>
+ where F: Future,
+ {
+ let id = usize::max_value();
+ self.tx.send(id).unwrap();
+ let mut f = executor::spawn(f);
+ loop {
+ if self.turn() {
+ match f.poll_future_notify(&self.notify, id)? {
+ Async::Ready(e) => return Ok(e),
+ Async::NotReady => {}
+ }
+ }
+ }
+ }
+
+ /// "Turns" this event loop one tick.
+ ///
+ /// This'll block the current thread until something happens, and once an
+ /// event happens this will act on that event.
+ ///
+ /// # Return value
+ ///
+ /// Returns `true` if the future passed to `run` should be polled or `false`
+ /// otherwise.
+ fn turn(&self) -> bool {
+ let task_id = self.rx.recv().unwrap();
+ if task_id == usize::max_value() {
+ return true
+ }
+
+ // This may be a spurious wakeup so we're not guaranteed to have a
+ // future associated with `task_id`, so do a fallible lookup.
+ //
+ // Note that we don't want to borrow `self.tasks` for too long so we
+ // try to extract the future here and leave behind a tombstone future
+ // which'll get replaced or removed later. This is how we support
+ // spawn-in-run.
+ let mut future = match self.tasks.borrow_mut().get_mut(task_id) {
+ Some(&mut Slot::Running(ref mut future)) => future.take().unwrap(),
+ Some(&mut Slot::Vacant { .. }) => return false,
+ None => return false,
+ };
+
+ // Drive this future forward. If it's done we remove it and if it's not
+ // done then we put it back in the tasks array.
+ let done = match future.poll_future_notify(&self.notify, task_id) {
+ Ok(Async::Ready(())) | Err(()) => true,
+ Ok(Async::NotReady) => false,
+ };
+ let mut tasks = self.tasks.borrow_mut();
+ if done {
+ tasks[task_id] = Slot::Vacant { next_vacant: self.next_vacant.get() };
+ self.next_vacant.set(task_id);
+ } else {
+ tasks[task_id] = Slot::Running(Some(future));
+ }
+
+ return false
+ }
+}
+
+impl<F> Executor<F> for Core
+ where F: Future<Item = (), Error = ()> + 'static,
+{
+ fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
+ self.spawn(future);
+ Ok(())
+ }
+}
+
+struct MyNotify {
+ // TODO: it's pretty unfortunate to use a `Mutex` here where the `Sender`
+ // itself is basically `Sync` as-is. Ideally this'd use something like
+ // an off-the-shelf mpsc queue as well as `thread::park` and
+ // `Thread::unpark`.
+ tx: Mutex<mpsc::Sender<usize>>,
+}
+
+impl Notify for MyNotify {
+ fn notify(&self, id: usize) {
+ drop(self.tx.lock().unwrap().send(id));
+ }
+}
diff --git a/third_party/rust/futures-0.1.31/tests/support/mod.rs b/third_party/rust/futures-0.1.31/tests/support/mod.rs
new file mode 100644
index 0000000000..297749777a
--- /dev/null
+++ b/third_party/rust/futures-0.1.31/tests/support/mod.rs
@@ -0,0 +1,134 @@
+#![allow(dead_code)]
+
+use std::fmt;
+use std::sync::Arc;
+use std::thread;
+
+use futures::{Future, IntoFuture, Async, Poll};
+use futures::future::FutureResult;
+use futures::stream::Stream;
+use futures::executor::{self, NotifyHandle, Notify};
+use futures::task;
+
+pub mod local_executor;
+
+pub fn f_ok(a: i32) -> FutureResult<i32, u32> { Ok(a).into_future() }
+pub fn f_err(a: u32) -> FutureResult<i32, u32> { Err(a).into_future() }
+pub fn r_ok(a: i32) -> Result<i32, u32> { Ok(a) }
+pub fn r_err(a: u32) -> Result<i32, u32> { Err(a) }
+
+pub fn assert_done<T, F>(f: F, result: Result<T::Item, T::Error>)
+ where T: Future,
+ T::Item: Eq + fmt::Debug,
+ T::Error: Eq + fmt::Debug,
+ F: FnOnce() -> T,
+{
+ assert_eq!(f().wait(), result);
+}
+
+pub fn assert_empty<T: Future, F: FnMut() -> T>(mut f: F) {
+ assert!(executor::spawn(f()).poll_future_notify(&notify_panic(), 0).ok().unwrap().is_not_ready());
+}
+
+pub fn sassert_done<S: Stream>(s: &mut S) {
+ match executor::spawn(s).poll_stream_notify(&notify_panic(), 0) {
+ Ok(Async::Ready(None)) => {}
+ Ok(Async::Ready(Some(_))) => panic!("stream had more elements"),
+ Ok(Async::NotReady) => panic!("stream wasn't ready"),
+ Err(_) => panic!("stream had an error"),
+ }
+}
+
+pub fn sassert_empty<S: Stream>(s: &mut S) {
+ match executor::spawn(s).poll_stream_notify(&notify_noop(), 0) {
+ Ok(Async::Ready(None)) => panic!("stream is at its end"),
+ Ok(Async::Ready(Some(_))) => panic!("stream had more elements"),
+ Ok(Async::NotReady) => {}
+ Err(_) => panic!("stream had an error"),
+ }
+}
+
+pub fn sassert_next<S: Stream>(s: &mut S, item: S::Item)
+ where S::Item: Eq + fmt::Debug
+{
+ match executor::spawn(s).poll_stream_notify(&notify_panic(), 0) {
+ Ok(Async::Ready(None)) => panic!("stream is at its end"),
+ Ok(Async::Ready(Some(e))) => assert_eq!(e, item),
+ Ok(Async::NotReady) => panic!("stream wasn't ready"),
+ Err(_) => panic!("stream had an error"),
+ }
+}
+
+pub fn sassert_err<S: Stream>(s: &mut S, err: S::Error)
+ where S::Error: Eq + fmt::Debug
+{
+ match executor::spawn(s).poll_stream_notify(&notify_panic(), 0) {
+ Ok(Async::Ready(None)) => panic!("stream is at its end"),
+ Ok(Async::Ready(Some(_))) => panic!("stream had more elements"),
+ Ok(Async::NotReady) => panic!("stream wasn't ready"),
+ Err(e) => assert_eq!(e, err),
+ }
+}
+
+pub fn notify_panic() -> NotifyHandle {
+ struct Foo;
+
+ impl Notify for Foo {
+ fn notify(&self, _id: usize) {
+ panic!("should not be notified");
+ }
+ }
+
+ NotifyHandle::from(Arc::new(Foo))
+}
+
+pub fn notify_noop() -> NotifyHandle {
+ struct Noop;
+
+ impl Notify for Noop {
+ fn notify(&self, _id: usize) {}
+ }
+
+ const NOOP : &'static Noop = &Noop;
+
+ NotifyHandle::from(NOOP)
+}
+
+pub trait ForgetExt {
+ fn forget(self);
+}
+
+impl<F> ForgetExt for F
+ where F: Future + Sized + Send + 'static,
+ F::Item: Send,
+ F::Error: Send
+{
+ fn forget(self) {
+ thread::spawn(|| self.wait());
+ }
+}
+
+pub struct DelayFuture<F>(F,bool);
+
+impl<F: Future> Future for DelayFuture<F> {
+ type Item = F::Item;
+ type Error = F::Error;
+
+ fn poll(&mut self) -> Poll<F::Item,F::Error> {
+ if self.1 {
+ self.0.poll()
+ } else {
+ self.1 = true;
+ task::current().notify();
+ Ok(Async::NotReady)
+ }
+ }
+}
+
+/// Introduces one `Ok(Async::NotReady)` before polling the given future
+pub fn delay_future<F>(f: F) -> DelayFuture<F::Future>
+ where F: IntoFuture,
+{
+ DelayFuture(f.into_future(), false)
+}
+
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-27 10:16:06 +0000