Skip to content

Commit 1b49ff6

Browse files
committed
Add thread.{get,set}-task and task.drop built-ins
1 parent c52323b commit 1b49ff6

10 files changed

Lines changed: 1178 additions & 135 deletions

File tree

‎design/mvp/Binary.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -340,6 +340,9 @@ canon ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
340340
| 0x2b 0x00 => (canon thread.yield-then-resume (core func)) 🧵
341341
| 0x2c 0x00 => (canon thread.suspend-then-promote (core func)) 🧵
342342
| 0x2d 0x00 => (canon thread.yield-then-promote (core func)) 🧵
343+
| 0x30 => (canon thread.set-task (core func)) 🧵
344+
| 0x31 => (canon thread.get-task (core func)) 🧵
345+
| 0x32 => (canon task.drop (core func)) 🧵
343346
| 0x40 shared?:<sh?> ft:<core:typeidx> => (canon thread.spawn-ref shared? ft (core func)) 🧵②
344347
| 0x41 shared?:<sh?> ft:<core:typeidx> tbl:<core:tableidx> => (canon thread.spawn-indirect shared? ft tbl (core func)) 🧵②
345348
| 0x42 shared?:<sh?> => (canon thread.available-parallelism shared? (core func)) 🧵②

‎design/mvp/CanonicalABI.md‎

