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,28 @@ def __init__(self, rep, own, scope = None):
     self.scope = scope
     self.lend_count = 0
 
+class Buffer:
+  cx: CallContext
+  ptr: int
+  length: int
+
+  def __init__(self, cx, ptr, length):
+    self.cx = cx
+    self.ptr = ptr
+    self.length = length
+
+class StreamElem:
+  t: ValType
+  active: bool
+  writer: Optional[Buffer]
+  reader: Optional[Buffer]
+
+  def __init__(self, t):
+    self.t = t
+    self.active = True
+    self.writer = None
+    self.reader = None
+
 class AsyncCallState(IntEnum):
   STARTING = 0
   STARTED = 1
@@ -400,6 +432,10 @@ class EventCode(IntEnum):
   CALL_RETURNED = AsyncCallState.RETURNED
   CALL_DONE = AsyncCallState.DONE
   YIELDED = 4
+  STREAM_READ = 5
+  STREAM_CLOSED = 6
+  STREAM_WROTE = 7
+  STREAM_CANCELLED = 8
 
 current_task = asyncio.Lock()
 
@@ -489,6 +525,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 +666,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)
@@ -704,11 +747,14 @@ def load_string_from_range(cx, ptr, tagged_code_units):
 def load_list(cx, ptr, elem_type):
   begin = load_int(cx, ptr, 4)
   length = load_int(cx, ptr + 4, 4)
-  return load_list_from_range(cx, begin, length, elem_type)
+  return lift_list(cx, begin, length, elem_type)
 
-def load_list_from_range(cx, ptr, length, elem_type):
+def lift_list(cx, ptr, length, elem_type):
   trap_if(ptr != align_to(ptr, alignment(elem_type)))
   trap_if(ptr + length * elem_size(elem_type) > len(cx.opts.memory))
+  return lift_list_after_checks(cx, ptr, length, elem_type)
+
+def lift_list_after_checks(cx, ptr, length, elem_type):
   a = []
   for i in range(length):
     a.append(load(cx, ptr + i * elem_size(elem_type), elem_type))
@@ -772,6 +818,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 +847,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)
@@ -989,19 +1040,22 @@ def store_probably_utf16_to_latin1_or_utf16(cx, src, src_code_units):
   return (ptr, latin1_size)
 
 def store_list(cx, v, ptr, elem_type):
-  begin, length = store_list_into_range(cx, v, elem_type)
+  begin, length = lower_list(cx, v, elem_type)
   store_int(cx, begin, ptr, 4)
   store_int(cx, length, ptr + 4, 4)
 
-def store_list_into_range(cx, v, elem_type):
+def lower_list(cx, v, elem_type):
   byte_length = len(v) * elem_size(elem_type)
   trap_if(byte_length >= (1 << 32))
   ptr = cx.opts.realloc(0, 0, alignment(elem_type), byte_length)
   trap_if(ptr != align_to(ptr, alignment(elem_type)))
   trap_if(ptr + byte_length > len(cx.opts.memory))
+  lower_list_after_checks(cx, v, ptr, elem_type)
+  return (ptr, len(v))
+
+def lower_list_after_checks(cx, v, ptr, elem_type):
   for i,e in enumerate(v):
     store(cx, e, elem_type, ptr + i * elem_size(elem_type))
-  return (ptr, len(v))
 
 def store_record(cx, v, ptr, fields):
   for f in fields:
@@ -1052,6 +1106,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 +1159,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 +1221,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))
@@ -1184,7 +1244,7 @@ def lift_flat_string(cx, vi):
 def lift_flat_list(cx, vi, elem_type):
   ptr = vi.next('i32')
   length = vi.next('i32')
-  return load_list_from_range(cx, ptr, length, elem_type)
+  return lift_list(cx, ptr, length, elem_type)
 
 def lift_flat_record(cx, vi, fields):
   record = {}
@@ -1248,6 +1308,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:
@@ -1259,7 +1320,7 @@ def lower_flat_string(cx, v):
   return [ptr, packed_length]
 
 def lower_flat_list(cx, v, elem_type):
-  (ptr, length) = store_list_into_range(cx, v, elem_type)
+  (ptr, length) = lower_list(cx, v, elem_type)
   return [ptr, length]
 
 def lower_flat_record(cx, v, fields):
@@ -1523,3 +1584,73 @@ 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`
+
+
+# TODO: do we want read/write to both support full/partial?
+async def canon_stream_write(t, sync, task, wsi, ptr, length, partial):
+  trap_if(length == 0 or length >= 2**31)
+  trap_if(ptr != align_to(ptr, alignment(t)))
+  trap_if(ptr + length > len(task.opts.memory))
+  s = self.inst.writable_streams.get(wsi)
+  trap_if(s.t != t)
+  trap_if(s.writer is not None)
+  s.writer = Buffer(task, ptr, length)
+  while s.writer:
+    if not s.active:
+      return pack_final_result(length - s.writer.length)
+    if not s.reader:
+      # TODO: this is all wrong
+      if not sync:
+        return [length - s.writer.length]
+      task.inst.calling_sync_import = True
+      await ... TODO
+      task.inst.calling_sync_import = False
+    copy_writer_to_reader(s)
+  return [length]
+
+def pack_final_result(nwritten):
+  assert(nwritten < 2**31)
+  return [ (1 << 31) | nwritten ]
+
+def copy_writer_to_reader(writer, reader):
+  copy_length = min(writer.length, reader.length)
+  v = lift_list_after_checks(writer.cx, writer.ptr, copy_length, t)
+  lower_list_after_checks(reader.cx, v, reader.ptr, t)
+  writer.ptr += copy_length * elem_size(t)
+  reader.ptr += copy_length * elem_size(t)
+  writer.length -= copy_length
+  reader.length -= copy_length
+  if writer.length == 0:
+    TODO
+  if reader.length == 0:
+    TODO
+
+### 🔀 `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, partial):
+  trap_if(read_length == 0)
+  trap_if(ptr != align_to(ptr, alignment(t)))
+  trap_if(ptr + read_length > len(task.opts.memory))
+  # TODO
+  pass
+
+### 🔀 `canon stream.drop`
+
+async def canon_stream_drop(task, rsi):
+  # TODO
+  pass