CABI: make futures make sense independently of streams

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -837,11 +837,13 @@ def copy(self, inst, buffer, on_copy, on_result, src, dst):
 class StreamEnd(Waitable):
   shared: ReadableStream
   copying: bool
+  done: bool
 
   def __init__(self, shared):
     Waitable.__init__(self)
     self.shared = shared
     self.copying = False
+    self.done = False
 
   def drop(self):
     trap_if(self.copying)
@@ -858,28 +860,110 @@ def copy(self, inst, src, on_copy, on_result):
 
 #### Future State
 
-class FutureEnd(StreamEnd):
-  def close_after_copy(self, copy_op, inst, buffer, on_result):
-    assert(buffer.remain() == 1)
-    def on_result_wrapper(result):
-      if buffer.remain() == 0:
-        self.shared.close()
-      on_result(result)
-    ret = copy_op(inst, buffer, on_copy = lambda _:(), on_result = on_result_wrapper)
-    if ret is not None and buffer.remain() == 0:
-      self.shared.close()
-      return StreamResult.CLOSED
-    return ret
+class FutureResult(IntEnum):
+  RESOLVED = 0
+  DROPPED = 1 # only returned for future.write
+  CANCELLED = 2
+
+Lift = Callable[[], any]
+Lower = Callable[[any], None]
+OnFutureResult = Callable[[FutureResult], None]
+
+class ReadableFuture:
+  t: ValType
+  read: Callable[[ComponentInstance, Lower, OnFutureResult], Optional[FutureResult]]
+  cancel: Callable[[], None]
+  drop: Callable[[]]
+
+class SharedFutureImpl(ReadableFuture):
+  dropped: bool
+  pending_inst: Optional[ComponentInstance]
+  pending_lift: Optional[Lift]
+  pending_lower: Optional[Lower]
+  pending_on_result: Optional[OnFutureResult]
+
+  def __init__(self, t):
+    self.t = t
+    self.dropped = False
+    self.reset_pending()
+
+  def reset_pending(self):
+    self.set_pending(None, None, None, None)
+
+  def set_pending(self, inst, lift, lower, on_result):
+    assert(not (lift and lower))
+    self.pending_inst = inst
+    self.pending_lift = lift
+    self.pending_lower = lower
+    self.pending_on_result = on_result
+
+  def reset_and_notify_pending(self, result):
+    pending_on_result = self.pending_on_result
+    self.reset_pending()
+    pending_on_result(result)
+
+  def cancel(self):
+    self.reset_pending_and_notify_pending(FutureResult.CANCELLED)
+
+  def drop(self):
+    assert(not self.dropped and not self.pending_lower)
+    self.dropped = True
+    if self.pending_lift:
+      self.reset_and_notify_pending(FutureResult.DROPPED)
+
+  def read(self, inst, lower, on_result):
+    assert(not self.dropped and not self.pending_lower)
+    if not self.pending_lift:
+      self.set_pending(inst, None, lower, on_result)
+      return None
+    else:
+      trap_if(inst is self.pending_inst and self.t is not None) # temporary
+      lower(self.pending_lift())
+      self.reset_and_notify_pending(FutureResult.RESOLVED)
+      return FutureResult.RESOLVED
+
+  def write(self, inst, lift, on_result):
+    assert(not self.pending_lift)
+    if self.dropped:
+      return FutureResult.DROPPED
+    elif not self.pending_lower:
+      self.set_pending(inst, lift, None, on_result)
+      return None
+    else:
+      trap_if(inst is self.pending_inst and self.t is not None) # temporary
+      self.pending_lower(lift())
+      self.reset_and_notify_pending(FutureResult.RESOLVED)
+      return FutureResult.RESOLVED
+
+class FutureEnd(Waitable):
+  shared: ReadableFuture
+  copying: bool
+  done: bool
+
+  def __init__(self, shared):
+    Waitable.__init__(self)
+    self.shared = shared
+    self.copying = False
+    self.done = False
+
+  def drop(self):
+    trap_if(self.copying)
+    Waitable.drop(self)
 
 class ReadableFutureEnd(FutureEnd):
-  def copy(self, inst, dst, on_copy, on_result):
-    return self.close_after_copy(self.shared.read, inst, dst, on_result)
+  def copy(self, inst, lower, on_result):
+    return self.shared.read(inst, lower, on_result)
+
+  def drop(self):
+    self.shared.drop()
+    FutureEnd.drop(self)
 
 class WritableFutureEnd(FutureEnd):
-  def copy(self, inst, src, on_copy, on_result):
-    return self.close_after_copy(self.shared.write, inst, src, on_result)
+  def copy(self, inst, lift, on_result):
+    return self.shared.write(inst, lift, on_result)
+
   def drop(self):
