-
-
Notifications
You must be signed in to change notification settings - Fork 2.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Implement a Send'able Handle for the single-threaded Runtime to spawn… #340
Conversation
// Create an mpsc channel for scheduling new tasks on this runtime | ||
// from a different thread | ||
// XXX: Should this use a bounded channel with an arbitrary limit instead, and then | ||
// panic spawning if the channel is full? |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Are there any opinions on this?
Also some further work is needed so that run()
ignores our channel instead of just running forever
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm not sure how to make run()
not wait. The way how I solved that in my code is that I'm not using a channel but just a Mutex<VecDequeue>
and then use turn()
instead of run()
and unpark whenever a future is scheduled.
Any better ideas?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
... let's just move it into the executor directly. There there's control over all that :)
// XXX: Should this use a bounded channel with an arbitrary limit instead, and then | ||
// panic spawning if the channel is full? | ||
let (spawn_sender, spawn_receiver) = mpsc::unbounded(); | ||
executor.spawn(spawn_receiver.for_each(|future| { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We might want to do this only the first time that handle()
is called, but it complicates things slightly. Opinions?
Ping me when you think this is reviewable 👍 |
Thanks, will do! Unfortunately got side-tracked with other things, but hopefully will have some time this weekend |
a4dbc16
to
36145e0
Compare
Sorry for the delay, this is now updated to have the handle implemented directly on the executor (and re-exported from the runtime). And it has a test @carllerche this should be good for reviewing now. Only open question from my side is whether this should actually be an unbounded channel, and if there should be a standalone function |
Spawning a future is already an unbounded action, so making the channel unbounded isn't changing anything. |
@tikue That was also my thought, and if you spawn too many futures you'll probably have another problem already before that channel is becoming one |
Any other comments, how should we move forward with this? :) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry for the delay (stuff came up)! Looks like a great start. I posted some feedback inline.
src/executor/current_thread/mod.rs
Outdated
@@ -304,10 +311,16 @@ impl<P: Park> CurrentThread<P> { | |||
pub fn new_with_park(park: P) -> Self { | |||
let unpark = park.unpark(); | |||
|
|||
// XXX: Should this use a bounded channel with an arbitrary limit instead, and then | |||
// panic spawning if the channel is full? | |||
let (spawn_sender, spawn_receiver) = mpsc::unbounded(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Unbounded is fine here. Generally, spawn
fns do not have back pressure. Instead, they either succeed or fail. One accepted pattern is for the executor to have a max # of tasks and spawn
to fail if that is reached. This feature can be punted to another PR though.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Removed the comment
src/executor/current_thread/mod.rs
Outdated
spawn_handle: Handle, | ||
|
||
/// Receiver for futures spawned from other threads | ||
spawn_receiver: executor::Spawn<mpsc::UnboundedReceiver<Box<Future<Item = (), Error = ()> + Send + 'static>>> |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This can probably be a plain old queue. There is no need to pull in all the future enabled stuff since this is for an executor. The easiest thing here is probably to use std::sync::mpsc
and try_recv
.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks, it indeed is much simpler with a normal mpsc queue. Changed it accordingly
src/executor/current_thread/mod.rs
Outdated
let notify = borrow.scheduler.notify(); | ||
|
||
loop { | ||
let res = borrow.enter(self.enter, || { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A lot of this can probably be simplified by not using the future enabled channel.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes
src/executor/current_thread/mod.rs
Outdated
/// | ||
/// This function panics if the spawn fails. Failure occurs if the `CurrentThread` | ||
/// instance of the `Handle` does not exist anymore. | ||
pub fn spawn<F>(&self, future: F) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For forwards compat, we will probably want to return Result
here (matching https://docs.rs/tokio/0.1.6/tokio/executor/trait.Executor.html#tymethod.spawn).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
@@ -8,6 +8,9 @@ use futures::Future; | |||
|
|||
use std::io; | |||
|
|||
// Re-export Handle from here | |||
pub use executor::current_thread::Handle; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For forwards compat, I would wrap this in a new type.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
…entThread executor to spawn new tasks
…ed on the Handle works
5e60587
to
f5c5ae5
Compare
…bility ... and consistency with the other spawn() functions.
…rentThread executor This simplifies the code quite a bit and is doing all we need anyway.
f5c5ae5
to
f66538a
Compare
Applied all the review comments in separate commits and rebased everything on top of latest master. Thanks for the review! |
@carllerche is there anything else left to be done here? |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks good to me 👍 Thanks for sticking with it.
… new tasks