Executors and partly reenable futures crate #1037

MajorBreakfast · 2018-06-12T00:57:00Z

This needs careful review. I'm not entirely sure what I did ^^' All I can say is that it builds now and some of the tests run.

Notes:

~~futures-core subcrate does not yet reexport from core~~
executor.spawn_obj(Box::new(future).into()) looks ugly ^^' I think we should introduce an IntoTaskObj trait. Unless there's going to be a simpler API layered on top of course.
The doc tests seem to require #![feature(arbitrary_self_types)]. Can this be done? Didn't seem to work for me. Maybe I just did it the wrong way.
Various places require Executor to be Sized. Should it become the default, so that we don't need to type it everywhere?
The Rust compiler is genius. It reports new errors blazing fast and the error messages are extremely helpful
The Task structs should probably be renamed to something like TaskContainer { task_obj, ... }.

MajorBreakfast · 2018-06-12T08:07:05Z

Because this PR is so big and requires careful vetting, #1034 should be reviewed and merged first. I will then rebase on top of it after it has landed.

MajorBreakfast · 2018-06-12T09:06:29Z

In this recent commit I renamed Task to TaskContainer. Previously we had two things called task:

TaskObj: The root future wrapped in a custom trait object
Task: The environment { task, exec, wake_handle } that contains it and other stuff

Calling both things "task" would be too confusing.

Alternative to what I implemented:

Rename TaskObj to something else. It's only a stopgap solution until dyn can be passed by value after all anyway (see RFC section). One possible good name would be RootFutureObj
Rename TaskContainer back to Task

I quite like the alternative. What do you think?

MajorBreakfast · 2018-06-12T12:49:44Z

Okay here are some updated thoughts on this:

What about we rename TaskObj to TaskFutureObj:

"Task" signifies that it's for a task and should have Output = ()
"Future" signifies that it can be polled. TaskFutureObj should therefore also implement Future to make this perfectly clear: Currently it just implements a method call poll_task() (which I already suggested to rename to poll). But we should go the extra mile and make it a real future type. Because it is Unpin executors will be able to call poll_unpin. So, the same signature as today's poll_task() is also available.
"Obj" signifies that it's a custom trait object
The trait will be called UnsafeTaskFuture

Edit: I no longer believe we should do this.

@cramertj
@aturon

cramertj · 2018-06-12T16:39:31Z

executor.span_obj(Box::new(future).into()) looks ugly ^^' I think we should introduce an IntoTaskObj trait. Unless there's going to be a simpler API layered on top of course.

The plan was originally to add a method to Context that would take a generic future, box it, and spawn it. However, that impl needs std and Context is defined in core, so I think we should add it via an extension trait here instead.

cramertj · 2018-06-12T18:39:29Z

Various places require Executor to be Sized. Should it become the default, so that we don't need to type it everywhere?

No-- the executor inside of task::Context is &mut dyn Executor.

MajorBreakfast · 2018-06-13T09:13:42Z

I now think we should stick with calling the top level future "task".

(And rename Task to TaskContainer to avoid confusion)

MajorBreakfast · 2018-06-20T20:46:41Z

I rebased and updated it

cramertj · 2018-06-20T22:18:01Z

futures-executor/src/local_pool.rs