-    trap_if(not self.shared.closed())
+    trap_if(not self.done)
     FutureEnd.drop(self)
 
 ### Despecialization
@@ -1211,16 +1295,15 @@ def lift_stream(cx, i, t):
   return lift_async_value(ReadableStreamEnd, cx, i, t)
 
 def lift_future(cx, i, t):
-  v = lift_async_value(ReadableFutureEnd, cx, i, t)
-  trap_if(v.closed())
-  return v
+  return lift_async_value(ReadableFutureEnd, cx, i, t)
 
 def lift_async_value(ReadableEndT, cx, i, t):
   assert(not contains_borrow(t))
   e = cx.inst.table.remove(i)
   trap_if(not isinstance(e, ReadableEndT))
   trap_if(e.shared.t != t)
   trap_if(e.copying)
+  trap_if(e.done)
   return e.shared
 
 ### Storing
@@ -1514,14 +1597,14 @@ def lower_borrow(cx, rep, t):
   return cx.inst.table.add(h)
 
 def lower_stream(cx, v, t):
+  assert(isinstance(v, ReadableStream))
   return lower_async_value(ReadableStreamEnd, cx, v, t)
 
 def lower_future(cx, v, t):
-  assert(not v.closed())
+  assert(isinstance(v, ReadableFuture))
   return lower_async_value(ReadableFutureEnd, cx, v, t)
 
 def lower_async_value(ReadableEndT, cx, v, t):
-  assert(isinstance(v, ReadableStream))
   assert(not contains_borrow(t))
   return cx.inst.table.add(ReadableEndT(v))
 
@@ -2161,47 +2244,41 @@ async def canon_stream_new(stream_t, task):
 
 async def canon_future_new(future_t, task):
   trap_if(not task.inst.may_leave)
-  shared = SharedStreamImpl(future_t.t)
+  shared = SharedFutureImpl(future_t.t)
   ri = task.inst.table.add(ReadableFutureEnd(shared))
   wi = task.inst.table.add(WritableFutureEnd(shared))
   return [ ri | (wi << 32) ]
 
-### 🔀 `canon {stream,future}.{read,write}`
+### 🔀 `canon stream.{read,write}`
 
 BLOCKED = 0xffff_ffff
 
 async def canon_stream_read(stream_t, opts, task, i, ptr, n):
-  return await copy(ReadableStreamEnd, WritableBufferGuestImpl, EventCode.STREAM_READ,
-                    stream_t, opts, task, i, ptr, n)
+  return await stream_copy(ReadableStreamEnd, WritableBufferGuestImpl, EventCode.STREAM_READ,
+                           stream_t, opts, task, i, ptr, n)
 
 async def canon_stream_write(stream_t, opts, task, i, ptr, n):
-  return await copy(WritableStreamEnd, ReadableBufferGuestImpl, EventCode.STREAM_WRITE,
-                    stream_t, opts, task, i, ptr, n)
-
-async def canon_future_read(future_t, opts, task, i, ptr):
-  return await copy(ReadableFutureEnd, WritableBufferGuestImpl, EventCode.FUTURE_READ,
-                    future_t, opts, task, i, ptr, 1)
+  return await stream_copy(WritableStreamEnd, ReadableBufferGuestImpl, EventCode.STREAM_WRITE,
+                           stream_t, opts, task, i, ptr, n)
 
-async def canon_future_write(future_t, opts, task, i, ptr):
-  return await copy(WritableFutureEnd, ReadableBufferGuestImpl, EventCode.FUTURE_WRITE,
-                    future_t, opts, task, i, ptr, 1)
-
-async def copy(EndT, BufferT, event_code, stream_or_future_t, opts, task, i, ptr, n):
+async def stream_copy(EndT, BufferT, event_code, stream_t, opts, task, i, ptr, n):
   trap_if(not task.inst.may_leave)
   e = task.inst.table.get(i)
   trap_if(not isinstance(e, EndT))
-  trap_if(e.shared.t != stream_or_future_t.t)
+  trap_if(e.shared.t != stream_t.t)
   trap_if(e.copying)
 
-  assert(not contains_borrow(stream_or_future_t))
+  assert(not contains_borrow(stream_t))
   cx = LiftLowerContext(opts, task.inst, borrow_scope = None)
-  buffer = BufferT(stream_or_future_t.t, cx, ptr, n)
+  buffer = BufferT(stream_t.t, cx, ptr, n)
 
   def copy_event(result, revoke_buffer):
     revoke_buffer()
     assert(e.copying)
     e.copying = False
-    return (event_code, i, pack_copy_result(result, buffer))
+    if result != StreamResult.CANCELLED:
+      e.done = True
+    return (event_code, i, pack_stream_result(result, buffer))
 
   def on_copy(revoke_buffer):
     e.set_event(partial(copy_event, StreamResult.COMPLETED, revoke_buffer))
