align multithreading and trap behavior with CM spec - #14146
Conversation
ae560e3 to
8506000
Compare
This updates Wasmtime's Component Model async and cooperative multithreading support to match the current specification, including: - Refined rules for trapping when a sync-typed function blocks. We now enforce this "lazily" rather than "eagerly", mwaning a sync-typed function is allowed to call an async-typed function or blocking intrinsic, and if it doesn't actually block, we won't trap. And if the call _does_ block, we will look for any eligible threads to run and run them until no such threads remain, only trapping if and when we still need to block and have no more threads to run. - Ensure that the predicate for determining which threads can be run when a sync-typed function is executing in an instance matches the spec. - Remove the previous "may block" bookkeeping at the task and root instance level, replacing it with (sub-)instance level tracking of whether any sync-typed function is running in that instance. - Run the event loop during start function calls since they are now allowed to call async-typed functions, create and resume threads, etc. Note that this includes `test/component-model` submodule updates which haven't yet been merged to the main branch of the upstream repo, but should be merged soon. See WebAssembly/component-model#696 Fixes bytecodealliance#14117 Co-authored-by: Alex Crichton <alex@alexcrichton.com>
8506000 to
5d88b4c
Compare
alexcrichton
left a comment
alexcrichton
left a comment
There was a problem hiding this comment.
I'd like to review more of concurrent.rs but here's some initial thoughts. It's at the point where whenever I expand context on github it just sends me randomly elsewhere in this diff and I keep losing my spot in the otherwise big diff in concurrent.rs. I'm hoping my changes in dicej#7 which reduce the number of files changed helps with that...
| // If there are any tasks belonging to | ||
| // an instance which may not suspend, trap with | ||
| // `CannotBlockSyncTask`: | ||
| let instances = reset | ||
| .store | ||
| .0 | ||
| .concurrent_state_mut()? | ||
| .table | ||
| .get_mut() | ||
| .iter_mut() | ||
| .filter_map(|entry| { | ||
| if let Some(task) = entry.downcast_ref::<GuestTask>() { | ||
| Some(task.instance) | ||
| } else { | ||
| None | ||
| } | ||
| }) | ||
| .collect::<Vec<_>>(); | ||
|
|
||
| for instance in instances { | ||
| if reset | ||
| .store | ||
| .0 | ||
| .instance_state(instance) | ||
| .concurrent_state() | ||
| .do_not_suspend | ||
| { | ||
| return Poll::Ready(Err(Trap::CannotBlockSyncTask.into())); | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
Stylistically I find this quite hard to read because everything is indented so far over and spread out across so many lines due to rustfmt's formatting. Could this method be split up with some helpers perhaps to reduce indentation? Either that or using some local variables which project out some state to avoid having to re-acquire fields through projection each time?
| // Note that the unsafety here should be ok because the | ||
| // validity of the component means that type-checks have | ||
| // already been performed. This means that the unsafety due | ||
| // to imports having the wrong type should not happen here. | ||
| // | ||
| // Also note we are calling new_started_impl because we have | ||
| // already checked for asyncness and are running on a fiber | ||
| // if required. |
There was a problem hiding this comment.
Can you update this comment while you're here. Notably this comment is intended to be safety-bearing and it's now inaccurate because new_started isn't used and instead it's split up here depending on how things are run.
| where | ||
| T: Send, |
There was a problem hiding this comment.
Personally I still find it valuable to not have this change in the API. I know I'm somewhat swimming upstream but I continue to not want to get into the habit of just readily adding constraints everywhere necessary.
I've sent a PR-to-this-branch as dicej#7 which removes the need for all of the Send changes in this PR. I believe that it's safe to have that PR due to the conditional nature of Send in async and the surrounding context.
| } | ||
| }; | ||
|
|
||
| if instance.id.get_mut(store.0).needs_startup() { |
There was a problem hiding this comment.
Could new_raw return a tuple where one result is "needs start" to avoid duplicating the logic necessary to deduce this?
| if asyncness == Asyncness::No { | ||
| instance.start_raw(store)?; | ||
| } else { | ||
| #[cfg(feature = "async")] | ||
| { | ||
| #[cfg(feature = "component-model-async")] | ||
| { | ||
| if store.0.concurrency_support() { | ||
| // With concurrency support enabled, we must | ||
| // run the start function inside the store's | ||
| // event loop in case it calls async | ||
| // functions or intrinsics, creates and | ||
| // resumes threads, etc. | ||
| let (tx, rx) = oneshot::channel(); | ||
| let token = StoreToken::new(store.as_context_mut()); | ||
| store.0.queue_task(move |store| { | ||
| _ = tx.send( | ||
| instance | ||
| .start_raw(&mut token.as_context_mut(store)) | ||
| .map(|()| instance), | ||
| ); | ||
| Ok(()) | ||
| })?; | ||
| instance = store | ||
| .as_context_mut() | ||
| .run_concurrent_trap_on_idle(async |_| { | ||
| rx.await.map_err(|_| { | ||
| format_err!("oneshot channel canceled") | ||
| }) | ||
| }) | ||
| .await???; | ||
| } else { | ||
| store.on_fiber(|store| instance.start_raw(store)).await??; | ||
| } | ||
| } | ||
| #[cfg(not(feature = "component-model-async"))] | ||
| { | ||
| _ = &mut instance; | ||
| store.on_fiber(|store| instance.start_raw(store)).await??; | ||
| } | ||
| } | ||
| #[cfg(not(feature = "async"))] | ||
| { | ||
| _ = &mut instance; | ||
| unreachable!(); | ||
| } | ||
| } |
There was a problem hiding this comment.
To avoid duplicating code, could this be modeled as:
#[cfg(component-model-async)]
if asyncness != No && store.concurrency_support() {
// do the concurrent thing
} else {
// use `start_raw`
}
#[cfg(not(component-model-async))]
// use start_rawSimilar to elsewhere I'm pretty worried about the extreme rightward drift here as it makes it pretty hard to understand what's actually going on.
Another possible alternative would be to take the body of the concurrent bits here and move them to concurrent.rs as a dedicated function.
| /// | ||
| /// Note that this step needs to be run on a fiber in async mode even | ||
| /// though it doesn't do any blocking work because an async resource | ||
| /// limiter may need to yield. |
There was a problem hiding this comment.
While you're here could you update this comment? Notably there's no longer a split with *_async and the comment about the fiber is no longer correct now that it's natively async
| Ok(false) | ||
| } | ||
|
|
||
| fn enter_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> { |
There was a problem hiding this comment.
Personally I'm pretty wary of adding new hooks like this because we have so many other locations where this is applicable but not otherwise called. For example don't all of these locations need to in theory call this to update the internal flag?
- {Host,FACT}-invoking
cabi_realloc - {Host,FACT}-invoking resource destructors
- FACT component-to-component trampolines
In all of those locations it's effectively sync but the internal do_not_suspend flag isn't updated I think?
Basically I think it's best if we avoid adding state where possible because there's so many places to maintain the state that it's best to lean on preexisting stae if we can. Could we look at the current task or otherwise have some accessor for what the suspending task's type is and/or something like that?
There was a problem hiding this comment.
- {Host,FACT}-invoking
cabi_realloc
cabi_realloc can't call any imports and thus can't suspend, correct?
- {Host,FACT}-invoking resource destructors
Yeah, I'll need to look into this.
- FACT component-to-component trampolines
I actually did a bunch of work for this (e.g. reverting may_leave back to a bitset which has two flags and propagating that change everywhere) and then reverted it all when I realized that the lazy task/thread creation in StoreOpaque::force_deferred_current_thread covered it. Did I miss something?
Could we look at the current task or otherwise have some accessor for what the suspending task's type is and/or something like that?
That's essentially what I did if you look at the places this function is called: query the task for its type and then publish that to the InstanceState so that other concurrent tasks and threads for that instance can see it. Alternatively, other tasks and threads could find out whether they're able to suspend by iterating over all known tasks in the store's table, looking for any that match the instance of interest and checking their type. That would be pretty expensive, though.
2 participants
This updates Wasmtime's Component Model async and cooperative multithreading support to match the current specification, including:
Refined rules for trapping when a sync-typed function blocks. We now enforce this "lazily" rather than "eagerly", mwaning a sync-typed function is allowed to call an async-typed function or blocking intrinsic, and if it doesn't actually block, we won't trap. And if the call does block, we will look for any eligible threads to run and run them until no such threads remain, only trapping if and when we still need to block and have no more threads to run.
Ensure that the predicate for determining which threads can be run when a sync-typed function is executing in an instance matches the spec.
Remove the previous "may block" bookkeeping at the task and root instance level, replacing it with (sub-)instance level tracking of whether any sync-typed function is running in that instance.
Run the event loop during start function calls since they are now allowed to call async-typed functions, create and resume threads, etc.
Note that this includes
test/component-modelsubmodule updates which haven't yet been merged to the main branch of the upstream repo, but should be merged soon. See WebAssembly/component-model#696Fixes #14117