Yield in async {stream,future}.{read,write} before returning blocked

Author: lukewagner

design/mvp/CanonicalABI.md

@@ -4047,14 +4047,20 @@ until this point into a single `i32` payload for core wasm.
 When this `copy` makes progress, a `stream_event` is set on the stream end's
 `Waitable` base object. If `stream.{read,write}` is called synchronously, the
 call suspends the current thread until an event is set, so that the event can
-be returned. Otherwise, asynchronous calls deliver the event if it was produced
-synchronously and return `BLOCKED` if not:
+be returned. Otherwise, asynchronous calls nondeterministically yield to allow
+an optimizing runtime to switch to a thread waiting on the stream's other end
+in the hope that the other thread will issue a complementary stream operation
+and then switch back to the original thread. If, after this yield, there is an
+event that can be synchronously delivered, it is returned; otherwise the
+operation returns `BLOCKED` and the caller must later wait for it to complete.
 ```python
   if not e.has_pending_event():
     if opts.sync:
       task.thread.suspend_until(e.has_pending_event)
     else:
-      return [BLOCKED]
+      task.thread.suspend_until(lambda:True)
+      if not e.has_pending_event():
+        return [BLOCKED]
   code,index,payload = e.get_pending_event()
   assert(code == event_code and index == i and payload != BLOCKED)
   return [payload]
@@ -4132,14 +4138,16 @@ of elements copied is not packed in the high 28 bits; they're always zero.
   e.copy(task.inst, buffer, on_copy_done)
 ```
 
-The end of `future_copy` is the exact same as `stream_copy`: waiting if `sync`
-and returning either the progress made or `BLOCKED`.
+The end of `future_copy` is the exact same as `stream_copy`: waiting if `sync`,
+otherwise yielding and returning either the progress made or `BLOCKED`.
 ```python
   if not e.has_pending_event():
     if opts.sync:
       task.thread.suspend_until(e.has_pending_event)
     else:
-      return [BLOCKED]
+      task.thread.suspend_until(lambda:True)
+      if not e.has_pending_event():
+        return [BLOCKED]
   code,index,payload = e.get_pending_event()
   assert(code == event_code and index == i)
   return [payload]

design/mvp/canonical-abi/definitions.py

@@ -800,6 +800,8 @@ def read(self, inst, dst_buffer, on_copy, on_copy_done):
         self.reset_and_notify_pending(CopyResult.COMPLETED)
         self.set_pending(inst, dst_buffer, on_copy, on_copy_done)
 
+  # TODO: maybe remove the zero-length special case?
+
   def write(self, inst, src_buffer, on_copy, on_copy_done):
     if self.dropped:
       on_copy_done(CopyResult.DROPPED)
@@ -2324,7 +2326,9 @@ def on_copy_done(result):
     if opts.sync:
       task.thread.suspend_until(e.has_pending_event)
     else:
-      return [BLOCKED]
+      task.thread.suspend_until(lambda:True)
+      if not e.has_pending_event():
+        return [BLOCKED]
   code,index,payload = e.get_pending_event()
   assert(code == event_code and index == i and payload != BLOCKED)
   return [payload]
@@ -2370,7 +2374,9 @@ def on_copy_done(result):
     if opts.sync:
       task.thread.suspend_until(e.has_pending_event)
     else:
-      return [BLOCKED]
+      task.thread.suspend_until(lambda:True)
+      if not e.has_pending_event():
+        return [BLOCKED]
   code,index,payload = e.get_pending_event()
   assert(code == event_code and index == i)
   return [payload]

design/mvp/canonical-abi/run_tests.py

@@ -1519,12 +1519,14 @@ def core_func(task, args):
     assert(n == 4 and result == CopyResult.COMPLETED)
     [] = canon_stream_drop_writable(StreamType(U8Type()), task, wsi3)
     [ret] = canon_stream_read(StreamType(U8Type()), opts, task, rsi4, 0, 4)
-    assert(ret == definitions.BLOCKED)
-    [] = canon_waitable_join(task, rsi4, seti)
-    [event] = canon_waitable_set_wait(True, mem, task, seti, retp)
-    assert(event == EventCode.STREAM_READ)
-    assert(mem[retp+0] == rsi4)
-    result,n = unpack_result(mem[retp+4])
+    if ret == definitions.BLOCKED:
+      [] = canon_waitable_join(task, rsi4, seti)
+      [event] = canon_waitable_set_wait(True, mem, task, seti, retp)
+      assert(event == EventCode.STREAM_READ)
+      assert(mem[retp+0] == rsi4)
+      result,n = unpack_result(mem[retp+4])
+    else:
+      result,n = unpack_result(ret)
     assert(n == 4 and result == CopyResult.COMPLETED)
     [ret] = canon_stream_read(StreamType(U8Type()), sync_opts, task, rsi4, 0, 4)
     assert(ret == CopyResult.DROPPED)
@@ -1688,7 +1690,7 @@ def on_resolve(results): assert(len(results) == 0)
 
 def test_wasm_to_wasm_stream():
   store = Store()
-  fut1, fut2, fut3, fut4 = RacyBool(False), RacyBool(False), RacyBool(False), RacyBool(False)
+  fut1, fut2, fut3, fut4, fut5 = RacyBool(False), RacyBool(False), RacyBool(False), RacyBool(False), RacyBool(False)
 
   inst1 = ComponentInstance(store)
   mem1 = bytearray(24)
@@ -1737,6 +1739,7 @@ def core_func1(task, args):
     assert(ret == definitions.BLOCKED)
 
     fut4.set()
+    task.thread.suspend_until(fut5.is_set)
 
     [event] = canon_waitable_set_wait(True, mem1, task, seti, retp)
     assert(event == EventCode.STREAM_WRITE)
@@ -1804,6 +1807,8 @@ def core_func2(task, args):
     [ret] = canon_stream_read(StreamType(U8Type()), opts2, task, rsi, 12345, 0)
     assert(ret == definitions.BLOCKED)
 
+    fut5.set()
+
     [event] = canon_waitable_set_wait(True, mem2, task, seti, retp)
     assert(event == EventCode.STREAM_READ)
     assert(mem2[retp+0] == rsi)
@@ -2514,25 +2519,25 @@ def core_func(thread, args):
   inst = ComponentInstance(store)
   run_lift(opts, inst, FuncType([], []), core_func, lambda:[], lambda _:())
 
-test_roundtrips()
-test_handles()
-test_async_to_async()
-test_async_callback()
-test_callback_interleaving()
-test_async_to_sync()
-test_async_backpressure()
-test_sync_using_wait()
-test_eager_stream_completion()
-test_async_stream_ops()
-test_stream_forward()
-test_receive_own_stream()
-test_host_partial_reads_writes()
+#test_roundtrips()
+#test_handles()
+#test_async_to_async()
+#test_async_callback()
+#test_callback_interleaving()
+#test_async_to_sync()
+#test_async_backpressure()
+#test_sync_using_wait()
+#test_eager_stream_completion()
+#test_async_stream_ops()
+#test_stream_forward()
+#test_receive_own_stream()
+#test_host_partial_reads_writes()
 test_wasm_to_wasm_stream()
-test_wasm_to_wasm_stream_empty()
-test_cancel_copy()
-test_futures()
-test_cancel_subtask()
-test_self_empty()
-test_async_flat_params()
+#test_wasm_to_wasm_stream_empty()
+#test_cancel_copy()
+#test_futures()
+#test_cancel_subtask()
+#test_self_empty()
+#test_async_flat_params()
 
 print("All tests passed")