aboutsummaryrefslogtreecommitdiff
path: root/embassy-executor/src
diff options
context:
space:
mode:
authorDániel Buga <[email protected]>2023-08-21 13:55:30 +0200
committerDániel Buga <[email protected]>2023-08-21 18:12:41 +0200
commit0a73c84df0936facecb3e1a97cf6f4795d321b87 (patch)
tree950ab1cbbb07997203fe7e36f3b2ece5e3e8097c /embassy-executor/src
parent96e0ace89e48cffd073749cc3b08835a0a7d6cc9 (diff)
Make AvailableTask public, deduplicate
Diffstat (limited to 'embassy-executor/src')
-rw-r--r--embassy-executor/src/raw/mod.rs113
-rw-r--r--embassy-executor/src/spawner.rs3
2 files changed, 65 insertions, 51 deletions
diff --git a/embassy-executor/src/raw/mod.rs b/embassy-executor/src/raw/mod.rs
index 7caa3302f..c1d82e18a 100644
--- a/embassy-executor/src/raw/mod.rs
+++ b/embassy-executor/src/raw/mod.rs
@@ -147,10 +147,7 @@ impl<F: Future + 'static> TaskStorage<F> {
147 pub fn spawn(&'static self, future: impl FnOnce() -> F) -> SpawnToken<impl Sized> { 147 pub fn spawn(&'static self, future: impl FnOnce() -> F) -> SpawnToken<impl Sized> {
148 let task = AvailableTask::claim(self); 148 let task = AvailableTask::claim(self);
149 match task { 149 match task {
150 Some(task) => { 150 Some(task) => task.initialize(future),
151 let task = task.initialize(future);
152 unsafe { SpawnToken::<F>::new(task) }
153 }
154 None => SpawnToken::new_failed(), 151 None => SpawnToken::new_failed(),
155 } 152 }
156 } 153 }
@@ -186,12 +183,16 @@ impl<F: Future + 'static> TaskStorage<F> {
186 } 183 }
187} 184}
188 185
189struct AvailableTask<F: Future + 'static> { 186/// An uninitialized [`TaskStorage`].
187pub struct AvailableTask<F: Future + 'static> {
190 task: &'static TaskStorage<F>, 188 task: &'static TaskStorage<F>,
191} 189}
192 190
193impl<F: Future + 'static> AvailableTask<F> { 191impl<F: Future + 'static> AvailableTask<F> {
194 fn claim(task: &'static TaskStorage<F>) -> Option<Self> { 192 /// Try to claim a [`TaskStorage`].
193 ///
194 /// This function returns `None` if a task has already been spawned and has not finished running.
195 pub fn claim(task: &'static TaskStorage<F>) -> Option<Self> {
195 task.raw 196 task.raw
196 .state 197 .state
197 .compare_exchange(0, STATE_SPAWNED | STATE_RUN_QUEUED, Ordering::AcqRel, Ordering::Acquire) 198 .compare_exchange(0, STATE_SPAWNED | STATE_RUN_QUEUED, Ordering::AcqRel, Ordering::Acquire)
@@ -199,12 +200,56 @@ impl<F: Future + 'static> AvailableTask<F> {
199 .map(|_| Self { task }) 200 .map(|_| Self { task })
200 } 201 }
201 202
202 fn initialize(self, future: impl FnOnce() -> F) -> TaskRef { 203 fn initialize_impl<S>(self, future: impl FnOnce() -> F) -> SpawnToken<S> {
203 unsafe { 204 unsafe {
204 self.task.raw.poll_fn.set(Some(TaskStorage::<F>::poll)); 205 self.task.raw.poll_fn.set(Some(TaskStorage::<F>::poll));
205 self.task.future.write(future()); 206 self.task.future.write(future());
207
208 let task = TaskRef::new(self.task);
209
210 SpawnToken::new(task)
206 } 211 }
207 TaskRef::new(self.task) 212 }
213
214 /// Initialize the [`TaskStorage`] to run the given future.
215 pub fn initialize(self, future: impl FnOnce() -> F) -> SpawnToken<F> {
216 self.initialize_impl::<F>(future)
217 }
218
219 /// Initialize the [`TaskStorage`] to run the given future.
220 ///
221 /// # Safety
222 ///
223 /// `future` must be a closure of the form `move || my_async_fn(args)`, where `my_async_fn`
224 /// is an `async fn`, NOT a hand-written `Future`.
225 #[doc(hidden)]
226 pub unsafe fn __initialize_async_fn<FutFn>(self, future: impl FnOnce() -> F) -> SpawnToken<FutFn> {
227 // When send-spawning a task, we construct the future in this thread, and effectively
228 // "send" it to the executor thread by enqueuing it in its queue. Therefore, in theory,
229 // send-spawning should require the future `F` to be `Send`.
230 //
231 // The problem is this is more restrictive than needed. Once the future is executing,
232 // it is never sent to another thread. It is only sent when spawning. It should be
233 // enough for the task's arguments to be Send. (and in practice it's super easy to
234 // accidentally make your futures !Send, for example by holding an `Rc` or a `&RefCell` across an `.await`.)
235 //
236 // We can do it by sending the task args and constructing the future in the executor thread
237 // on first poll. However, this cannot be done in-place, so it'll waste stack space for a copy
238 // of the args.
239 //
240 // Luckily, an `async fn` future contains just the args when freshly constructed. So, if the
241 // args are Send, it's OK to send a !Send future, as long as we do it before first polling it.
242 //
243 // (Note: this is how the generators are implemented today, it's not officially guaranteed yet,
244 // but it's possible it'll be guaranteed in the future. See zulip thread:
245 // https://rust-lang.zulipchat.com/#narrow/stream/187312-wg-async/topic/.22only.20before.20poll.22.20Send.20futures )
246 //
247 // The `FutFn` captures all the args, so if it's Send, the task can be send-spawned.
248 // This is why we return `SpawnToken<FutFn>` below.
249 //
250 // This ONLY holds for `async fn` futures. The other `spawn` methods can be called directly
251 // by the user, with arbitrary hand-implemented futures. This is why these return `SpawnToken<F>`.
252 self.initialize_impl::<FutFn>(future)
208 } 253 }
209} 254}
210 255
@@ -223,6 +268,13 @@ impl<F: Future + 'static, const N: usize> TaskPool<F, N> {
223 } 268 }
224 } 269 }
225 270
271 fn spawn_impl<T>(&'static self, future: impl FnOnce() -> F) -> SpawnToken<T> {
272 match self.pool.iter().find_map(AvailableTask::claim) {
273 Some(task) => task.initialize_impl::<T>(future),
274 None => SpawnToken::new_failed(),
275 }
276 }
277
226 /// Try to spawn a task in the pool. 278 /// Try to spawn a task in the pool.
227 /// 279 ///
228 /// See [`TaskStorage::spawn()`] for details. 280 /// See [`TaskStorage::spawn()`] for details.
@@ -231,14 +283,7 @@ impl<F: Future + 'static, const N: usize> TaskPool<F, N> {
231 /// is currently free. If none is free, a "poisoned" SpawnToken is returned, 283 /// is currently free. If none is free, a "poisoned" SpawnToken is returned,
232 /// which will cause [`Spawner::spawn()`](super::Spawner::spawn) to return the error. 284 /// which will cause [`Spawner::spawn()`](super::Spawner::spawn) to return the error.
233 pub fn spawn(&'static self, future: impl FnOnce() -> F) -> SpawnToken<impl Sized> { 285 pub fn spawn(&'static self, future: impl FnOnce() -> F) -> SpawnToken<impl Sized> {
234 let task = self.pool.iter().find_map(AvailableTask::claim); 286 self.spawn_impl::<F>(future)
235 match task {
236 Some(task) => {
237 let task = task.initialize(future);
238 unsafe { SpawnToken::<F>::new(task) }
239 }
240 None => SpawnToken::new_failed(),
241 }
242 } 287 }
243 288
244 /// Like spawn(), but allows the task to be send-spawned if the args are Send even if 289 /// Like spawn(), but allows the task to be send-spawned if the args are Send even if
@@ -254,40 +299,8 @@ impl<F: Future + 'static, const N: usize> TaskPool<F, N> {
254 where 299 where
255 FutFn: FnOnce() -> F, 300 FutFn: FnOnce() -> F,
256 { 301 {
257 // When send-spawning a task, we construct the future in this thread, and effectively 302 // See the comment in AvailableTask::__initialize_async_fn for explanation.
258 // "send" it to the executor thread by enqueuing it in its queue. Therefore, in theory, 303 self.spawn_impl::<FutFn>(future)
259 // send-spawning should require the future `F` to be `Send`.
260 //
261 // The problem is this is more restrictive than needed. Once the future is executing,
262 // it is never sent to another thread. It is only sent when spawning. It should be
263 // enough for the task's arguments to be Send. (and in practice it's super easy to
264 // accidentally make your futures !Send, for example by holding an `Rc` or a `&RefCell` across an `.await`.)
265 //
266 // We can do it by sending the task args and constructing the future in the executor thread
267 // on first poll. However, this cannot be done in-place, so it'll waste stack space for a copy
268 // of the args.
269 //
270 // Luckily, an `async fn` future contains just the args when freshly constructed. So, if the
271 // args are Send, it's OK to send a !Send future, as long as we do it before first polling it.
272 //
273 // (Note: this is how the generators are implemented today, it's not officially guaranteed yet,
274 // but it's possible it'll be guaranteed in the future. See zulip thread:
275 // https://rust-lang.zulipchat.com/#narrow/stream/187312-wg-async/topic/.22only.20before.20poll.22.20Send.20futures )
276 //
277 // The `FutFn` captures all the args, so if it's Send, the task can be send-spawned.
278 // This is why we return `SpawnToken<FutFn>` below.
279 //
280 // This ONLY holds for `async fn` futures. The other `spawn` methods can be called directly
281 // by the user, with arbitrary hand-implemented futures. This is why these return `SpawnToken<F>`.
282
283 let task = self.pool.iter().find_map(AvailableTask::claim);
284 match task {
285 Some(task) => {
286 let task = task.initialize(future);
287 unsafe { SpawnToken::<FutFn>::new(task) }
288 }
289 None => SpawnToken::new_failed(),
290 }
291 } 304 }
292} 305}
293 306
diff --git a/embassy-executor/src/spawner.rs b/embassy-executor/src/spawner.rs
index 2b6224045..5a3a0dee1 100644
--- a/embassy-executor/src/spawner.rs
+++ b/embassy-executor/src/spawner.rs
@@ -33,7 +33,8 @@ impl<S> SpawnToken<S> {
33 } 33 }
34 } 34 }
35 35
36 pub(crate) fn new_failed() -> Self { 36 /// Return a SpawnToken that represents a failed spawn.
37 pub fn new_failed() -> Self {
37 Self { 38 Self {
38 raw_task: None, 39 raw_task: None,
39 phantom: PhantomData, 40 phantom: PhantomData,