Lines changed: 129 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,9 @@ specified here.
6565
* [`canon thread.yield-then-resume`](#-canon-threadyield-then-resume) 🧵
6666
* [`canon thread.suspend-then-promote`](#-canon-threadsuspend-then-promote) 🧵
6767
* [`canon thread.yield-then-promote`](#-canon-threadyield-then-promote) 🧵
68+
* [`canon thread.set-task`](#-canon-threadset-task) 🧵
69+
* [`canon thread.get-task`](#-canon-threadget-task) 🧵
70+
* [`canon task.drop`](#-canon-taskdrop) 🧵
6871
* [`canon error-context.new`](#-canon-error-contextnew) 📝
6972
* [`canon error-context.debug-message`](#-canon-error-contextdebug-message) 📝
7073
* [`canon error-context.drop`](#-canon-error-contextdrop) 📝
@@ -119,7 +122,7 @@ into the `handles` or `threads` fields of `ComponentInstance`.
119122
```python
120123
class ComponentInstance:
121124
store: Store
122-
handles: Table[ResourceHandle | Waitable | WaitableSet | ErrorContext]
125+
handles: Table[ResourceHandle | Waitable | WaitableSet | ErrorContext | Task]
123126
threads: Table[Thread]
124127
may_leave: bool
125128
backpressure: int
@@ -609,8 +612,7 @@ class Task:
609612
on_resolve: OnResolve
610613
state: State
611614
num_borrows: int
612-
implicit_thread: Optional[Thread]
613-
threads: list[Thread]
615+
waiting_to_enter: Optional[Thread]
614616

615617
def __init__(self, ft, opts, inst, on_start, on_resolve):
616618
self.ft = ft
@@ -620,22 +622,12 @@ class Task:
620622
self.on_resolve = on_resolve
621623
self.state = Task.State.INITIAL
622624
self.num_borrows = 0
623-
self.implicit_thread = None
624-
self.threads = []
625+
self.waiting_to_enter = None
625626
```
626627

627628
The `Task.needs_exclusive` predicate returns whether this task's implicit thread
628-
(`Task.implicit_thread`) has *not* opted in to multiple concurrent linear memory
629-
shadow stacks (via "stackful" lift) and thus, according to [Component Invariant]
630-
#2, requires serialization with all the other implicit threads in the component
631-
instance that have similarly not opted in. This question only applies to
632-
`async`-typed functions, since synchronous functions can't block and thus can
633-
always execute in a LIFO fashion using a single linear memory shadow stack. When
634-
`needs_exclusive` is true, core wasm execution is gated on acquiring the
635-
`ComponentInstance.exclusive_thread` lock. Due to cooperativity, the
636-
`exclusive_thread` "lock" is simply a mutable field holding either `None`, when
637-
unlocked, or, when locked, a reference to the `Task.implicit_thread` currently
638-
holding the lock.
629+
requires the `ComponentInstance.exclusive_thread` lock to acquired and released
630+
in order to enforce [Component Invariant] #2.
639631
```python
640632
def needs_exclusive(self):
641633
assert(self.ft.async_)
@@ -668,29 +660,24 @@ exports.
668660
```python
669661
def enter_implicit_thread(self):
670662
assert(self.state == Task.State.INITIAL)
671-
self.implicit_thread = current_thread()
672663
if self.ft.async_:
673664
def has_backpressure():
674665
return (self.inst.backpressure > 0 or
675666
(self.needs_exclusive() and self.inst.exclusive_thread is not None))
676667
if has_backpressure() or self.inst.num_waiting_to_enter > 0:
668+
self.waiting_to_enter = current_thread()
677669
self.inst.num_waiting_to_enter += 1
678-
self.implicit_thread.wait_until(lambda: not has_backpressure())
670+
current_thread().wait_until(lambda: not has_backpressure())
679671
self.inst.num_waiting_to_enter -= 1
672+
self.waiting_to_enter = None
680673
if self.deliver_pending_cancel():
681674
self.cancel()
682675
return False
683676
if self.needs_exclusive():
684677
assert(self.inst.exclusive_thread is None)
685-
self.inst.exclusive_thread = self.implicit_thread
686-
self.register_thread(self.implicit_thread)
678+
self.inst.exclusive_thread = current_thread()
679+
current_thread().index = self.inst.threads.add(current_thread())
687680
return True
688-
689-
def register_thread(self, thread):
690-
assert(thread not in self.threads and thread.task is self)
691-
self.threads.append(thread)
692-
assert(thread.index is None)
693-
thread.index = self.inst.threads.add(thread)
694681
```
695682
Since the order in which suspended threads are resumed is nondeterministic (see
696683
`Store.tick` below), once `Task.enter_implicit_thread` suspends the task's
@@ -704,58 +691,48 @@ above definition ensures the following properties:
704691
backpressure (i.e., disabling backpressure never unleashes an unstoppable
705692
thundering herd of pending tasks).
706693

707-
Once a task's implicit thread has cleared the backpressure gate, it is added to
708-
the lists of threads running inside the current task and component instance by
709-
`Task.register_thread()` (which is also called by `thread.new-indirect`, below).
694+
As shown above, only once a task has cleared the backpressure gate is its
695+
implicit thread visibly added to the component-instance-wide `threads` table.
710696

711697
Symmetrically, the `Task.exit_implicit_thread` method is called before a task's
712698
implicit thread returns to reverse the effects of `Task.enter_implicit_thread`.
713-
In particular, if the `exclusive_thread` lock was acquired, it is released.
714-
`Task.unregister_thread` (which is also called by `thread.new-indirect`, below)
715-
traps if the task's last thread is unregistered and the task has not yet
716-
returned a value to its caller.
699+
Note that, due to `thread.set-task`, `current_thread().task` may not be `self`.
700+
However, because threads can't change instances and because `needs_exclusive`
701+
must stay invariant over the lifetime of the thread, the uses of `self` below
702+
are all valid.
717703
```python
718704
def exit_implicit_thread(self):
719-
assert(current_thread() is self.implicit_thread)
720-
self.unregister_thread(self.implicit_thread)
705+
self.inst.threads.remove(current_thread().index)
721706
if self.ft.async_ and self.needs_exclusive():
722-
assert(self.inst.exclusive_thread is self.implicit_thread)
707+
assert(self.inst.exclusive_thread is current_thread())
723708
self.inst.exclusive_thread = None
724-
725-
def unregister_thread(self, thread):
726-
assert(thread in self.threads and thread.task is self)
727-
self.threads.remove(thread)
728-
if len(self.threads) == 0:
729-
trap_if(self.state != Task.State.RESOLVED)
730-
assert(self.num_borrows == 0)
731-
assert(thread.index is not None)
732-
self.inst.threads.remove(thread.index)
733709
```
734710

735711
The `Task.request_cancellation` method implements the `OnCancel` callback that
736712
is called by the `subtask.cancel` built-in. If a task's implicit thread is
737713
waiting to start (in `Task.enter_implicit_thread`, defined above) due to
738714
backpressure, then it is immediately cancelled without running any guest code.
739-
Otherwise, if the task's implicit thread is `cancellable` and `ready`, it is
740-
resumed, allowing it to execute until returning or blocking. Currently, only
741-
`callback` threads that have returned to their event loop are `cancellable`.
742-
When resumed, the `callback` thread will call `Task.deliver_pending_cancel`,
743-
which will return `True`, leading to the `TASK_CANCELLED` event code being
744-
passed to the `callback` function. Ideally, the `callback` function will
745-
promptly call `task.cancel` (or `task.return`) to resolve the task before the
746-
`OnCancel` callback returns, allowing `subtask.cancel` to return without
747-
blocking.
715+
Otherwise, if the task contains one or more threads that are `cancellable` and
716+
`ready`, one nondeterministically picked and resumed to allow the task execute
717+
until returning or blocking. Currently, only `callback` threads that have
718+
returned to their event loop are `cancellable`. When resumed, the `callback`
719+
thread will call `Task.deliver_pending_cancel`, which will return `True`,
720+
leading to the `TASK_CANCELLED` event code being passed to the `callback`
721+
function. Ideally, the `callback` function will promptly call `task.cancel` (or
722+
`task.return`) to resolve the task before the `OnCancel` callback returns,
723+
allowing `subtask.cancel` to return without blocking.
748724
```python
749725
def request_cancellation(self):
750726
if self.state == Task.State.INITIAL:
751727
self.state = Task.State.PENDING_CANCEL
752-
self.implicit_thread.resume()
728+
self.waiting_to_enter.resume()
753729
assert(self.state == Task.State.RESOLVED)
754730
else:
755731
assert(self.state == Task.State.STARTED)
756732
self.state = Task.State.PENDING_CANCEL
757-
if self.implicit_thread.cancellable and self.implicit_thread.ready():
758-
self.implicit_thread.resume()
733+
candidates = { t for t in self.inst.threads if t.task is self and t.cancellable and t.ready() }
734+
if candidates:
735+
random.choice(list(candidates)).resume()
759736

760737
def has_pending_cancel(self):
761738
return self.state == Task.State.PENDING_CANCEL
@@ -3367,17 +3344,22 @@ their arguments were lowered or not.
33673344
assert(types_match_values(flat_ft.params, flat_args))
33683345
```
33693346

3370-
If the `async` `canonopt` is *not* specified, a `lift`ed function then calls
3371-
the core wasm callee, passing the lowered arguments in core function parameters
3372-
and receiving the return value as core function results. Once the core results
3373-
are lifted according to `lift_flat_values` above, the optional `post-return`
3374-
function (specified as a `canonopt` immediate of `canon lift`) is called,
3375-
passing the same core wasm results as parameters so that the `post-return`
3376-
function can free any associated allocations.
3347+
If the `async` `canonopt` is *not* specified, a `lift`ed function then calls the
3348+
core wasm callee, passing the lowered arguments in core function parameters and
3349+
receiving the return value as core function results. While executing, a sync
3350+
implicit thread may *temporarily* change task membership (e.g., when servicing
3351+
waitables associated with other tasks), but by the time the thread returns, it
3352+
must have switched back to its original task or else there is a trap (to ensure
3353+
that neither fibers nor dynamic return-type checks are required for sync-to-sync
3354+
calls). Once the core results are lifted according to `lift_flat_values` above,
3355+
the optional `post-return` function (specified as a `canonopt` immediate of
3356+
`canon lift`) is called, passing the same core wasm results as parameters so
3357+
that the `post-return` function can free any associated allocations.
33773358
```python
33783359
if not opts.async_:
33793360
flat_results = call_and_trap_on_throw(callee, flat_args)
33803361
assert(types_match_values(flat_ft.results, flat_results))
3362+
trap_if(thread.task is not task)
33813363
result = lift_flat_values(cx, MAX_FLAT_RESULTS, CoreValueIter(flat_results), ft.result_type())
33823364
task.return_(result)
33833365
if opts.post_return is not None:
@@ -3394,11 +3376,11 @@ functions can always be implemented by a plain synchronous function call
33943376
without the need for fibers which would otherwise be necessary if the
33953377
`post-return` function performed a blocking operation.
33963378

3397-
In both of the `async` cases below (with or without `callback`), the
3398-
`task.return` built-in must be called, providing the return value as core wasm
3399-
*parameters* to the `task.return` built-in (rather than as core function
3400-
results as in the synchronous case). If `task.return` is *not* called by the
3401-
time the `Task`'s last `Thread` exits, there is a trap (in `Task.unregister_thread`).
3379+
In both of the `async` cases below (with or without `callback`), the return
3380+
value is provided by calling the `task.return` built-in, passing the return
3381+
value as core wasm *parameters* (rather than as core function results as in
3382+
the synchronous case). If `task.return` never ends up being called, the
3383+
task will never complete for the caller.
34023384

34033385
In the `async` non-`callback` ("stackful async") case, there is a single call
34043386
to the core wasm callee which must return empty core results. Waiting for async
@@ -3426,7 +3408,7 @@ function (specified as a `funcidx` immediate in `canon lift`) until the
34263408
if thread.task.deliver_pending_cancel():
34273409
event = (EventCode.TASK_CANCELLED, 0, 0)
34283410
else:
3429-
assert(inst.exclusive_thread is task.implicit_thread)
3411+
assert(inst.exclusive_thread is thread)
34303412
inst.exclusive_thread = None
34313413
thread.cancellable = True
34323414
match code:
@@ -3444,7 +3426,7 @@ function (specified as a `funcidx` immediate in `canon lift`) until the
34443426
trap()
34453427
assert(inst.exclusive_thread is None)
34463428
thread.cancellable = False
3447-
inst.exclusive_thread = task.implicit_thread
3429+
inst.exclusive_thread = thread
34483430
event_code, p1, p2 = event
34493431
[packed] = call_and_trap_on_throw(opts.callback, [event_code, p1, p2])
34503432
code,si = unpack_callback_result(packed)
@@ -4490,17 +4472,17 @@ class CoreFuncRef:
44904472
callee: Callable[[list[CoreValType]], list[CoreValType]]
44914473

44924474
def canon_thread_new_indirect(ft, ftbl: Table[CoreFuncRef], fi, c):
4493-
task = current_task()
4494-
trap_if(not task.inst.may_leave)
4475+
inst = current_instance()
4476+
trap_if(not inst.may_leave)
44954477
f = ftbl.get(fi)
44964478
assert(ft == CoreFuncType(['i32'], []) or ft == CoreFuncType(['i64'], []))
44974479
trap_if(f.t != ft)
44984480
def thread_func():
44994481
[] = call_and_trap_on_throw(f.callee, [c])
4500-
task.unregister_thread(new_thread)
4501-
new_thread = Thread(task, thread_func)
4482+
inst.threads.remove(new_thread.index)
4483+
new_thread = Thread(current_task(), thread_func)
45024484
assert(new_thread.suspended())
4503-
task.register_thread(new_thread)
4485+
new_thread.index = inst.threads.add(new_thread)
45044486
return [new_thread.index]
45054487
```
45064488
The newly-created thread starts out in a "suspended" state and so, to
@@ -4679,6 +4661,77 @@ def canon_thread_yield_then_promote(i):
46794661
```
46804662

46814663

4664+
###### 🧵 `canon thread.set-task`
4665+
4666+
For a canonical definition:
4667+
```wat
4668+
(canon thread.set-task (core func $thread.set-task))
4669+
```
4670+
validation specifies:
4671+
* `$thread.set-task` is given type `(func (param i32))`
4672+
4673+
Calling `$thread.set-task` invokes the following function which sets the current
4674+
thread's containing task to be the task referenced by the given `i32` index.
4675+
```python
4676+
def canon_thread_set_task(taski):
4677+
thread = current_thread()
4678+
trap_if(not thread.task.inst.may_leave)
4679+
new_task = thread.task.inst.handles.get(taski)
4680+
trap_if(not isinstance(new_task, Task))
4681+
thread.task = new_task
4682+
return []
4683+
```
4684+
At the moment, the only source of task indices is `thread.get-task`.
4685+
4686+
4687+
###### 🧵 `canon thread.get-task`
4688+
4689+
For a canonical definition:
4690+
```wat
4691+
(canon thread.get-task (core func $thread.get-task))
4692+
```
4693+
validation specifies:
4694+
* `$thread.get-task` is given type `(func (result i32))`
4695+
4696+
Calling `$thread.get-task` invokes the following function which adds a new
4697+
handle containing a reference to the current task to the current component
4698+
instance's `handles` table, returning the `i32` index of the new task handle,
4699+
which can be passed as an operand to `thread.set-task`.
4700+
```python
4701+
def canon_thread_get_task():
4702+
task = current_task()
4703+
trap_if(not task.inst.may_leave)
4704+
taski = task.inst.handles.add(task)
4705+
return [taski]
4706+
```
4707+
Note that each call to `thread.get-task` unconditionally allocates a new handle
4708+
and thus must be paired with a call to `task.drop` to avoid leaking handles.
4709+
4710+
4711+
###### 🧵 `canon task.drop`
4712+
4713+
For a canonical definition:
4714+
```wat
4715+
(canon thread.drop (core func $task.drop))
4716+
```
4717+
validation specifies:
4718+
* `$thread.drop` is given type `(func (param i32))`
4719+
4720+
Calling `$task.drop` invokes the following function which drops the task handle
4721+
at the given index.
4722+
```python
4723+
def canon_task_drop(taski):
4724+
inst = current_instance()
4725+
trap_if(not inst.may_leave)
4726+
task = inst.handles.remove(taski)
4727+
trap_if(not isinstance(task, Task))
4728+
return []
4729+
```
4730+
Note that `task.drop` only drops a *reference* to a task; it does not change the
4731+
*state* of the task or destroy the task (which may well have have other handle,
4732+
thread and subtask referents).
4733+
4734+
46824735
### 📝 `canon error-context.new`
46834736

46844737
For a canonical definition:

0 commit comments

Comments
 (0)