Trap on call to {stream,future}.cancel-{read,write} after a prior cancellation blocked

Author: scott-wagner

design/mvp/CanonicalABI.md

@@ -1523,7 +1523,8 @@ class CopyState(Enum):
   IDLE = 1
   SYNC_COPYING = 2
   ASYNC_COPYING = 3
-  DONE = 4
+  CANCELLING_COPY = 4
+  DONE = 5
 
 class CopyEnd(Waitable):
   state: CopyState
@@ -1533,18 +1534,23 @@ class CopyEnd(Waitable):
     self.state = CopyState.IDLE
 
   def copying(self):
-    return self.state == CopyState.SYNC_COPYING or self.state == CopyState.ASYNC_COPYING
+    match self.state:
+      case CopyState.IDLE | CopyState.DONE:
+        return False
+      case CopyState.SYNC_COPYING | CopyState.ASYNC_COPYING | CopyState.CANCELLING_COPY:
+        return True
+    assert(False)
 
   def drop(self):
     trap_if(self.copying())
     Waitable.drop(self)
 ```
 As shown in `drop`, attempting to drop a readable or writable end while a copy
-is in progress traps. This means that client code must take care to wait for
-these operations to finish (potentially cancelling them via
-`stream.cancel-{read,write}`) before dropping. The `SYNC_COPY` vs. `ASYNC_COPY`
-distinction is tracked in the state to determine whether the copy operation can
-be cancelled.
+is in progress or in the process of being cancelled traps. This means that
+client code must take care to wait for these operations to finish (potentially
+cancelling them via `stream.cancel-{read,write}`) before dropping. The
+`SYNC_COPY` vs. `ASYNC_COPY` distinction is tracked in the state to determine
+whether the copy operation can be cancelled.
 
 Given the above, we can define the concrete `{Readable,Writable}StreamEnd`
 classes which are almost entirely symmetric, with the only difference being
@@ -4305,6 +4311,7 @@ def cancel_copy(EndT, event_code, stream_or_future_t, async_, thread, i):
   trap_if(not isinstance(e, EndT))
   trap_if(e.shared.t != stream_or_future_t.t)
   trap_if(e.state != CopyState.ASYNC_COPYING)
+  e.state = CopyState.CANCELLING_COPY
   if not e.has_pending_event():
     e.shared.cancel()
     if not e.has_pending_event():
@@ -4321,7 +4328,8 @@ unconditionally traps if it transitively attempts to make a synchronous call to
 `cancel-read` or `cancel-write` (regardless of whether the cancellation would
 have completed without blocking). There is also a trap if there is not
 currently an async copy in progress (sync copies do not expect or check for
-cancellation and thus cannot be cancelled).
+cancellation and thus cannot be cancelled and repeatedly cancelling the same
+async copy after the first call blocked is not allowed).
 
 The *first* check for `e.has_pending_event()` catches the case where the copy has
 already racily finished, in which case we must *not* call `cancel()`. Calling

design/mvp/canonical-abi/definitions.py

@@ -887,7 +887,8 @@ class CopyState(Enum):
   IDLE = 1
   SYNC_COPYING = 2
   ASYNC_COPYING = 3
-  DONE = 4
+  CANCELLING_COPY = 4
+  DONE = 5
 
 class CopyEnd(Waitable):
   state: CopyState
@@ -897,7 +898,12 @@ def __init__(self):
     self.state = CopyState.IDLE
 
   def copying(self):
-    return self.state == CopyState.SYNC_COPYING or self.state == CopyState.ASYNC_COPYING
+    match self.state:
+      case CopyState.IDLE | CopyState.DONE:
+        return False
+      case CopyState.SYNC_COPYING | CopyState.ASYNC_COPYING | CopyState.CANCELLING_COPY:
+        return True
+    assert(False)
 
   def drop(self):
     trap_if(self.copying())
@@ -2471,6 +2477,7 @@ def cancel_copy(EndT, event_code, stream_or_future_t, async_, thread, i):
   trap_if(not isinstance(e, EndT))
   trap_if(e.shared.t != stream_or_future_t.t)
   trap_if(e.state != CopyState.ASYNC_COPYING)
+  e.state = CopyState.CANCELLING_COPY
   if not e.has_pending_event():
     e.shared.cancel()
     if not e.has_pending_event():

design/mvp/canonical-abi/run_tests.py

@@ -2031,6 +2031,11 @@ def core_func(thread, args):
     host_source.block_cancel()
     [ret] = canon_stream_cancel_read(StreamType(U8Type()), True, thread, rsi)
     assert(ret == definitions.BLOCKED)
+    try:
+      canon_stream_cancel_read(StreamType(U8Type()), True, thread, rsi)
+      assert(False)
+    except Trap:
+      pass
     host_source.write([7,8])
     host_source.unblock_cancel()
     [seti] = canon_waitable_set_new(thread)