task/future: support spawning locally #24
Conversation
It uses a thread local variable to detect if it's in thread pool. Spawning locally can utilize maximum cache locality. Signed-off-by: Jay Lee <busyjaylee@gmail.com>
BusyJay
force-pushed
the
local-spawn-future
branch
from
577950c
to
031ffd1
Compare
Signed-off-by: Jay Lee <busyjaylee@gmail.com>
src/task/future.rs
Outdated
| @@ -147,10 +149,57 @@ unsafe fn task_cell(task: *const Task) -> TaskCell { | |||
| #[inline] | |||
| unsafe fn clone_task(task: *const Task) -> TaskCell { | |||
| let task_cell = task_cell(task); | |||
| let extras = { &mut *task_cell.0.extras.get() }; | |||
| // `remote` is none only when it has been constructed but never been polled. | |||
There was a problem hiding this comment.
On the first poll, remote also seems None.
Signed-off-by: Jay Lee <busyjaylee@gmail.com>
Signed-off-by: Jay Lee <busyjaylee@gmail.com>
| if ptr.get().is_null() { | ||
| // It's out of polling process, has to be spawn to global queue. | ||
| // It needs to clone to make it safe as it's unclear whether `self` | ||
| // is still used inside method `spawn` after `TaskCell` is dropped. |
There was a problem hiding this comment.
Strictly speaking, there is still some risk because Waker is accidentally Sync. rust-lang/rust#66481
It means the user might be able to move the reference to other threads (for example, using crossbeam::scoped`) and break our assumption.
However, I think it can be ignored. Maybe we can give some documentation and use expect to give some messages?
Signed-off-by: Jay Lee <busyjaylee@gmail.com>
src/task/future.rs
Outdated
| @@ -85,6 +88,23 @@ unsafe fn waker(task: *const Task) -> Waker { | |||
| #[inline] | |||
| unsafe fn clone_raw(this: *const ()) -> RawWaker { | |||
| let task_cell = clone_task(this as *const Task); | |||
| let extras = { &mut *task_cell.0.extras.get() }; | |||
There was a problem hiding this comment.
I think we should use an immutable reference here and get the mutable reference only at L98 to make the program sound.
There was a problem hiding this comment.
Two wakers in different threads concurrently, then there can be two multable references to the extras at the same time.
There was a problem hiding this comment.
If you mean clone_raw can be executed concurrently due to L100, then it's unsound either to move it to L98. Lock is required.
There was a problem hiding this comment.
Ahh, you're right. Executing L98 and L91 concurrently is unsound.
| .remote | ||
| .as_ref() | ||
| .expect("waker should not be moved to other threads by reference") | ||
| .spawn(TaskCell(task.clone().into_owned())); |
There was a problem hiding this comment.
Sorry I still don't understand why we need a clone here...
There was a problem hiding this comment.
TaskCell can be dropped inside method spawn, which can make self invalid.
src/task/future.rs
Outdated
| @@ -427,29 +483,62 @@ mod tests { | |||
| assert_eq!(res_rx.recv().unwrap(), 4); | |||
| } | |||
|
|
|||
| struct ForwardWaker { | |||
There was a problem hiding this comment.
It's identical to WakeLater?
There was a problem hiding this comment.
Exactly. Didn't notice it before.
Signed-off-by: Jay Lee <busyjaylee@gmail.com>
Signed-off-by: Jay Lee <busyjaylee@gmail.com>
| @@ -257,6 +240,15 @@ impl crate::pool::Runner for Runner { | |||
| task.status.store(COMPLETED, SeqCst); | |||
| return true; | |||
| } | |||
| let extras = { &mut *task.extras.get() }; | |||
There was a problem hiding this comment.
I think it should be executed before poll. Otherwise, before the first poll finished, waking a waker moved to other threads will panic.
There was a problem hiding this comment.
We can use mut_extras here.
There was a problem hiding this comment.
No, it won't. Waking a polling future will mark it NOTIFIED instead of actually waking it up.
There was a problem hiding this comment.
We can use mut_extras here
They are not the same extras. Besides, task_cell is moved.
It uses a thread local variable to detect if it's in thread pool.
Spawning locally can utilize maximum cache locality.