sync: update fuzz_oneshot test

tokio-sync/Cargo.toml

@@ -29,4 +29,4 @@ tokio-futures = { version = "0.2.0", path = "../tokio-futures" }
 env_logger = { version = "0.5", default-features = false }
 # tokio = { version = "0.2.0", path = "../tokio" }
 tokio-test = { version = "0.2.0", path = "../tokio-test" }
-loom = { version = "0.1.1", features = ["futures"] }
+loom = { git = "https://github.com/carllerche/loom", branch = "std-future2", features = ["futures"] }

tokio-sync/src/oneshot.rs

@@ -1,6 +1,8 @@
 //! A channel for sending a single message between asynchronous tasks.
 
-use crate::loom::{sync::atomic::AtomicUsize, sync::CausalCell, task::Waker};
+use crate::loom::{sync::atomic::AtomicUsize, sync::CausalCell};
+
+use tokio_futures::ready;
 
 use std::fmt;
 use std::future::Future;
@@ -9,8 +11,7 @@ use std::pin::Pin;
 use std::sync::atomic::Ordering::{self, AcqRel, Acquire};
 use std::sync::Arc;
 use std::task::Poll::{Pending, Ready};
-use std::task::{Context, Poll};
-use tokio_futures::ready;
+use std::task::{Context, Poll, Waker};
 
 /// Sends a value to the associated `Receiver`.
 ///

tokio-sync/tests/fuzz_oneshot.rs

@@ -1,15 +1,16 @@
-#![cfg(feature = "broken")]
 #![deny(warnings, rust_2018_idioms)]
 
 #[path = "../src/oneshot.rs"]
 #[allow(warnings)]
 mod oneshot;
 
-use futures::{self, Async, Future};
+// use futures::{self, Async, Future};
 use loom;
-use loom::futures::block_on;
+use loom::futures::{block_on, poll_future};
 use loom::thread;
 
+use std::task::Poll::{Ready, Pending};
+
 #[test]
 fn smoke() {
  loom::fuzz(|| {
@@ -34,16 +35,14 @@ fn changing_rx_task() {
  });
 
  let rx = thread::spawn(move || {
- let t1 = block_on(futures::future::poll_fn(|| Ok::<_, ()>(rx.poll().into()))).unwrap();
-
- match t1 {
- Ok(Async::Ready(value)) => {
+ match poll_future(&mut rx) {
+ Ready(Ok(value)) => {
  // ok
  assert_eq!(1, value);
  None
  }
- Ok(Async::NotReady) => Some(rx),
- Err(_) => unreachable!(),
+ Ready(Err(_)) => unimplemented!(),
+ Pending => Some(rx),
  }
  })
  .join()
@@ -57,6 +56,30 @@ fn changing_rx_task() {
  });
 }
 
+// TODO: Move this into `oneshot` proper.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+struct OnClose<'a> {
+ tx: &'a mut oneshot::Sender<i32>,
+}
+
+impl<'a> OnClose<'a> {
+ fn new(tx: &'a mut oneshot::Sender<i32>) -> Self {
+ OnClose { tx }
+ }
+}
+
+impl<'a> Future for OnClose<'a> {
+ type Output = ();
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+ self.get_mut().tx.poll_close(cx)
+ }
+}
+
 #[test]
 fn changing_tx_task() {
  loom::fuzz(|| {
@@ -67,23 +90,19 @@ fn changing_tx_task() {
  });
 
  let tx = thread::spawn(move || {
- let t1 = block_on(futures::future::poll_fn(|| {
- Ok::<_, ()>(tx.poll_close().into())
- }))
- .unwrap();
+ let t1 = poll_future(&mut OnClose::new(&mut tx));
 
  match t1 {
- Ok(Async::Ready(())) => None,
- Ok(Async::NotReady) => Some(tx),
- Err(_) => unreachable!(),
+ Ready(()) => None,
+ Pending => Some(tx),
  }
  })
  .join()
  .unwrap();
 
  if let Some(mut tx) = tx {
  // Previous task parked, use a new task...
- block_on(futures::future::poll_fn(move || tx.poll_close())).unwrap();
+ block_on(OnClose::new(&mut tx));
  }
  });
 }