Deliver stream/future dropped events even when not reading/writing

Author: lukewagner

design/mvp/CanonicalABI.md

@@ -1265,7 +1265,6 @@ class Waitable:
       wset.elems.append(self)
 
   def drop(self):
-    assert(not self.has_pending_event())
     assert(not self.has_sync_waiter)
     self.join(None)
 ```
@@ -1646,7 +1645,7 @@ buffer of length `1`, the second half of case `3` and cases `4` and `5` do not
 apply to futures. Enumerating the cases in the order that they are handled in
 the code below:
 1. The other end was racily dropped before this end could be notified, in which
-   case the call immediately completes, reporting `DROPPED` and nothing copied.
+   case `End.drop` already left a pending event and so there's nothing to do.
 2. The other end has not currently provided a buffer, in which case this end
    must block until the other end shows up with a buffer.
 3. Both this and the other end have provided buffers that can copy at least 1
@@ -1668,7 +1667,7 @@ the code below:
     assert(self.buffer is None)
     self.state = End.State.COPYING
     if self.other is None:
-      self.notify(End.Result.DROPPED, progress = 0)
+      assert(self.has_pending_event())
     elif self.other.buffer is None:
       self.buffer = buffer
     elif buffer.remain() > 0 and self.other.buffer.remain() > 0:
@@ -1721,14 +1720,15 @@ cancellation. In the future, guest components may be given the same capability.
 
 The `End.drop` method is called by `{stream,future}.drop-{readable,writable}` to
 update the `other` end's state and possibly set a pending notification for the
-other end, if doing so wouldn't clobber an already-pending notification.
+other end, if the other end isn't already `DONE` and doing so wouldn't clobber
+an already-pending notification.
 ```python
   def drop(self):
     assert(not self.copying_or_cancelling())
     if self.other is not None:
       assert(self is self.other.other)
       self.other.other = None
-      if self.other.copying_or_cancelling() and not self.other.has_pending_event():
+      if self.other.state != End.State.DONE and not self.other.has_pending_event():
         self.other.notify(End.Result.DROPPED)
       self.other = None
     Waitable.drop(self)

design/mvp/canonical-abi/definitions.py

@@ -742,7 +742,6 @@ def join(self, wset):
       wset.elems.append(self)
 
   def drop(self):
-    assert(not self.has_pending_event())
     assert(not self.has_sync_waiter)
     self.join(None)
 
@@ -945,7 +944,7 @@ def copy(self, buffer: Buffer, is_read: bool):
     assert(self.buffer is None)
     self.state = End.State.COPYING
     if self.other is None:
-      self.notify(End.Result.DROPPED, progress = 0)
+      assert(self.has_pending_event())
     elif self.other.buffer is None:
       self.buffer = buffer
     elif buffer.remain() > 0 and self.other.buffer.remain() > 0:
@@ -980,7 +979,7 @@ def drop(self):
     if self.other is not None:
       assert(self is self.other.other)
       self.other.other = None
-      if self.other.copying_or_cancelling() and not self.other.has_pending_event():
+      if self.other.state != End.State.DONE and not self.other.has_pending_event():
         self.other.notify(End.Result.DROPPED)
       self.other = None
     Waitable.drop(self)

design/mvp/canonical-abi/run_tests.py

@@ -1673,6 +1673,14 @@ def core_func(args):
   lift_and_run(opts, inst, ft, core_func, on_start, on_resolve)
   assert(host_writer.readable_end is return_value)
 
+  # A stream whose writable end has already been dropped can still be lowered
+  # into and lifted out of the component; whoever finally reads it sees DROPPED.
+  host_writer = HostWriter(U8Type(), [])
+  lift_and_run(opts, inst, ft, core_func, on_start, on_resolve)
+  assert(host_writer.readable_end is return_value)
+  host_reader = HostReader(return_value)
+  assert(host_reader.dropped and host_reader.take() == [])
+
 
 def test_receive_own_stream():
   store = Store()
@@ -2254,6 +2262,66 @@ def core_func(args):
   lift_and_run(opts, inst, caller_ft, core_func, lambda:[], lambda _:())
 
 
+def test_stream_drop_both_ends_while_idle():
+  store = Store()
+  inst = ComponentInstance(store)
+  mem = bytearray(24)
+  opts = mk_opts(memory=MemInst(mem, 'i32'), async_ = True)
+  stream_t = StreamType(U8Type())
+  future_t = FutureType(U8Type())
+
+  def core_func(args):
+    assert(len(args) == 0)
+    [] = canon_task_return([], opts, [])
+    [packed] = canon_stream_new(stream_t)
+    rsi,wsi = unpack_new_ends(packed)
+    [] = canon_stream_drop_writable(stream_t, wsi)
+    [] = canon_stream_drop_readable(stream_t, rsi)
+
+    # Dropping one end notifies the other end even when it has no read or write
+    # in flight: the DROPPED event is delivered (once) through the waitable set,
+    # after which the end is done and further reads trap.
+    retp = 8
+    [seti] = canon_waitable_set_new()
+    [packed] = canon_stream_new(stream_t)
+    rsi,wsi = unpack_new_ends(packed)
+    [] = canon_waitable_join(rsi, seti)
+    [] = canon_stream_drop_writable(stream_t, wsi)
+    [event] = canon_waitable_set_poll(MemInst(mem, 'i32'), seti, retp)
+    assert(event == EventCode.STREAM_READ)
+    assert(mem[retp+0] == rsi)
+    result,n = unpack_result(mem[retp+4])
+    assert(n == 0 and result == End.Result.DROPPED)
+    [event] = canon_waitable_set_poll(MemInst(mem, 'i32'), seti, retp)
+    assert(event == EventCode.NONE)
+    trapped = False
+    try:
+      canon_stream_read(stream_t, opts, rsi, 0, 4)
+    except Trap:
+      trapped = True
+    assert(trapped)
+    [] = canon_waitable_join(rsi, 0)
+    [] = canon_stream_drop_readable(stream_t, rsi)
+
+    # Same for the writable end of a future (using wait instead of poll); once
+    # notified, the writable end may be dropped without having written.
+    [packed] = canon_future_new(future_t)
+    rfi,wfi = unpack_new_ends(packed)
+    [] = canon_waitable_join(wfi, seti)
+    [] = canon_future_drop_readable(future_t, rfi)
+    [event] = canon_waitable_set_wait(MemInst(mem, 'i32'), seti, retp)
+    assert(event == EventCode.FUTURE_WRITE)
+    assert(mem[retp+0] == wfi)
+    assert(mem[retp+4] == End.Result.DROPPED)
+    [] = canon_waitable_join(wfi, 0)
+    [] = canon_future_drop_writable(future_t, wfi)
+    [] = canon_waitable_set_drop(seti)
+    return []
+
+  caller_ft = FuncType([], [], async_ = True)
+  lift_and_run(opts, inst, caller_ft, core_func, lambda:[], lambda _:())
+
+
 def test_cancel_subtask():
   store = Store()
   ft = FuncType([U8Type()], [U8Type()], async_ = True)
@@ -2982,6 +3050,7 @@ def on_resolve(v):
 test_cancel_copy()
 test_futures()
 test_future_drop_readable_with_pending_write()
+test_stream_drop_both_ends_while_idle()
 test_cancel_subtask()
 test_self_copy(None)
 test_self_copy(U8Type())