···82828383New async primitives that disallow intra-task concurrency, clone of `futures` and `futures-concurrency` for the new primitives.
84848585-## TODO:
8686-- [x] ScopedFuture
8787-- [ ] static combinators (Join Race etc), see futures-concurrency
8888-- [ ] `#[bsync]` or some compiler ScopedFuture generation
8989-- [ ] growable combinators (eg. `FutureGroup`, `FuturesUnordered`) (require alloc?)
9090-- [ ] unsound (needs `Forget`) multithreading
9191-- [ ] "rethinking async rust"
9292-- [ ] all of the above for streams
9393-- [ ] rfc?
94859586channels: need lifetimed receievers, probably needs `Forget` (arc-like channels would be unsafe)
9687···141132- need tons of interior mutability, since immutable/can't move means `poll` cannot take `&mut self`, cells everywhere
142133 - nvm lots of unsafe code, but nothing really unsound
143134- potentially bad error messages? stuff like `join!` will have to output code that manually sets up the waker self ref
135135+136136+## TODO:
137137+- [x] ScopedFuture
138138+- [x] static combinators (Join Race etc), see futures-concurrency
139139+- [x] `#[async_scoped]` or some compiler ScopedFuture generation
140140+- [ ] doubly linked list waker registration
141141+- [ ] repeating static time reactors - eg. make event poll every N seconds
142142+- [ ] io uring reactors
143143+- [ ] growable combinators (eg. `FutureGroup`, `FuturesUnordered`) (require alloc?)
144144+- [ ] unsound (needs `Forget`) multithreading
145145+- [ ] "rethinking async rust"
146146+- [ ] all of the above for streams
147147+- [ ] rfc?
148148+149149+# Chapter 3
150150+151151+man this really sucks
152152+153153+i need better things to do
154154+155155+issues from ch 2
156156+- works great[*]
157157+- *: incompatible with event loop - something still has to poll
158158+ - we're back to doubly linked list of waker registration in event loop
159159+ - this requires Forget
160160+- ScopedFuture - Future interop sucks
161161+162162+163163+structured concurrency with regular combinators:
164164+- scope holds tasks
165165+- scope cancels tasks when dropped
166166+- tasks are ran by central executor
167167+168168+pure structured concurrency with borrow checker:
169169+- high level block_on(), any task wake wakes every level up
170170+- tasks have events
171171+172172+how do the tasks register to an event loop? they don't fuck
173173+174174+175175+```rust
176176+struct Task<F: Future> {
177177+ inner: F,
178178+ prev: *const Task,
179179+ next: *const Task,
180180+ waker: Waker,
181181+}
182182+```
183183+184184+&waker is passed into all sub-tasks, calling wake clone etc panics!!!
185185+this is pretty jank
186186+187187+also waker doesn't have a lifetime so a safe code could easily register to external source that outlives the task
188188+this is unsound
189189+190190+we need
191191+```rust
192192+struct WakerRegister {
193193+ prev: *const WakerRegister,
194194+ next: *const WakerRegister,
195195+}
196196+```
197197+198198+# Borrow Checked Structured Concurrency
199199+200200+An async system needs to have the following components:
201201+202202+- event loop : polls events and schedules tasks when they are ready
203203+- tasks : state machines that progress and can await events from the event loop
204204+- task combinators : tasks that compose other tasks into useful logic structures
205205+206206+Tasks will register themselves to events on the event loop, which will need to outlive tasks and register pointers to wake the tasks, so that they can again be polled.
207207+This is incompatible with the borrow checker because the task pointers (wakers) are being stored inside an event loop with a lifetime that may exceed the tasks'.
208208+209209+`Waker` is effectively a `*const dyn Wake`. It is implemented using a custom `RawWakerVTable` rather than a `dyn` ptr to allow for `Wake::wake(self)`, which is not object safe.
210210+This method is necessary for runtime implementations that rely on the wakers to be effectively `Arc<Task>`, since `wake(self)` consumes `self` and decrements the reference count.
211211+212212+There are two types of sound async runtimes that currently exist in rust:
213213+214214+[tokio](https://github.com/tokio-rs) and [smol](https://github.com/smol-rs) work using the afformentioned reference counting system to ensure wakers aren't dangling pointers to tasks that no longer exist.
215215+216216+[embassy](https://github.com/embassy-rs/embassy) and [rtic](https://github.com/rtic-rs/rtic) work by ensuring tasks are stored in `static` task pools for `N` tasks. Scheduled tasks are represented by an intrusively linked list to avoid allocation, and wakers can't be dangling pointers because completed tasks will refuse to add themselves back to the linked list, or will be replaced by a new task. This is useful in environments where it is desirable to avoid heap allocation, but requires the user annotate the maximum number of a specific task that can exist at one time, and fails to spawn tasks when they exceed that limit.
217217+218218+An async runtime where futures are allocated to the stack cannot be sound under this model because `Future::poll` allows any safe `Future` implementation to store or clone wakers wherever they want, which become dangling pointers after the stack allocated future goes out of scope. In order to prevent this, we must have a circular reference, where the task (`Task<'scope>: Wake`) contains a `&'scope Waker` and the `Waker` contains `*const dyn Wake`. For that to be safe, the `Waker` must never be moved. This cannot be possible because something needs to register the waker:
219219+220220+```rust
221221+// waker is moved
222222+poll(self: Pin<&mut Self>, cx: &mut Context<'_> /* '_ is the duration of the poll fn */) -> Poll<Self::Output> {
223223+ // waker is moved!
224224+ // can't register &Waker bc we run into the same problem,
225225+ // and the waker only lives for '_
226226+ store.register(cx.waker())
227227+}
228228+```
229229+230230+Effectively, by saying our waker can't move, we are saying it must be stored by the task, which means it can't be a useful waker. Instead, what we could do is have a waker-register (verb, not noun) that facilitates the binding of an immovable waker to an immovable task, where the waker is guaranteed to outlive the task:
231231+232232+```rust
233233+pub trait Wake<'task> {
234234+ fn wake(&self);
235235+ fn register_waker(&mut self, waker: &'task Waker);
236236+}
237237+238238+pub struct Waker {
239239+ task: *const dyn Wake,
240240+ valid: bool, // task is in charge of invalidating when it goes out of scope
241241+}
242242+243243+pub struct WakerRegistration<'poll> {
244244+ task: &'poll mut dyn Wake,
245245+}
246246+247247+impl<'poll> WakerRegistration<'poll> {
248248+ pub fn register<'task>(self, slot: &'task Waker)
249249+ where
250250+ 'task: 'poll,
251251+ Self: 'task
252252+ {
253253+ *slot = Waker::new(self.task as *const dyn Wake);
254254+ *task.register_waker(slot)
255255+ }
256256+}
257257+```
258258+259259+This system works better because `WakerRegistration` only lives
260260+261261+262262+263263+Experienced rust programmers might be reading this and thinking I am stupid (true) because `Forget`
264264+265265+An astute (soon to be disappointed) reader might be thinking, as I did when I first learned about this sytem, "what if we ensured that there was only one `Waker` per `Task`, and gave the task a pointer to the waker, so that it could disable the waker when dropped?"
266266+267267+Unfortunately, there are a multitude of issues with this system
268268+269269+- In order to hold a pointer to the Waker from the
270270+- Preventing a `Waker` from moving means panicking on the
271271+272272+Even if it was guaranteed that wakers could not be moved or `cloned` (by panicking on `clone`), and registration occured via `&Waker`, the task would still be unable to
273273+274274+https://conradludgate.com/posts/async-stack
275275+276276+## Structured Concurrency
277277+278278+https://trio.discourse.group/t/discussion-notes-on-structured-concurrency-or-go-statement-considered-harmful/25
279279+https://kotlinlang.org/docs/coroutines-basics.html
280280+281281+Notably, the structured concurrency pattern fits very nicely with our hypothetical unsound stack based async runtime.
282282+283283+## WeakCell Pattern & Forget trait
284284+285285+https://github.com/rust-lang/rfcs/pull/3782
286286+287287+288288+There are two solutions:
289289+290290+- `Wakers` panic on `clone()`
291291+292292+## Waker allocation problem & intra task concurrency
293293+294294+we can't do intra task concurrency because WeakRegistrations
295295+296296+
+4-10
futures-compat/src/lib.rs
···1313//! that expects it to live for 'scope, and then the ScopedFutureWrapper is
1414//! dropped
1515//!
1616-//!
1717-//! ## TRIGGER WARNING
1818-//!
1916//! This code is not for the faint of heart. Read at your own risk.
20172118use std::{
···3431/// can transmute between them, but the waker will be completely invalid!
35323633/// wraps an internal ScopedFuture, implements Future
3737-pub struct ScopedFutureWrapper<'scope, F: ScopedFuture<'scope> + 'scope> {
3434+pub struct ScopedFutureWrapper<'scope, F: ScopedFuture<'scope>> {
3835 inner: UnsafeCell<F>,
3936 marker: PhantomData<&'scope ()>,
4037}
···4441{
4542 type Output = F::Output;
46434747- fn poll(
4848- self: std::pin::Pin<&mut Self>,
4949- cx: &mut std::task::Context<'_>,
5050- ) -> Poll<Self::Output> {
4444+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
5145 // # Safety
5246 //
5347 // Transmutes `Waker` into `&'scope dyn Wake`.
···10296 marker: PhantomData<&'scope ()>,
10397}
10498105105-impl<'scope, F: Future> ScopedFuture<'scope>
9999+impl<'scope, F: Future + 'scope> ScopedFuture<'scope>
106100 for UnscopedFutureWrapper<'scope, F>
107101{
108102 type Output = F::Output;
···140134 }
141135}
142136143143-impl<'scope, F: Future> UnscopedFutureWrapper<'scope, F> {
137137+impl<'scope, F: Future + 'scope> UnscopedFutureWrapper<'scope, F> {
144138 pub unsafe fn from_future(f: F) -> Self {
145139 Self {
146140 inner: f.into(),
+3-2
futures-core/src/lib.rs
···3535/// The waker is no longer tied to the actual future's lifetime, making it
3636/// unsound to not have either static tasks or reference counting.
3737/// To avoid this, we want to use a &'scope waker instead, with 1 waker / task.
3838+#[must_use = "futures do nothing unless you `.await` or poll them"]
3839#[diagnostic::on_unimplemented(
3939- message = "The type `{Self}` must implement the `ScopedFuture` trait.",
4040- label = "Missing `ScopedFuture` implementation",
4040+ message = "`{Self}` is not a `ScopedFuture`",
4141+ label = "`{Self}` is not a `ScopedFuture`",
4142 note = "If you are trying to await a `task::Future` from within a `ScopedFuture`, note that the systems are incompatible."
4243)]
4344pub trait ScopedFuture<'scope> {
+49-5
futures-core/src/task.rs
···11-/// A task that can be woken.
22-///
33-/// This acts as a handle for a reactor to indicate when a `ScopedFuture` is
44-/// once again ready to be polled.
55-pub trait Wake<'scope> {
11+use std::{mem, ptr::NonNull, sync::atomic::AtomicPtr};
22+33+// Task: Wake
44+//
55+// Wake must not outlive event loop/storage
66+pub trait Wake<'events> {
67 fn wake(&self);
88+ // task can't outlive event loop -> holds &'events
99+ fn register_waker(&self, waker: &'events Waker);
1010+}
1111+1212+// type Waker<'events> = Option<NonNull<dyn Wake<'events>>>;
1313+1414+pub struct Waker<'events> {
1515+ task: *const dyn Wake<'events>,
1616+}
1717+1818+// wakers must outlive 'task
1919+impl<'events> Waker<'events> {
2020+ pub fn new(task: *const dyn Wake<'events>) -> Self {
2121+ Self { task: task.into() }
2222+ }
2323+2424+ pub fn wake(self) {
2525+ unsafe { self.task.as_ref() }.inspect(|task| task.wake());
2626+ }
2727+}
2828+2929+pub struct WakerRegistration<'events> {
3030+ task: &'events dyn Wake<'events>, // valid for all of 'events
3131+}
3232+3333+impl<'events> WakerRegistration<'events> {
3434+ // slot is valid for all 'events
3535+ pub fn register(self, slot: &'events mut Waker<'events>) {
3636+ // Cast from 'events to 'static
3737+ //
3838+ // # Safety
3939+ //
4040+ // This is safe because the drop guard guarantees that the task ptr (which lives for static)
4141+ // becomes null when the wake is dropped, ensuring the dangling pointer is never dereferenced.
4242+ let dangling_task = unsafe {
4343+ mem::transmute::<&'events dyn Wake<'events>, *const dyn Wake<'events>>(
4444+ self.task,
4545+ )
4646+ };
4747+ slot.task = dangling_task;
4848+4949+ (*self.task).register_waker(slot);
5050+ }
751}
+107-56
futures-derive/src/lib.rs
···11use proc_macro::TokenStream;
22use quote::{ToTokens, quote};
33use syn::{
44- Expr, ExprAwait, FnArg, GenericArgument, ItemFn, Pat, ReturnType,
55- Signature, parse_macro_input, visit_mut::VisitMut,
44+ Expr, ExprAwait, FnArg, GenericArgument, ItemFn, ReturnType,
55+ parse_macro_input, parse_quote, parse2, visit_mut::VisitMut,
66};
7788+/// Takes async fn that returns anonymous `Future` impl.
99+/// Generates fn that returns `UnscopedFutureWrapper` wrapper for the the anonymous `Future` impl.
1010+///
1111+/// ```rust
1212+/// fn my_func<'a, 'b>(a: &'a A, b: &'b B) -> impl ScopedFuture<'a + 'b, Output = T> + 'a + 'b {
1313+/// let output = async move { [body] } // compilers turns this into -> impl Future<Output = T> + 'a + 'b
1414+/// unsafe { UnscopedFutureWrapper::from_future(output) }
1515+/// }
1616+/// ```
1717+///
1818+/// see https://rust-lang.github.io/rfcs/2394-async_await.html#lifetime-capture-in-the-anonymous-future
1919+/// for more context on lifetime capture
2020+/// - resulting ScopedFuture needs to be constrained to not outlive the lifetimes of any references
2121+///
2222+/// to actually implement this (capture all lifetimes) we use `ScopedFuture<'_> + '_` so the compiler can infer
2323+/// lifetimes from the anonymous future impl returned by the actual inner async fn
824#[proc_macro_attribute]
925pub fn async_scoped(_: TokenStream, item: TokenStream) -> TokenStream {
1010- let mut input = parse_macro_input!(item as ItemFn);
1111-2626+ let mut item_fn = parse_macro_input!(item as ItemFn);
1227 // Wraps *every* async expression within the function block with
1328 // `ScopedFutureWrapper`, allowing them to be treated as regular `Future`
1429 // impls.
···1631 // This will cause a compiler error if any expression being awaited is not
1732 // a `ScopedFuture`, which is intentional because the `Future` and
1833 // `ScopedFuture` systems are incompatible.
1919- ScopedFutureWrappingVisitor.visit_item_fn_mut(&mut input);
3434+ ScopedFutureWrappingVisitor.visit_item_fn_mut(&mut item_fn);
20352121- // Wrap the function with `UnscopedFutureWrapper` to convert it back into
2222- // a `ScopedFuture`.
2323- wrap_async_with_scoped(&input).into()
3636+ // disable async since it is moved to the block
3737+ item_fn.sig.asyncness = None;
3838+3939+ // wrap block with UnscopedFutureWrapper
4040+ let block = *item_fn.block;
4141+ *item_fn.block = parse_quote! {
4242+ {
4343+ let future = async move #block;
4444+ unsafe { futures_compat::UnscopedFutureWrapper::from_future(future) }
4545+ }
4646+ };
4747+4848+ let output = match &item_fn.sig.output {
4949+ ReturnType::Default => quote! { () },
5050+ ReturnType::Type(_, ty) => quote! { #ty },
5151+ };
5252+5353+ let has_lifetime_dependency =
5454+ item_fn.sig.inputs.iter().any(|param| match param {
5555+ FnArg::Receiver(receiver) => receiver.reference.is_some(),
5656+ FnArg::Typed(pat) => has_lifetime_dependency(&pat.ty),
5757+ });
5858+5959+ // set outer fn output to ScopedFuture<'_/'static, Output = #output>
6060+ item_fn.sig.output = if has_lifetime_dependency {
6161+ parse_quote! { -> impl futures_core::ScopedFuture<'_, Output = #output> + '_ }
6262+ } else {
6363+ parse_quote! { -> impl futures_core::ScopedFuture<'static, Output = #output> }
6464+ };
6565+6666+ item_fn.to_token_stream().into()
2467}
25686969+/// This currently is impossible to do the `futures_compat` workarounds not
7070+/// being compatible with closures.
7171+///
2672/// Takes async fn that returns anonymous `Future` impl.
2773/// Generates fn that returns `UnscopedFutureWrapper` wrapper for the the anonymous `Future` impl.
2874///
···3985///
4086/// to actually implement this (capture all lifetimes) we use `ScopedFuture<'_> + '_` so the compiler can infer
4187/// lifetimes from the anonymous future impl returned by the actual inner async fn
4242-fn wrap_async_with_scoped(
4343- ItemFn {
4444- attrs,
4545- vis,
4646- sig:
4747- Signature {
4848- constness,
4949- unsafety,
5050- ident,
5151- generics,
5252- inputs,
5353- output,
5454- ..
5555- },
5656- block,
5757- }: &ItemFn,
5858-) -> proc_macro2::TokenStream {
5959- let output = match output {
6060- ReturnType::Default => quote! { () },
6161- ReturnType::Type(_, ty) => quote! { #ty },
6262- };
8888+// #[proc_macro]
8989+// pub fn closure(input: TokenStream) -> TokenStream {
9090+// // let ExprClosure {
9191+// // attrs,
9292+// // lifetimes,
9393+// // constness,
9494+// // movability,
9595+// // capture,
9696+// // inputs,
9797+// // output,
9898+// // body,
9999+// // ..
100100+// // } = parse_macro_input!(input as ExprClosure);
101101+// let mut closure = parse_macro_input!(input as ExprClosure);
102102+// // disable async because we move it to inner
103103+// closure.asyncness = None;
104104+// let body = closure.body;
105105+106106+// // let output = match closure.output {
107107+// // ReturnType::Default => parse_quote! { () },
108108+// // ReturnType::Type(_, ty) => parse_quote! { #ty },
109109+// // };
631106464- let inner_args: Vec<syn::Ident> = inputs
6565- .iter()
6666- .filter_map(|param| match param {
6767- FnArg::Receiver(_) => Some(quote::format_ident!("self")),
6868- FnArg::Typed(typed) => {
6969- if let Pat::Ident(ident) = &*typed.pat {
7070- Some(ident.ident.to_owned())
7171- } else {
7272- None
7373- }
7474- }
7575- })
7676- .collect();
111111+// // let outer_output =
112112+// // parse_quote! { futures_core::ScopedFuture<'_, Output = #output> + '_ };
771137878- let has_lifetime_dependency = inputs.iter().any(|param| match param {
7979- FnArg::Receiver(receiver) => receiver.reference.is_some(),
8080- FnArg::Typed(pat) => has_lifetime_dependency(&pat.ty),
8181- });
114114+// closure.body = parse_quote! {{
115115+// let output = async move { #body };
116116+// unsafe { futures_compat::UnscopedFutureWrapper::from_future(output) }
117117+// }};
118118+// // closure.output = outer_output;
119119+// closure.to_token_stream().into()
120120+// }
821218383- let outer_output = if has_lifetime_dependency {
8484- quote! { futures_core::ScopedFuture<'_, Output = #output> + '_ }
8585- } else {
8686- quote! { futures_core::ScopedFuture<'static, Output = #output> }
8787- };
122122+/// Wraps a block of optionally async statements and expressions in an anonymous `ScopedFuture` impl.
123123+///
124124+/// This generates a modified block of the form:
125125+///
126126+/// ```rust
127127+/// {
128128+/// let output = async { <original block, mapped to convert all `ScopedFuture` to `Future`> };
129129+/// unsafe { futures_compat::UnscopedFutureWrapper::from_future(output) }
130130+/// }
131131+/// ```
132132+#[proc_macro]
133133+pub fn block(input: TokenStream) -> TokenStream {
134134+ // block is formed { **expr/stmt }, so we need to surround the inputs in {}
135135+ let input = proc_macro2::TokenStream::from(input);
136136+ let block_input = quote! { { #input } };
881378989- quote! {
9090- #(#attrs)* #vis #constness #unsafety fn #ident #generics (#inputs) -> impl #outer_output {
9191- async #constness #unsafety fn __inner (#inputs) -> #output #block
138138+ let mut block = parse2(block_input).expect("Failed to parse as block.");
921399393- let future = __inner(#(#inner_args),*);
140140+ ScopedFutureWrappingVisitor.visit_block_mut(&mut block);
941419595- unsafe { futures_compat::UnscopedFutureWrapper::from_future(future) }
142142+ quote! {
143143+ {
144144+ let output = async #block;
145145+ unsafe { futures_compat::UnscopedFutureWrapper::from_future(output) }
96146 }
97147 }
148148+ .into()
98149}
99150100151/// Determines if typed pattern contains a reference or dependency on a
+2-14
futures/src/lib.rs
···11#![no_std]
2233pub use futures_combinators;
44-use futures_compat::{ScopedFutureWrapper, UnscopedFutureWrapper};
54pub use futures_core;
65use futures_core::ScopedFuture;
76pub use futures_derive::async_scoped;
···1615}
17161817#[async_scoped]
1919-fn test(a: i32, b: &i32) -> () {
2020- // evil().await;
2121- let x = inner(a, &b).await;
2222- // async {}.await;
2323-2424- let test_block = futures_derive::block! { 1 + 1; 2 }.await;
2525-2626- // let test_closure = futures_derive::closure! { |&ab, &cd| ab + cd };
2727-2828- // let asdf = futures_derive::closure! { |a: &i32| {
2929- // *a + b
3030- // }};
3131- // let x = asdf(&a).await;
1818+fn test(a: i32, b: &i32) -> i32 {
1919+ futures_derive::block! { 1 + *b; 2 }.await
3220}
33213422fn test2<'a>(a: i32) {}