Add 'stream' type

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -166,6 +166,10 @@ class Own(ValType):
 class Borrow(ValType):
   rt: ResourceType
 
+@dataclass
+class Stream(ValType):
+  t: ValType
+
 ### Despecialization
 
 def despecialize(t):
@@ -193,6 +197,7 @@ def alignment(t):
     case Variant(cases)     : return alignment_variant(cases)
     case Flags(labels)      : return alignment_flags(labels)
     case Own(_) | Borrow(_) : return 4
+    case Stream(_)          : return 4
 
 def alignment_record(fields):
   a = 1
@@ -243,6 +248,7 @@ def elem_size(t):
     case Variant(cases)     : return elem_size_variant(cases)
     case Flags(labels)      : return elem_size_flags(labels)
     case Own(_) | Borrow(_) : return 4
+    case Stream(_)          : return 4
 
 def elem_size_record(fields):
   s = 0
@@ -303,6 +309,8 @@ class ComponentInstance:
   active_sync_task: bool
   pending_sync_tasks: list[asyncio.Future]
   async_subtasks: Table[AsyncSubtask]
+  readable_streams: Table[ReadableStreamElem]
+  writable_streams: Table[WritableStreamElem]
 
   def __init__(self):
     self.may_leave = True
@@ -314,6 +322,8 @@ def __init__(self):
     self.active_sync_task = False
     self.pending_sync_tasks = []
     self.async_subtasks = Table[AsyncSubtask]()
+    self.readable_streams = Table[ReadableStreamElem]()
+    self.writable_streams = Table[WritableStreamElem]()
 
 class HandleTables:
   rt_to_table: MutableMapping[ResourceType, Table[HandleElem]]
@@ -388,6 +398,12 @@ def __init__(self, rep, own, scope = None):
     self.scope = scope
     self.lend_count = 0
 
+class StreamElem:
+  t: ValType
+
+  def __init__(self, t):
+    self.t = t
+
 class AsyncCallState(IntEnum):
   STARTING = 0
   STARTED = 1
@@ -400,6 +416,9 @@ class EventCode(IntEnum):
   CALL_RETURNED = AsyncCallState.RETURNED
   CALL_DONE = AsyncCallState.DONE
   YIELDED = 4
+  STREAM_READ = 5
+  STREAM_CLOSED = 6
+  STREAM_WROTE = 7
 
 current_task = asyncio.Lock()
 
@@ -489,6 +508,12 @@ def async_subtask_made_progress(self, subtask):
     subtask.enqueued = True
     self.events.put_nowait(subtask)
 
+  def stream_new(self, t):
+    s = StreamElem(t)
+    rsi = self.inst.readable_streams.add(s)
+    wsi = self.inst.writable_streams.add(s)
+    return (rsi, wsi)
+
   def create_borrow(self):
     self.borrow_count += 1
 
@@ -624,6 +649,7 @@ def load(cx, ptr, t):
     case Flags(labels)  : return load_flags(cx, ptr, labels)
     case Own()          : return lift_own(cx, load_int(cx, ptr, 4), t)
     case Borrow()       : return lift_borrow(cx, load_int(cx, ptr, 4), t)
+    case Stream()       : return lift_stream(cx, load_int(cx, ptr, 4), t)
 
 def load_int(cx, ptr, nbytes, signed = False):
   return int.from_bytes(cx.opts.memory[ptr : ptr+nbytes], 'little', signed=signed)
@@ -772,6 +798,10 @@ def lift_borrow(cx, i, t):
     cx.track_owning_lend(h)
   return h.rep
 
+def lift_stream(cx, i, t):
+  # TODO
+  pass
+
 ### Storing
 
 def store(cx, v, t, ptr):
@@ -797,6 +827,7 @@ def store(cx, v, t, ptr):
     case Flags(labels)  : store_flags(cx, v, ptr, labels)
     case Own()          : store_int(cx, lower_own(cx.opts, v, t), ptr, 4)
     case Borrow()       : store_int(cx, lower_borrow(cx.opts, v, t), ptr, 4)
+    case Stream()       : store_int(cx, lower_stream(cx.opts, v, t), ptr, 4)
 
 def store_int(cx, v, ptr, nbytes, signed = False):
   cx.opts.memory[ptr : ptr+nbytes] = int.to_bytes(v, nbytes, 'little', signed=signed)
@@ -1052,6 +1083,10 @@ def lower_borrow(cx, rep, t):
   cx.create_borrow()
   return cx.inst.handles.add(t.rt, h)
 
+def lower_stream(cx, rep, t):
+  # TODO
+  pass
+
 ### Flattening
 
 MAX_FLAT_PARAMS = 16
@@ -1101,6 +1136,7 @@ def flatten_type(t):
     case Variant(cases)       : return flatten_variant(cases)
     case Flags(labels)        : return ['i32']
     case Own(_) | Borrow(_)   : return ['i32']
+    case Stream(_)            : return ['i32']
 
 def flatten_record(fields):
   flat = []
@@ -1162,6 +1198,7 @@ def lift_flat(cx, vi, t):
     case Flags(labels)  : return lift_flat_flags(vi, labels)
     case Own()          : return lift_own(cx, vi.next('i32'), t)
     case Borrow()       : return lift_borrow(cx, vi.next('i32'), t)
+    case Stream()       : return lift_stream(cx, vi.next('i32'), t)
 
 def lift_flat_unsigned(vi, core_width, t_width):
   i = vi.next('i' + str(core_width))
@@ -1248,6 +1285,7 @@ def lower_flat(cx, v, t):
     case Flags(labels)  : return lower_flat_flags(v, labels)
     case Own()          : return [lower_own(cx, v, t)]
     case Borrow()       : return [lower_borrow(cx, v, t)]
+    case Stream()       : return [lower_stream(cx, v, t)]
 
 def lower_flat_signed(i, core_bits):
   if i < 0:
@@ -1523,3 +1561,49 @@ async def canon_task_yield(task):
   trap_if(task.opts.callback is not None)
   await task.yield_()
   return []
+
+### 🔀 `canon stream.new`
+
+async def canon_stream_new(t, task, ptr):
+  rsi, wsi = task.stream_new(t)
+  store(task, wsi, U32(), ptr)
+  return [rsi]
+
+### 🔀 `canon stream.write`
+
+async def canon_stream_write(sync, task, wsi, ptr, length, partial):
+  trap_if(length == 0)
+  s = self.inst.writable_streams.get(wsi)
+  if sync:
+    trap_if(s.pending_writes)
+    while partial > 0 and not canceled:
+      if not s.pending_reads:
+        task.inst.calling_sync_import = True
+        await ...
+        task.inst.calling_sync_import = False
+      ... copy as much as you can
+      partial -= ... as much as you could
+    return [nwritten]
+  if s.pending_reads:
+    ... do the read
+    return [nwritten]
+  return [0]
+
+### 🔀 `canon stream.close`
+
+async def canon_stream_close(task, wsi):
+  # TODO
+  pass
+
+### 🔀 `canon stream.read`
+
+async def canon_stream_read(sync, task, rsi, ptr, length):
+  trap_if(length == 0)
+  # TODO
+  pass
+
+### 🔀 `canon stream.drop`
+
+async def canon_stream_drop(task, rsi):
+  # TODO
+  pass