Add backpressure.{inc,dec}, deprecate backpressure.set (#560)

Author: lukewagner

design/mvp/Async.md

@@ -466,9 +466,13 @@ export calls can start piling up, each consuming some of the component's finite
 private resources (like linear memory), requiring the component to be able to
 exert *backpressure* to allow some tasks to finish (and release private
 resources) before admitting new async export calls. To do this, a component may
-call the [`backpressure.set`] built-in to set a component-instance-wide
-"backpressure" flag that causes subsequent export calls to immediately return
-in the "starting" state without calling the component's Core WebAssembly code.
+call the [`backpressure.inc`] built-in to increment a component-instance-wide
+"backpressure" counter until resources are freed and then call
+[`backpressure.dec`] to decrement the counter. When the backpressure counter is
+greater than zero, new export calls immediately return in the "starting" state
+without calling the component's Core WebAssembly code. By using a counter
+instead of a boolean flag, unrelated pieces of code can report backpressure for
+distinct limited resources without prior coordination.
 
 In addition to *explicit* backpressure set by wasm code, there is also an
 *implicit* source of backpressure used to protect non-reentrant core wasm code.
@@ -1137,7 +1141,8 @@ comes after:
 [Canonical Built-in]: Explainer.md#canonical-built-ins
 [`context.get`]: Explainer.md#-contextget
 [`context.set`]: Explainer.md#-contextset
-[`backpressure.set`]: Explainer.md#-backpressureset
+[`backpressure.inc`]: Explainer.md#-backpressureinc-and-backpressuredec
+[`backpressure.dec`]: Explainer.md#-backpressureinc-and-backpressuredec
 [`task.return`]: Explainer.md#-taskreturn
 [`task.cancel`]: Explainer.md#-taskcancel
 [`subtask.cancel`]: Explainer.md#-subtaskcancel
@@ -1155,7 +1160,6 @@ comes after:
 [`unpack_callback_result`]: CanonicalABI.md#canon-lift
 [`canon_lower`]: CanonicalABI.md#canon-lower
 [`canon_context_get`]: CanonicalABI.md#-canon-contextget
-[`canon_backpressure_set`]: CanonicalABI.md#-canon-backpressureset
 [`canon_waitable_set_wait`]: CanonicalABI.md#-canon-waitable-setwait
 [`canon_task_return`]: CanonicalABI.md#-canon-taskreturn
 [`canon_task_cancel`]: CanonicalABI.md#-canon-taskcancel

design/mvp/Binary.md

@@ -290,7 +290,9 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 0x03 rt:<typeidx>                                   => (canon resource.drop rt (core func))
            | 0x07 rt:<typeidx>                                   => (canon resource.drop rt async (core func)) 🚝
            | 0x04 rt:<typeidx>                                   => (canon resource.rep rt (core func))
-           | 0x08                                                => (canon backpressure.set (core func)) 🔀
+           | 0x08                                                => (canon backpressure.set (core func)) 🔀✕
+           | 0x24                                                => (canon backpressure.inc (core func)) 🔀
+           | 0x25                                                => (canon backpressure.dec (core func)) 🔀
            | 0x09 rs:<resultlist> opts:<opts>                    => (canon task.return rs opts (core func)) 🔀
            | 0x05                                                => (canon task.cancel (core func)) 🔀
            | 0x0a 0x7f i:<u32>                                   => (canon context.get i32 i (core func)) 🔀

design/mvp/CanonicalABI.md

@@ -43,7 +43,8 @@ being specified here.
   * [`canon resource.rep`](#canon-resourcerep)
   * [`canon context.get`](#-canon-contextget) 🔀
   * [`canon context.set`](#-canon-contextset) 🔀
-  * [`canon backpressure.set`](#-canon-backpressureset) 🔀
+  * [`canon backpressure.set`](#-canon-backpressureset) 🔀✕
+  * [`canon backpressure.{inc,dec}`](#-canon-backpressureincdec) 🔀
   * [`canon task.return`](#-canon-taskreturn) 🔀
   * [`canon task.cancel`](#-canon-taskcancel) 🔀
   * [`canon yield`](#-canon-yield) 🔀
@@ -279,15 +280,15 @@ class ComponentInstance:
   store: Store
   table: Table
   may_leave: bool
-  backpressure: bool
+  backpressure: int
   exclusive: bool
   num_waiting_to_enter: int
 
   def __init__(self, store):
     self.store = store
     self.table = Table()
     self.may_leave = True
-    self.backpressure = False
+    self.backpressure = 0
     self.exclusive = False
     self.num_waiting_to_enter = 0
 ```
@@ -829,8 +830,8 @@ to avoid the need to otherwise-endlessly allocate guest memory for blocked
 async calls until OOM. When backpressure is enabled, `enter` will block until
 backpressure is disabled. There are three sources of backpressure:
  1. *Explicit backpressure* is triggered by core wasm calling
-    `backpressure.set` which, in `canon_backpressure_set` (defined below),
-    sets the `ComponentInstance.backpressure` flag.
+    `backpressure.{inc,dec}` which modify the `ComponentInstance.backpressure`
+    counter.
  2. *Implicit backpressure* triggered when `Task.needs_exclusive()` is true and
     the `exclusive` lock is already held.
  3. *Residual backpressure* triggered by explicit or implicit backpressure
@@ -842,7 +843,7 @@ backpressure is disabled. There are three sources of backpressure:
   def enter(self):
     assert(self.thread is not None)
     def has_backpressure():
-      return self.inst.backpressure or (self.needs_exclusive() and self.inst.exclusive)
+      return self.inst.backpressure > 0 or (self.needs_exclusive() and self.inst.exclusive)
     if has_backpressure() or self.inst.num_waiting_to_enter > 0:
       self.inst.num_waiting_to_enter += 1
       completed = self.thread.suspend_until(lambda: not has_backpressure(), cancellable = True)
@@ -3533,7 +3534,12 @@ def canon_context_set(t, i, task, v):
 ```
 
 
-### 🔀 `canon backpressure.set`
+### 🔀✕ `canon backpressure.set`
+
+> This built-in is deprecated in favor of `backpressure.{inc,dec}` and will be
+> removed once producer tools have transitioned. Producer tools should avoid
+> emitting calls to both `set` and `inc`/`dec` since `set` will clobber the
+> counter.
 
 For a canonical definition:
 ```wat
@@ -3542,16 +3548,42 @@ For a canonical definition:
 validation specifies:
 * `$f` is given type `(func (param $enabled i32))`
 
-Calling `$f` invokes the following function, which sets or clears the
-`ComponentInstance.backpressure` flag. `Task.enter` waits for this flag to be
-clear before allowing new tasks to start.
+Calling `$f` invokes the following function, which sets the `backpressure`
+counter to `1` or `0`. `Task.enter` waits for `backpressure` to be `0` before
+allowing new tasks to start.
 ```python
 def canon_backpressure_set(task, flat_args):
-  trap_if(task.opts.sync)
   assert(len(flat_args) == 1)
-  task.inst.backpressure = bool(flat_args[0])
+  task.inst.backpressure = int(bool(flat_args[0]))
+  return []
+```
+
+### 🔀 `canon backpressure.{inc,dec}`
+
+For a canonical definition:
+```wat
+(canon backpressure.inc (core func $inc))
+(canon backpressure.dec (core func $dec))
+```
+validation specifies:
+* `$inc`/`$dec` are given type `(func)`
+
+Calling `$inc` or `$dec` invokes one of the following functions:
+```python
+def canon_backpressure_inc(task):
+  assert(0 <= task.inst.backpressure < 2**16)
+  task.inst.backpressure += 1
+  trap_if(task.inst.backpressure == 2**16)
+  return []
+
+def canon_backpressure_dec(task):
+  assert(0 <= task.inst.backpressure < 2**16)
+  task.inst.backpressure -= 1
+  trap_if(task.inst.backpressure < 0)
   return []
 ```
+`Task.enter` waits for `backpressure` to return to `0` before allowing new
+tasks to start, implementing [backpressure].
 
 
 ### 🔀 `canon task.return`

design/mvp/Explainer.md

@@ -1417,7 +1417,9 @@ canon ::= ...
         | (canon resource.rep <typeidx> (core func <id>?))
         | (canon context.get <valtype> <u32> (core func <id>?)) 🔀
         | (canon context.set <valtype> <u32> (core func <id>?)) 🔀
-        | (canon backpressure.set (core func <id>?)) 🔀
+        | (canon backpressure.set (core func <id>?)) 🔀✕
+        | (canon backpressure.inc (core func <id>?)) 🔀
+        | (canon backpressure.dec (core func <id>?)) 🔀
         | (canon task.return (result <valtype>)? <canonopt>* (core func <id>?)) 🔀
         | (canon task.cancel (core func <id>?)) 🔀
         | (canon yield cancellable? (core func <id>?)) 🔀
@@ -1580,7 +1582,10 @@ future (as described [here][context-local storage]).
 
 For details, see [`canon_context_set`] in the Canonical ABI explainer.
 
-###### 🔀 `backpressure.set`
+###### 🔀✕ `backpressure.set`
+
+> This built-in is deprecated in favor of `backpressure.{inc,dec}` and will be
+> removed once producer tools have transitioned.
 
 | Synopsis                   |                       |
 | -------------------------- | --------------------- |
@@ -1594,6 +1599,26 @@ to the component (until the flag is unset). This allows the component to exert
 
 For details, see [`canon_backpressure_set`] in the Canonical ABI explainer.
 
+###### 🔀 `backpressure.inc` and `backpressure.dec`
+
+| Synopsis                   |            |
+| -------------------------- | ---------- |
+| Approximate WIT signature  | `func()`   |
+| Canonical ABI signature    | `[] -> []` |
+
+The `backpressure.{inc,dec}` built-ins allow code running in a component to
+prevent new incoming export calls to the component by enabling [backpressure].
+These built-ins increment and decrement a per-component-instance counter that,
+when greater than zero, enables backpressure.
+
+If these built-ins would overflow or underflow a 16-bit unsigned integer, they
+trap instead. As a composable convention, each piece of code that calls
+`backpressure.inc` must take responsibility for calling `backpressure.dec`
+exactly once when the source of backpressure subsides.
+
+For details, see [`canon_backpressure_{inc,dec}`] in the Canonical ABI
+explainer.
+
 ###### 🔀 `task.return`
 
 The `task.return` built-in takes as parameters the result values of the
@@ -3047,6 +3072,7 @@ For some use-case-focused, worked examples, see:
 [`canon_context_get`]: CanonicalABI.md#-canon-contextget
 [`canon_context_set`]: CanonicalABI.md#-canon-contextset
 [`canon_backpressure_set`]: CanonicalABI.md#-canon-backpressureset
+[`canon_backpressure_{inc,dec}`]: CanonicalABI.md#-canon-backpressureincdec
 [`canon_task_return`]: CanonicalABI.md#-canon-taskreturn
 [`canon_task_cancel`]: CanonicalABI.md#-canon-taskcancel
 [`canon_yield`]: CanonicalABI.md#-canon-yield

design/mvp/canonical-abi/definitions.py

@@ -251,15 +251,15 @@ class ComponentInstance:
   store: Store
   table: Table
   may_leave: bool
-  backpressure: bool
+  backpressure: int
   exclusive: bool
   num_waiting_to_enter: int
 
   def __init__(self, store):
     self.store = store
     self.table = Table()
     self.may_leave = True
-    self.backpressure = False
+    self.backpressure = 0
     self.exclusive = False
     self.num_waiting_to_enter = 0
 
@@ -540,7 +540,7 @@ def needs_exclusive(self):
   def enter(self):
     assert(self.thread is not None)
     def has_backpressure():
-      return self.inst.backpressure or (self.needs_exclusive() and self.inst.exclusive)
+      return self.inst.backpressure > 0 or (self.needs_exclusive() and self.inst.exclusive)
     if has_backpressure() or self.inst.num_waiting_to_enter > 0:
       self.inst.num_waiting_to_enter += 1
       completed = self.thread.suspend_until(lambda: not has_backpressure(), cancellable = True)
@@ -2125,9 +2125,22 @@ def canon_context_set(t, i, task, v):
 ### 🔀 `canon backpressure.set`
 
 def canon_backpressure_set(task, flat_args):
-  trap_if(task.opts.sync)
   assert(len(flat_args) == 1)
-  task.inst.backpressure = bool(flat_args[0])
+  task.inst.backpressure = int(bool(flat_args[0]))
+  return []
+
+### 🔀 `canon backpressure.{inc,dec}`
+
+def canon_backpressure_inc(task):
+  assert(0 <= task.inst.backpressure < 2**16)
+  task.inst.backpressure += 1
+  trap_if(task.inst.backpressure == 2**16)
+  return []
+
+def canon_backpressure_dec(task):
+  assert(0 <= task.inst.backpressure < 2**16)
+  task.inst.backpressure -= 1
+  trap_if(task.inst.backpressure < 0)
   return []
 
 ### 🔀 `canon task.return`

design/mvp/canonical-abi/run_tests.py

@@ -565,11 +565,11 @@ def core_eager_producer(task, args):
   fut1_2 = RacyBool(False)
   def core_toggle(task, args):
     assert(len(args) == 0)
-    [] = canon_backpressure_set(task, [1])
+    [] = canon_backpressure_inc(task)
     task.thread.suspend_until(fut1_1.is_set)
     [] = canon_task_return(task, [], producer_opts, [])
     task.thread.suspend_until(fut1_2.is_set)
-    [] = canon_backpressure_set(task, [0])
+    [] = canon_backpressure_dec(task)
     return []
   toggle_callee = partial(canon_lift, producer_opts, producer_inst, toggle_ft, core_toggle)
 
@@ -1038,9 +1038,9 @@ def test_async_backpressure():
   producer1_done = False
   def producer1_core(task, args):
     nonlocal producer1_done
-    canon_backpressure_set(task, [1])
+    canon_backpressure_inc(task)
     task.thread.suspend_until(fut.is_set)
-    canon_backpressure_set(task, [0])
+    canon_backpressure_dec(task)
     canon_task_return(task, [], producer_opts, [])
     producer1_done = True
     return []