1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
use std::rc::{Rc, Weak};
use crate::remote::Sender as RemoteSender;
use crate::{ExecutorCore, JoinEventLake, JoinHandle, TaskOptions};
/// Represents the set of tasks currently registered with an executor and allows new tasks
/// to be registered.
///
/// The task set acts like a handle and can be cheaply cloned and used from anywhere on the same
/// thread as the executor, including from inside the futures of tasks running on the same executor.
/// Every clone is functionally identical.
#[derive(Clone, Debug)]
pub struct TaskSet {
// We only keep a weak reference, so even if the owner of this type does something odd, we do
// not extend the lifetime of the `Executor`, which is governed by the `Executor` owner.
core: Weak<ExecutorCore>,
// Retained independently so submissions after executor drop can still produce a valid
// disconnected join handle.
join_events: JoinEventLake,
}
impl TaskSet {
pub(crate) fn new(core: &Rc<ExecutorCore>) -> Self {
Self {
core: Rc::downgrade(core),
join_events: core.join_event_lake(),
}
}
/// Registers a future and its runtime integration options as a new task.
///
/// The future will be polled until it completes, after which it will be dropped, unless the
/// executor is commanded to shut down, in which case the future may be dropped before it
/// completes. A submission after shutdown returns a join handle that resolves to
/// [`JoinError`](crate::JoinError). Use [`TaskOptions::default()`] when no execution context
/// or placement metadata is required.
pub fn add<F, R>(&self, future: F, options: TaskOptions) -> JoinHandle<R>
where
F: IntoFuture<Output = R> + 'static,
R: 'static,
{
let Some(core) = self.core.upgrade() else {
return self.join_events.disconnected();
};
core.add_task(future, options)
}
/// Registers a task whose result is delivered to a cross-thread receiver.
///
/// Arty uses this integration boundary to create a non-[`Send`] future on its destination
/// worker while returning a `Send` result handle to the submitting thread. Keeping the
/// concrete future at this call site lets the executor store it inline instead of behind a
/// `Pin<Box<dyn Future>>`. Obtain `result_tx` from
/// [`remote::Pool::channel`](crate::remote::Pool::channel).
pub fn add_with_remote_result<F, R>(&self, future: F, options: TaskOptions, result_tx: RemoteSender<R>)
where
F: IntoFuture<Output = R> + 'static,
R: Send + 'static,
{
let Some(core) = self.core.upgrade() else {
return;
};
core.add_remote_task(future, options, result_tx);
}
/// Begins executor shutdown without borrowing executor-owned polling state.
pub fn begin_shutdown(&self) {
if let Some(core) = self.core.upgrade() {
core.request_shutdown();
}
}
}
#[cfg(test)]
mod tests {
use static_assertions::assert_not_impl_any;
use super::*;
#[test]
fn thread_safety() {
assert_not_impl_any!(TaskSet: Send, Sync);
}
}