-struct Task {
-    fut: Box<Future<Item = (), Error = Never>>,
-    map: LocalMap,
+struct TaskContainer {


Can you add a comment to this that describes what it's for? Something like "wrapper for TaskObj that implements Future with a unit output" would help-- it's less obvious what its purpose is now that it no longer pairs the task with a LocalMap.

N.B. I think that TaskObj should implement Future-- we should add that impl in std. Opened rust-lang/rust#51667

Already made TaskObj implement Future

I‘ll add the comment

I'm getting rid of TaskContainer entirely. We can use TaskObj directly. Can't do that in ThreadPool, but here it's simple

cramertj · 2018-06-20T22:22:28Z

futures-executor/src/local_pool.rs

-    pub fn run_until<F>(&mut self, mut f: F, exec: &mut Executor) -> Result<F::Item, F::Error>
-        where F: Future
+    pub fn run_until<F, Exec>(&mut self, mut f: F, exec: &mut Exec) -> F::Output
+        where F: Future + Unpin, Exec: Executor + Sized


This Unpin bound isn't needed. Can you add this macro to futures-core and use it instead?

cramertj · 2018-06-20T22:22:49Z

futures-executor/src/local_pool.rs

@@ -202,14 +197,14 @@ lazy_static! {
 ///
 /// Use a [`LocalPool`](LocalPool) if you need finer-grained control over
 /// spawned tasks.
-pub fn block_on<F: Future>(f: F) -> Result<F::Item, F::Error> {
+pub fn block_on<F: Future + Unpin>(f: F) -> F::Output {


cramertj · 2018-06-20T22:24:25Z

futures-executor/src/spawn.rs

-    type Item = ();
-    type Error = Never;
-    fn poll(&mut self, cx: &mut Context) -> Poll<(), Never> {
+impl<F: Future<Output = ()> + Unpin + Send + 'static> Future for Spawn<F> {


This Unpin bound isn't needed.

cramertj · 2018-06-20T22:26:24Z

futures-executor/src/spawn.rs

@@ -131,6 +132,7 @@ struct MySender<F, T> {
    tx: Option<Sender<T>>,
    keep_running_flag: Arc<AtomicBool>,
 }
+impl<F, T> Unpin for MySender<F, T> {} // ToDo: May I do this?


I would only add it where F: Unpin, and I would remove the Unpin bound on the Future impl. You will want to use the unsafe_pinned_field macro to create a safe function for accessing PinMut<F>

cramertj · 2018-06-20T22:29:39Z

futures-executor/src/thread_pool.rs

@@ -96,23 +96,22 @@ impl ThreadPool {
    ///
    /// Note that the function will return when the provided future completes,
    /// even if some of the tasks it spawned are still running.
-    pub fn run<F: Future>(&mut self, f: F) -> Result<F::Item, F::Error> {
+    pub fn run<F: Future + Unpin>(&mut self, f: F) -> F::Output {


cramertj · 2018-06-20T22:30:37Z

futures-util/src/sink/mod.rs

@@ -9,7 +9,7 @@ use super::future::Either;
 use core::mem::PinMut;

 mod close;
-mod fanout;
+// mod fanout;


Why is fanout commented out?

Because the tests fail. There is no more iter_ok() for instance. Fixing now...

No this is a rabbit hole... it requires converting future_util/src/stream/forwards.rs and probably various other things

Can you leave fanout, but disable the tests? I'd like to make sure I know what code still needs fixing to work on 0.3 (regardless of tests) and what still needs fixing

Done. Tests are disabled now.

cramertj · 2018-06-20T22:31:12Z

futures-util/src/stream/futures_unordered.rs

-impl<T: Sync> Sync for FuturesUnordered<T> {}
+unsafe impl<T: Send> Send for FuturesUnordered<T> {}
+unsafe impl<T: Sync> Sync for FuturesUnordered<T> {}
+impl<T: Unpin> Unpin for FuturesUnordered<T> {}


FuturesUnordered should be Unpin regardless of T.

cramertj · 2018-06-20T22:31:54Z

futures-util/src/stream/futures_unordered.rs

@@ -99,12 +103,12 @@ enum Dequeue<T> {
 }

 impl<T> FuturesUnordered<T>
-    where T: Future,
+    where T: Future + Unpin


Here and elsewhere, FuturesUnordered shouldn't require that the future is Unpin, since it's allocating it. Instead it should track usage to ensure the future isn't moved.

cramertj · 2018-06-20T22:34:55Z

futures-util/src/stream/futures_unordered.rs

-            Waker::new(hide_lt(ptr))
+            let ptr: Arc<Node<T>> = handle.0.clone();
+            let ptr: *mut ArcNode<T> = mem::transmute(ptr);
+            let ptr = mem::transmute(ptr as *mut UnsafeWake); // Hide lifetime


Can you supply manual type arguments to mem::transmute? It's hard to tell what it's transmuting between here.

MajorBreakfast · 2018-06-21T11:53:59Z

I've split FuturesUnordered into multiple files. Whenever a file is large I spend most of my time scrolling and trying to find stuff instead of getting things done ^^'

MajorBreakfast · 2018-06-21T13:50:07Z

@cramertj To make reviewing easier, look at the "Future + !Unpin support for FuturesUnordered" commit directly. I split the file in the preceding commit. This PR seems a bit bigger than it really is (it's still big of course)

cramertj · 2018-06-21T21:05:35Z

futures-util/src/stream/futures_unordered/mod.rs

+                    // At this point, it may be worth yielding the thread &
+                    // spinning a few times... but for now, just yield using the
+                    // task system.
+                    cx.waker().wake();


Nit: you can use local_waker().wake() here

cramertj · 2018-06-21T21:09:02Z

futures-util/src/stream/futures_unordered/iter_mut.rs

+use super::node::Node;
+
+#[derive(Debug)]
+/// Mutable iterator over all futures in the unordered set.


Subtle: this is unsound now because it gives mutable access to the futures allowing them to be mem::replaced, although they may have been polled. This should either yield PinMut<F> or iter_mut should be changed to include an F: Unpin bound. Perhaps offer both as iter_mut and iter_pin_mut?

Good catch! Done.

cramertj · 2018-06-21T21:10:08Z

futures-util/src/stream/futures_unordered/node.rs

+            let ptr = mem::transmute::<Arc<Node<T>>, *mut ArcNode<T>>(ptr);
+            let ptr = ptr as *mut UnsafeWake;
+            // Hide lifetime
+            let ptr = mem::transmute::<*mut UnsafeWake, *mut UnsafeWake>(ptr);


This looks like a no-op?

It hides the lifetime. Before my PR it was in a function called hide_lt. Otherwise we get:

error[E0310]: the parameter type `T` may not live long enough --> futures-util/src/stream/futures_unordered/node.rs:96:23 | 91 | impl<'a, T> From<NodeToHandle<'a, T>> for LocalWaker { | - help: consider adding an explicit lifetime bound `T: 'static`... ... 96 | let ptr = ptr as *mut UnsafeWake; | ^^^ | note: ...so that the type `stream::futures_unordered::node::ArcNode<T>` will meet its required lifetime bounds

cramertj · 2018-06-21T21:12:41Z

futures-util/src/stream/futures_unordered/node.rs

+unsafe impl<T> Send for ArcNode<T> {}
+unsafe impl<T> Sync for ArcNode<T> {}
+
+unsafe impl<T> UnsafeWake for ArcNode<T> {


This impl looks like it could be written as an impl of the Wake trait for Node<T>, which would make the From impls above trivial.

It can't because the UnsafeWake trait is used to hide the lifetime. I've added a bunch of comments in my next commit.

Gotcha, thanks!

cramertj · 2018-06-21T21:12:50Z

futures-util/src/stream/futures_unordered/node.rs

+
+#[doc(hidden)]
+impl<'a, T> From<NodeToHandle<'a, T>> for LocalWaker {
+    fn from(handle: NodeToHandle<'a, T>) -> LocalWaker {


This and the function below are the same modulo the call to either LocalWaker::new or Waker::new-- can you break out the shared bits into a function?

…r_pin_mut`

cramertj · 2018-06-22T21:54:27Z

futures-util/src/stream/futures_unordered/iter_pin_mut.rs

+
+#[derive(Debug)]
+/// Mutable iterator over all futures in the unordered set.
+pub struct IterPinMut<'a, F: 'a> {


Rather than duplicating the logic between IterPinMut and IterMut, I'd implement IterMut in terms of IterPinMut. The Unpin bound will make it safe to go from PinMut of the yielded value to &mut of the value.

cramertj · 2018-06-22T21:55:27Z

futures-util/src/stream/futures_unordered/mod.rs

@@ -155,6 +157,15 @@ impl<T> FuturesUnordered<T> {
        }
    }

+    /// Returns an iterator that allows modifying each future in the set.


Can you add an Unpin bound to the iter_mut method itself? It could be confusing to have iter_mut be callable but the result not be usable as an iterator.

cramertj · 2018-06-22T21:56:06Z

futures-util/src/stream/futures_unordered/iter_mut.rs

@@ -10,7 +10,7 @@ pub struct IterMut<'a, F: 'a> {
    pub(super) _marker: PhantomData<&'a mut FuturesUnordered<F>>


Can you add an Unpin bound to the IterMut type itself, too?

cramertj · 2018-06-22T23:02:46Z

futures-util/src/stream/futures_unordered/iter_mut.rs

-            self.len -= 1;
-            Some(future)
-        }
+        PinMut::new(&mut self.0).next().map(|f| unsafe { PinMut::get_mut(f) })


Why the PinMut::new here? You should be able to remove that and have self.0.next().map(|f| unsafe { PinMut::get_mut(f) })

cramertj · 2018-06-22T23:31:58Z

@MajorBreakfast Looks great! Thanks for being so patient with all the changes.

MajorBreakfast · 2018-06-22T23:52:48Z

🎉

MajorBreakfast mentioned this pull request Jun 12, 2018

0.3 alpha burndown issue #1033

Closed

13 tasks

MajorBreakfast force-pushed the 0.3 branch from a883c6b to f8286dc Compare June 20, 2018 20:29

MajorBreakfast added 4 commits June 20, 2018 22:43

Executors and partly reenable futures crate

04842ae

Rename res to output

54d6b2f

Rename Task to TaskContainer

8baad68

Update to latest nightly

883ddff

MajorBreakfast force-pushed the 0.3 branch from f8286dc to 883ddff Compare June 20, 2018 20:44

cramertj reviewed Jun 20, 2018

View reviewed changes

MajorBreakfast added 3 commits June 21, 2018 10:29

Split futures_unordered.rs

95273d6

Future + !Unpin support for FuturesUnordered

52c0fe4

transmute type args

cc6fa32

MajorBreakfast added 2 commits June 21, 2018 14:34

Use TaskObj directly in LocalPool

146e818

Fix bug in stream iter doc

ff85f87

MajorBreakfast added 2 commits June 21, 2018 15:23

Remove Unpin restriction from LocalPool

9c20c18

Lift Unpin restriction from Spawn

3fa5c6e

MajorBreakfast added 3 commits June 21, 2018 21:45

Move ThreadNotify into local_pool.rs

bec00f7

Prefix free function with module

2df964b

Make some more doctests run successfully

41853e7

cramertj reviewed Jun 21, 2018

View reviewed changes

MajorBreakfast added 3 commits June 22, 2018 02:39

Make all enabled doc tests green

ddbb39d

Additionally export Poll under task module (like in std)

f0c8787

Improve Node of FuturesUnordered

d7cc312

MajorBreakfast force-pushed the 0.3 branch from 2faee43 to d7cc312 Compare June 22, 2018 16:01

MajorBreakfast added 5 commits June 22, 2018 19:31

Mostly updates to comments

5a3c6c1

Use cx.local_waker().wake()

ea55900

Bring back `SinkExt::Fanout" but keep its tests disabled for now

a8472bd

Rename Inner to ReadyToRunQueue for clarity

d4d5234

Restrict FuturesUnorderdered::iter_mut to Unpin futures. Add `ite…

a2e07b0

…r_pin_mut`

cramertj reviewed Jun 22, 2018

View reviewed changes

Add more Unpin bounds to FuturesUnordered::iter_mut

7206d59

cramertj reviewed Jun 22, 2018

View reviewed changes

Implement IterMut in terms of IterPinMut

f235429

MajorBreakfast force-pushed the 0.3 branch from da7c346 to f235429 Compare June 22, 2018 23:21

cramertj merged commit 7a67744 into rust-lang:0.3 Jun 22, 2018

		@@ -10,7 +10,7 @@ pub struct IterMut<'a, F: 'a> {
		pub(super) _marker: PhantomData<&'a mut FuturesUnordered<F>>

Executors and partly reenable futures crate #1037

Executors and partly reenable futures crate #1037

Conversation

MajorBreakfast commented Jun 12, 2018 • edited Loading

MajorBreakfast commented Jun 12, 2018 • edited Loading

MajorBreakfast commented Jun 12, 2018 • edited Loading

MajorBreakfast commented Jun 12, 2018 • edited Loading

cramertj commented Jun 12, 2018

cramertj commented Jun 12, 2018

MajorBreakfast commented Jun 13, 2018 • edited Loading

MajorBreakfast commented Jun 20, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MajorBreakfast commented Jun 21, 2018

MajorBreakfast commented Jun 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cramertj Jun 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cramertj commented Jun 22, 2018

MajorBreakfast commented Jun 22, 2018

MajorBreakfast commented Jun 12, 2018 •

edited

Loading

MajorBreakfast commented Jun 12, 2018 •

edited

Loading

MajorBreakfast commented Jun 12, 2018 •

edited

Loading

MajorBreakfast commented Jun 12, 2018 •

edited

Loading

MajorBreakfast commented Jun 13, 2018 •

edited

Loading

cramertj Jun 22, 2018 •

edited

Loading