@@ -2211,7 +2288,9 @@ def on_result(result):
 
   result = e.copy(task.inst, buffer, on_copy, on_result)
   if result is not None:
-    return [pack_copy_result(result, buffer)]
+    if result != StreamResult.CANCELLED:
+      e.done = True
+    return [pack_stream_result(result, buffer)]
   else:
     e.copying = True
     if opts.sync:
@@ -2222,13 +2301,68 @@ def on_result(result):
     else:
       return [BLOCKED]
 
-def pack_copy_result(result, buffer):
+def pack_stream_result(result, buffer):
   assert(0 <= result < 2**4)
   assert(buffer.progress <= Buffer.MAX_LENGTH < 2**28)
   packed = result | (buffer.progress << 4)
   assert(packed != BLOCKED)
   return packed
 
+### 🔀 `canon future.{read,write}`
+
+async def canon_future_read(future_t, opts, task, i, flat_args):
+  assert(len(flat_args) == int(bool(future_t.t)))
+  def lower(v):
+    if future_t.t:
+      assert(not contains_borrow(future_t))
+      cx = LiftLowerContext(opts, task.inst, borrow_scope = None)
+      store(cx, v, future_t.t, flat_args[0])
+
+  return await future_copy(ReadableFutureEnd, EventCode.FUTURE_READ, lower,
+                           future_t, opts, task, i)
+
+async def canon_future_write(future_t, opts, task, i, flat_args):
+  def lift():
+    if future_t.t:
+      assert(not contains_borrow(future_t))
+      cx = LiftLowerContext(opts, task.inst, borrow_scope = None)
+      [v] = lift_flat_values(cx, MAX_FLAT_ASYNC_PARAMS, CoreValueIter(flat_args), [future_t.t])
+      return v
+
+  return await future_copy(WritableFutureEnd, EventCode.FUTURE_WRITE, lift,
+                           future_t, opts, task, i)
+
+async def future_copy(EndT, event_code, copy, future_t, opts, task, i):
+  trap_if(not task.inst.may_leave)
+  e = task.inst.table.get(i)
+  trap_if(not isinstance(e, EndT))
+  trap_if(e.shared.t != future_t.t)
+  trap_if(e.copying or e.done)
+
+  def on_result(result):
+    def copy_event():
+      assert(e.copying)
+      e.copying = False
+      if result != FutureResult.CANCELLED:
+        e.done = True
+      return (event_code, i, int(result))
+    e.set_event(copy_event)
+
+  result = e.copy(task.inst, copy, on_result)
+  if result is not None:
+    if result != FutureResult.CANCELLED:
+      e.done = True
+    return [int(result)]
+  else:
+    e.copying = True
+    if opts.sync:
+      await task.wait_on(e.wait_for_pending_event(), sync = True)
+      code,index,payload = e.get_event()
+      assert(code == event_code and index == i)
+      return [payload]
+    else:
+      return [BLOCKED]
+
 ### 🔀 `canon {stream,future}.cancel-{read,write}`
 
 async def canon_stream_cancel_read(stream_t, sync, task, i):
@@ -2260,21 +2394,21 @@ async def cancel_copy(EndT, event_code, stream_or_future_t, sync, task, i):
   assert(not e.copying and code == event_code and index == i)
   return [payload]
 
-### 🔀 `canon {stream,future}.close-{readable,writable}`
+### 🔀 `canon {stream,future}.drop-{readable,writable}`
 
-async def canon_stream_close_readable(stream_t, task, i):
-  return await close(ReadableStreamEnd, stream_t, task, i)
+async def canon_stream_drop_readable(stream_t, task, i):
+  return await drop(ReadableStreamEnd, stream_t, task, i)
 
-async def canon_stream_close_writable(stream_t, task, hi):
-  return await close(WritableStreamEnd, stream_t, task, hi)
+async def canon_stream_drop_writable(stream_t, task, hi):
+  return await drop(WritableStreamEnd, stream_t, task, hi)
 
-async def canon_future_close_readable(future_t, task, i):
-  return await close(ReadableFutureEnd, future_t, task, i)
+async def canon_future_drop_readable(future_t, task, i):
+  return await drop(ReadableFutureEnd, future_t, task, i)
 
-async def canon_future_close_writable(future_t, task, hi):
-  return await close(WritableFutureEnd, future_t, task, hi)
+async def canon_future_drop_writable(future_t, task, hi):
+  return await drop(WritableFutureEnd, future_t, task, hi)
 
-async def close(EndT, stream_or_future_t, task, hi):
+async def drop(EndT, stream_or_future_t, task, hi):
   trap_if(not task.inst.may_leave)
   e = task.inst.table.remove(hi)
   trap_if(not isinstance(e, EndT))