Implement the async pipelines and fsMultiSync

faststreams.nim

@@ -1,6 +1,6 @@
 import
-  faststreams/[inputs, outputs]
+  faststreams/[inputs, outputs, pipelines, multisync]
 
 export
-  inputs, outputs
+  inputs, outputs, pipelines, multisync
 

faststreams/async_backend.nim

@@ -6,6 +6,8 @@ type
     waitAsyncClose
     dontWaitAsyncClose
 
+const debugHelpers* = defined(debugHelpers)
+
 when faststreams_async_backend == "chronos":
   import
     chronos

faststreams/buffers.nim

@@ -3,93 +3,129 @@ import
   stew/[ptrops, ranges/ptr_arith],
   async_backend
 
+export
+  deques
+
 type
   PageSpan* = object
     startAddr*, endAddr*: ptr byte
 
   Page* = object
-    startOffset*: Natural
+    consumedTo*: Natural
-    endOffset*: Natural
+    writtenTo*: Natural
     data*: ref string
 
   PageRef* = ref Page
 
   PageBuffers* = ref object
     pageSize*: Natural
-    maxWriteSize*: Natural
+    maxBufferedBytes*: Natural
-    backPressureLimit*: Natural
 
     queue*: Deque[PageRef]
-    getters: seq[Future[void]]
+    waitingReader*: Future[void]
-    putters: seq[Future[void]]
+    waitingWriter*: Future[void]
 
-    eofReached: bool
+    eofReached*: bool
-
-    totalBytesRead*: Natural
-    totalBytesWritten*: Natural
 
 const
   nimPageSize* = 4096
-  pageMetadataSize* = offsetof(Page, data)
   nimAllocatorMetadataSize* = 32
     # TODO: Get this legally from the Nim allocator.
     # The goal is to make perfect page-aligned allocations
-    # that get fast O(0) treatment.
+    # that go through a fast O(0) path in the allocator.
-  defaultPageSize* = 4096 - (pageMetadataSize + nimAllocatorMetadataSize)
+  defaultPageSize* = 4096 - nimAllocatorMetadataSize
   maxStackUsage* = 16384
 
-proc openArrayToPair*(a: var openarray[byte]): (ptr byte, Natural) =
+when debugHelpers:
+  proc describeBuffers*(context: static string, buffers: PageBuffers) =
+    debugEcho context, " :: buffers"
+    for page in buffers.queue:
+      debugEcho " page ", page.data[][page.consumedTo ..<
+                                      min(page.consumedTo + 16, page.writtenTo)]
+      debugEcho "  len = ", page.data[].len
+      debugEcho "  start = ", page.consumedTo
+      debugEcho "  written to = ", page.writtenTo
+
+  func contents*(buffers: PageBuffers): string =
+    for page in buffers.queue:
+      result.add page.data[][page.consumedTo ..< page.writtenTo - 1]
+else:
+  template describeBuffers*(context: static string, buffers: PageBuffers) =
+    discard
+
+func openArrayToPair*(a: var openarray[byte]): (ptr byte, Natural) =
   (addr a[0], Natural(a.len))
 
-template pageBaseAddr*(page: PageRef): ptr byte =
+template allocationStart*(page: PageRef): ptr byte =
   cast[ptr byte](addr page.data[][0])
 
-func pageStartAddr*(page: PageRef): ptr byte =
+func readableStart*(page: PageRef): ptr byte =
-  offset(cast[ptr byte](addr page.data[][0]), page.startOffset)
+  offset(cast[ptr byte](addr page.data[][0]), page.consumedTo)
 
-func pageEndAddr*(page: PageRef): ptr byte =
+func readableEnd*(page: PageRef): ptr byte =
-  offset(cast[ptr byte](addr page.data[][0]), page.endOffset)
+  offset(cast[ptr byte](addr page.data[][0]), page.writtenTo)
+
+template writableStart*(page: PageRef): ptr byte =
+  readableEnd(page)
+
+func allocationEnd*(page: PageRef): ptr byte =
+  offset(cast[ptr byte](addr page.data[][0]), page.data[].len)
 
 func pageLen*(page: PageRef): Natural =
-  page.endOffset - page.startOffset
+  page.writtenTo - page.consumedTo
 
 template pageChars*(page: PageRef): untyped =
-  let baseAddr = cast[ptr UncheckedArray[char]](pageBaseAddr(page))
+  let baseAddr = cast[ptr UncheckedArray[char]](allocationStart(page))
-  toOpenArray(baseAddr, page.startOffset, page.endOffset - 1)
+  toOpenArray(baseAddr, page.consumedTo, page.writtenTo - 1)
 
-func span*(page: PageRef, writable: static[bool] = false): PageSpan =
+func obtainReadableSpan*(page: PageRef, writable: static[bool] = false): PageSpan =
-  let baseAddr = page.pageBaseAddr
+  let baseAddr = page.allocationStart
-  PageSpan(startAddr: offset(baseAddr, page.startOffset),
+  result = PageSpan(startAddr: offset(baseAddr, page.consumedTo),
-           endAddr: offset(baseAddr, when writable: page.data[].len
+                    endAddr: offset(baseAddr, page.writtenTo))
-                                     else: page.endOffset))
+  page.consumedTo = page.writtenTo
 
-template writableSpan*(page: PageRef): PageSpan =
+func writableSpan*(page: PageRef): PageSpan =
-  span(page, writable = true)
+  let baseAddr = allocationStart(page)
+  PageSpan(startAddr: offset(baseAddr, page.writtenTo),
+           endAddr: offset(baseAddr, page.data[].len))
+
+func fullSpan*(page: PageRef): PageSpan =
+  let baseAddr = page.allocationStart
+  PageSpan(startAddr: baseAddr, endAddr: offset(baseAddr, page.data[].len))
 
 func initPageBuffers*(pageSize: Natural,
-                      maxWriteSize = high(int)): PageBuffers =
+                      maxBufferedBytes: Natural = 0): PageBuffers =
+  # TODO: remove the unbuferred streams
   if pageSize > 0:
     return PageBuffers(pageSize: pageSize,
-                       maxWriteSize: maxWriteSize)
+                       maxBufferedBytes: maxBufferedBytes)
 
 template allocRef[T: not ref](x: T): ref T =
   let res = new type(x)
   res[] = x
   res
 
+func trackWrittenToEnd*(buffers: PageBuffers) =
+  if buffers.queue.len > 0:
+    let page = buffers.queue.peekLast
+    page.writtenTo = page.data[].len
+
+func trackWrittenTo*(buffers: PageBuffers, spanHeadPos: ptr byte) =
+  if buffers != nil and buffers.queue.len > 0:
+    var topPage = buffers.queue.peekLast
+    topPage.writtenTo = distance(topPage.allocationStart, spanHeadPos)
+
 func addWritablePage*(buffers: PageBuffers, pageSize: Natural): PageRef =
-  result = PageRef(data: allocRef newString(pageSize),
+  trackWrittenToEnd(buffers)
-                   endOffset: pageSize)
+  result = PageRef(data: allocRef newString(pageSize))
   buffers.queue.addLast result
 
 func getWritablePage*(buffers: PageBuffers,
                       preferredSize: Natural): PageRef =
-  if buffers.queue.len == 1:
+  if buffers.queue.len > 0:
-    let recycledPage = buffers.queue.peekLast
+    let lastPage = buffers.queue.peekLast
-    if recycledPage.endOffset == 0 and
+    if lastPage.writtenTo < lastPage.data[].len:
-       recycledPage.data[].len == preferredSize:
+      return lastPage
-      recycledPage.endOffset = recycledPage.data[].len
-      return recycledPage
 
   return addWritablePage(buffers, preferredSize)
 
@@ -97,26 +133,120 @@ func addWritablePage*(buffers: PageBuffers): PageRef =
   buffers.addWritablePage(buffers.pageSize)
 
 template getWritableSpan*(buffers: PageBuffers): PageSpan =
-  getWritablePage(buffers, buffers.pageSize).span(writable = true)
+  let page = getWritablePage(buffers, buffers.pageSize)
+  writableSpan(page)
 
-proc getReadableSpan*(buffers: PageBuffers): PageSpan =
+func nextReadableSpan*(buffers: PageBuffers, span: var PageSpan) =
-  if buffers.queue.len > 1:
+  let
+    firstPage = buffers.queue.peekFirst
+    pageReadableEnd = firstPage.readableEnd
+
+  if span.endAddr == nil:
+    doAssert buffers.queue.len > 0
+    span = obtainReadableSpan buffers.queue[0]
+  elif span.endAddr != pageReadableEnd:
+    # Check whether the span points within the current page:
+    doAssert distance(firstPage.allocationStart, span.endAddr) >= 0 and
+             distance(span.endAddr, pageReadableEnd) >= 0
+    span.endAddr = pageReadableEnd
+    firstPage.consumedTo = firstPage.writtenTo
+  else:
+    doAssert buffers.queue.len > 1
     discard buffers.queue.popFirst
+    span = obtainReadableSpan buffers.queue[0]
 
-  buffers.queue[0].span
+func stringFromBytes(src: pointer, srcLen: Natural): string =
+  result = newString(srcLen)
+  copyMem(addr result[0], src, srcLen)
 
-func ensureRunway*(buffers: PageBuffers, neededRunway: Natural): PageSpan =
+func nextAlignedSize*(minSize, pageSize: Natural): Natural =
-  doAssert buffers.queue.len == 0
+  # TODO: This is not perfectly accurate. Revisit later
-  buffers.pageSize = neededRunway
+  ((minSize div pageSize) + 1) * pageSize
-  getWritableSpan(buffers)
+
+func appendUnbufferedWrite*(buffers: PageBuffers,
+                            src: pointer, srcLen: Natural) =
+  if buffers.queue.len == 0:
+    buffers.queue.addLast PageRef(
+      data: allocRef stringFromBytes(src, srcLen),
+      writtenTo: srcLen)
+  else:
+    var
+      src = src
+      srcLen = srcLen
+      lastPage = buffers.queue.peekLast
+      lastPageLen = lastPage.data[].len
+      unusedBytes = lastPageLen - lastPage.writtenTo
+
+    if unusedBytes > 0:
+      let unusedBytesStart = offset(addr lastPage.data[0], lastPage.writtenTo)
+      if unusedBytes >= srcLen:
+        copyMem(unusedBytesStart, src, srcLen)
+        lastPage.writtenTo += srcLen
+        return
+      else:
+        copyMem(unusedBytesStart, src, unusedBytes)
+        lastPage.writtenTo = lastPageLen
+        src = offset(src, unusedBytes)
+        srcLen -= unusedBytes
+
+    let nextPageSize = nextAlignedSize(srcLen, buffers.pageSize)
+    let nextPage = buffers.addWritablePage(nextPageSize)
+
+    copyMem(addr nextPage.data[0], src, srcLen)
+    nextPage.writtenTo = srcLen
+
+template hasDelayedWritesAtPageStart(page: PageRef): bool =
+  page.consumedTo < 0
+
+func ensureRunway*(buffers: PageBuffers,
+                   currentHeadPos: var PageSpan,
+                   neededRunway: Natural) =
+  if currentHeadPos.startAddr == nil:
+    # This is a brand new stream, just like we recomend.
+    let page = buffers.addWritablePage(neededRunway)
+    currentHeadPos = page.fullSpan
+  else:
+    # This is a more complicated path that should almost never
+    # trigger in practice in a typically implemented code that
+    # calls `ensureRunway` at the beggining of a transformation.
+    doAssert buffers.queue.len > 0
+    let currPage = buffers.queue.peekLast
+
+    if currPage.hasDelayedWritesAtPageStart:
+      # There is not much we can do here. The outstanding cursors
+      # may point to the current page. We won't honor the runway
+      # request.
+      return
+
+    let
+      oldData = currPage.data
+      bytesWrittenToCurrPage = distance(currPage.readableStart,
+                                        currentHeadPos.startAddr)
+      replacementPageSize = neededRunway + bytesWrittenToCurrPage
+
+    currPage.data = allocRef newString(replacementPageSize)
+    currPage.consumedTo = bytesWrittenToCurrPage
+    currPage.writtenTo = bytesWrittenToCurrPage
+
+    # We copy the old data over the new page
+    copyMem(addr currPage.data[][0], addr oldData[][0],
+            bytesWrittenToCurrPage)
+
+    currentHeadPos = currPage.writableSpan
 
 template len*(buffers: PageBuffers): Natural =
   buffers.queue.len
 
-func totalBufferredBytes*(buffers: PageBuffers): Natural =
+func totalBufferedBytes*(buffers: PageBuffers): Natural =
-  for i in 1 ..< buffers.queue.len:
+  for i in 0 ..< buffers.queue.len:
     result += buffers.queue[i].pageLen
 
+func canAcceptWrite*(buffers: PageBuffers, writeSize: Natural): bool =
+  true or # TODO Remove this line
+  buffers.queue.len == 0 or
+  buffers.maxBufferedBytes == 0 or
+  buffers.totalBufferedBytes < buffers.maxBufferedBytes
+
 template popFirst*(buffers: PageBuffers): PageRef =
   buffers.queue.popFirst
 
@@ -126,24 +256,15 @@ template `[]`*(buffers: PageBuffers, idx: Natural): PageRef =
 func splitLastPageAt*(buffers: PageBuffers, address: ptr byte) =
   var
     topPage = buffers.queue.peekLast
-    newPage = PageRef()
+    splitPosition = distance(topPage.allocationStart, address)
-    splitPosition = distance(topPage.pageBaseAddr, address)
+    newPage = PageRef(
-
+      data: topPage.data,
-  newPage[] = topPage[]
+      consumedTo: splitPosition,
-  topPage.endOffset = splitPosition
+      writtenTo: splitPosition)
-  newPage.startOffset = splitPosition
 
+  topPage.writtenTo = splitPosition
   buffers.queue.addLast newPage
 
-func endLastPageAt*(buffers: PageBuffers, address: ptr byte) =
-  if buffers != nil and buffers.queue.len > 0:
-    var topPage = buffers.queue.peekLast
-    topPage.endOffset = distance(topPage.pageBaseAddr, address)
-
-func nextAlignedSize*(minSize, pageSize: Natural): Natural =
-  # TODO: This is not perfectly accurate. Revisit later
-  ((minSize div pageSize) + 1) * pageSize
-
 iterator consumePages*(buffers: PageBuffers): PageRef =
   doAssert buffers != nil
 
@@ -161,17 +282,14 @@ iterator consumePages*(buffers: PageBuffers): PageRef =
     discard buffers.queue.popFirst
 
   if recycledPage != nil:
-    recycledPage.startOffset = 0
+    recycledPage.consumedTo = 0
-    recycledPage.endOffset = 0
+    recycledPage.writtenTo = 0
     buffers.queue.addLast recycledPage
 
 iterator consumePageBuffers*(buffers: PageBuffers): (ptr byte, Natural) =
   for page in consumePages(buffers):
-    yield (page.pageStartAddr,
+    yield (page.readableStart,
-           Natural(page.endOffset - page.startOffset))
+           Natural(page.writtenTo - page.consumedTo))
-
-template wasEofReached*(buffers: PageBuffers): bool =
-  buffers.eofReached
 
 # BEWARE! These templates violate the double evaluation
 #         safety measures in order to produce better inlined
@@ -217,15 +335,11 @@ template implementWrites*(buffersParam: PageBuffers,
       let bytesWritten = writeBlock
       # TODO: Can we repair the buffers here?
       if bytesWritten != writeLenVar: raiseError()
-      buffers.totalBytesWritten += bytesWritten
 
   if srcLen > 0:
     doAssert src != nil
     let bytesWritten = writeBlock
     if bytesWritten != writeLenVar: raiseError()
-    # TODO: Fix this after removing the unbuffered streams
-    if buffers != nil:
-      buffers.totalBytesWritten += bytesWritten
 
 type
   ReadFlag* = enum
@@ -253,19 +367,17 @@ template implementSingleRead*(buffersParam: PageBuffers,
       bestPageSize = nextAlignedSize(readLenVar, buffers.pageSize)
       page = getWritablePage(buffers, bestPageSize)
 
-    readStartVar = page.pageStartAddr
+    readStartVar = writableStart(page)
-    readLenVar = page.endOffset - page.startOffset
+    readLenVar = page.data[].len - page.writtenTo
 
     # TODO: what if we exit with an exception here?
     # Are the side-effects of `getWritablePage` above OK to keep?
     bytesRead = readBlock
-    page.endOffset = page.startOffset + bytesRead
+    page.writtenTo += bytesRead
 
   if (bytesRead == 0 and zeroReadIsNotEof notin flags) or
      (partialReadIsEof in flags and bytesRead < readLenVar):
     buffers.eofReached = true
-  else:
-    buffers.totalBytesRead += bytesRead
 
   bytesRead
 

faststreams/chronos_adapters.nim

@@ -20,26 +20,28 @@ const
   closingErrMsg = "Failed to close Chronos transport"
   writeIncompleteErrMsg = "Failed to write all bytes to Chronos transport"
 
-proc chronosCloseWait(t: StreamTransport) {.async, raises: [Defect, IOError].} =
+proc chronosCloseWait(t: StreamTransport)
+                     {.async, raises: [Defect, IOError].} =
   fsTranslateErrors closingErrMsg:
     await t.closeWait()
 
-proc chronosReadOnce(s: ChronosInputStream, dst: pointer, dstLen: Natural): Future[Natural]
+proc chronosReadOnce(s: ChronosInputStream,
-                    {.raises: [IOError, Defect], async.} =
+                     dst: pointer, dstLen: Natural): Future[Natural]
+                    {.async, raises: [IOError, Defect].} =
   fsTranslateErrors readingErrMsg:
     return implementSingleRead(s.buffers, dst, dstLen, {},
                                readStartAddr, readLen):
       await s.transport.readOnce(readStartAddr, readLen)
 
 proc chronosWrites(s: ChronosOutputStream, src: pointer, srcLen: Natural)
-                  {.raises: [IOError, Defect], async.} =
+                  {.async, raises: [IOError, Defect].} =
   fsTranslateErrors writeIncompleteErrMsg:
     implementWrites(s.buffers, src, srcLen, "StreamTransport"
                     writeStartAddr, writeLen):
       await s.transport.write(writeStartAddr, writeLen)
 
 # TODO: Use the Raising type here
-let ChronosInputStreamVTable = InputStreamVTable(
+let chronosInputVTable = InputStreamVTable(
   readSync: proc (s: InputStream, dst: pointer, dstLen: Natural): Natural
                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
     var cs = ChronosInputStream(s)
@@ -61,14 +63,14 @@ let ChronosInputStreamVTable = InputStreamVTable(
 )
 
 func chronosInput*(s: StreamTransport,
-                   pageSize = buffers.defaultPageSize,
+                   pageSize = defaultPageSize,
                    allowWaitFor = false): InputStreamHandle =
-  InputStreamHandle(s: ChronosInputStream(
+  makeHandle ChronosInputStream(
-    vtable: vtableAddr ChronosInputStreamVTable,
+    vtable: vtableAddr chronosInputVTable,
-    pageSize: pageSize,
+    buffers: initPageBuffers(pageSize),
-    allowWaitFor: allowWaitFor))
+    allowWaitFor: allowWaitFor)
 
-let ChronosOutputStreamVTable = OutputStreamVTable(
+let chronosOutputVTable = OutputStreamVTable(
   writeSync: proc (s: OutputStream, src: pointer, srcLen: Natural)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
     var cs = ChronosOutputStream(s)
@@ -90,17 +92,11 @@ let ChronosOutputStreamVTable = OutputStreamVTable(
 )
 
 func chronosOutput*(s: StreamTransport,
-                    pageSize = buffers.defaultPageSize,
+                    pageSize = defaultPageSize,
                     allowWaitFor = false): OutputStreamHandle =
-  var stream = ChronosOutputStream(
+  makeHandle ChronosOutputStream(
-    vtable: vtableAddr(SnappyStreamVTable),
+    vtable: vtableAddr(chronosOuputVTable),
-    pageSize: pageSize,
+    buffers: initPageBuffers(pageSize)
-    minWriteSize: 1,
-    maxWriteSize: high(int),
     transport: s,
     allowWaitFor: allowWaitFor)
 
-  stream.initWithSinglePage()
-
-  OutputStreamHandle(s: stream)
-

faststreams/inputs.nim

@@ -8,14 +8,17 @@ export
 
 type
   InputStream* = ref object of RootObj
-    vtable: ptr InputStreamVTable # This is nil for unsafe memory inputs
+    vtable*: ptr InputStreamVTable # This is nil for unsafe memory inputs
-    buffers: PageBuffers          # This is nil for unsafe memory inputs
+    buffers*: PageBuffers          # This is nil for unsafe memory inputs
-    span: PageSpan
+    span*: PageSpan
-    spanEndPos: Natural
+    spanEndPos*: Natural
-    closeFut: Future[void]        # This is nil before `close` is called
+    closeFut: Future[void]         # This is nil before `close` is called
+    when debugHelpers:
+      name*: string
 
   LayeredInputStream* = ref object of InputStream
-    subStream*: InputStream
+    source*: InputStream
+    allowWaitFor*: bool
 
   InputStreamHandle* = object
     s*: InputStream
@@ -34,7 +37,7 @@ type
   CloseAsyncProc* = proc (s: InputStream): Future[void]
                          {.nimcall, gcsafe, raises: [IOError, Defect].}
 
-  GetLenSyncProc* = proc (s: InputStream): Natural
+  GetLenSyncProc* = proc (s: InputStream): Option[Natural]
                          {.nimcall, gcsafe, raises: [IOError, Defect].}
 
   InputStreamVTable* = object
@@ -50,6 +53,12 @@ type
   FileInputStream = ref object of InputStream
     file: File
 
+template Async*(s: InputStream): AsyncInputStream =
+  AsyncInputStream(s)
+
+template Sync*(s: AsyncInputStream): InputStream =
+  InputStream(s)
+
 proc disconnectInputDevice(s: InputStream) =
   # TODO
   # Document the behavior that closeAsync is preferred
@@ -92,14 +101,17 @@ proc close*(s: InputStream,
       else:
         asyncCheck s.closeFut
 
-proc close*(s: AsyncInputStream): Future[void]
+template close*(sp: AsyncInputStream) =
-           {.raises: [IOError, Defect].} =
   ## Starts the asychronous closing of the stream and returns a future that
   ## tracks the closing operation.
-  s.disconnectInputDevice()
+  let s = InputStream sp
-  s.preventFurtherReading()
+  disconnectInputDevice(s)
-  result = InputStream(s).closeFut
+  preventFurtherReading(s)
-  doAssert result != nil
+  if s.closeFut != nil:
+    await s.closeFut
+
+proc closeAsync*(s: AsyncInputStream) {.async.} =
+  close s
 
 template closeNoWait*(sp: AsyncInputStream|InputStream) =
   ## Close the stream without waiting even if's async.
@@ -143,7 +155,7 @@ template vtableAddr*(vtable: InputStreamVTable): ptr InputStreamVTable =
   {.noSideEffect.}:
     unsafeAddr vtable
 
-let MemFileInputVTable = InputStreamVTable(
+let memFileInputVTable = InputStreamVTable(
   closeSync: proc (s: InputStream)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
     try:
@@ -151,9 +163,9 @@ let MemFileInputVTable = InputStreamVTable(
     except OSError as err:
       raise newException(IOError, "Failed to close file", err)
   ,
-  getLenSync: proc (s: InputStream): Natural
+  getLenSync: proc (s: InputStream): Option[Natural]
                    {.nimcall, gcsafe, raises: [IOError, Defect].} =
-    s.span.len
+    some s.span.len
 )
 
 proc memFileInput*(filename: string, mappedSize = -1, offset = 0): InputStreamHandle
@@ -191,7 +203,7 @@ proc memFileInput*(filename: string, mappedSize = -1, offset = 0): InputStreamHa
     mappedSize = memFile.size
 
   makeHandle MemFileInputStream(
-    vtable: vtableAddr MemFileInputVTable,
+    vtable: vtableAddr memFileInputVTable,
     span: PageSpan(
       startAddr: head,
       endAddr: offset(head, mappedSize)),
@@ -203,19 +215,37 @@ proc readableNow*(s: InputStream): bool =
 template readableNow*(s: AsyncInputStream): bool =
   readableNow InputStream(s)
 
+func flipPage(s: InputStream) =
+  doAssert s.buffers.len > 1
+  discard s.buffers.popFirst
+  s.span = obtainReadableSpan s.buffers[0]
+  s.spanEndPos += s.span.len
+
+func getBestContiguousRunway(s: InputStream): Natural =
+  result = s.span.len
+  if result == 0:
+    if s.buffers != nil and s.buffers.len > 1:
+      flipPage s
+      result = s.span.len
+
 func totalUnconsumedBytes*(s: InputStream): Natural =
   ## Returns the number of bytes that are currently sitting within the stream
   ## buffers and that can be consumed with `read` or `advance`.
-  result = s.span.len
+  let
-  if s.buffers != nil:
+    localRunway = s.span.len
-    result += s.buffers.totalBufferredBytes
+    runwayInBuffers = if s.buffers == nil: 0
-    # result += s.buffers.totalBytesRead - s.spanEndPos
+                      else: s.buffers.totalBufferedBytes
+
+  if localRunway == 0 and runwayInBuffers > 0:
+    flipPage s
+
+  localRunway + runwayInBuffers
 
 template totalUnconsumedBytes*(s: AsyncInputStream): Natural =
   ## Alias for InputStream.totalUnconsumedBytes
   totalUnconsumedBytes InputStream(s)
 
-let FileInputVTable = InputStreamVTable(
+let fileInputVTable = InputStreamVTable(
   readSync: proc (s: InputStream, dst: pointer, dstLen: Natural): Natural
                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
     let file = FileInputStream(s).file
@@ -224,7 +254,7 @@ let FileInputVTable = InputStreamVTable(
                         readStartAddr, readLen):
       file.readBuffer(readStartAddr, readLen)
   ,
-  getLenSync: proc (s: InputStream): Natural
+  getLenSync: proc (s: InputStream): Option[Natural]
                    {.nimcall, gcsafe, raises: [IOError, Defect].} =
     let
       s = FileInputStream(s)
@@ -235,7 +265,7 @@ let FileInputVTable = InputStreamVTable(
     let endPos = getFilePos(s.file)
     setFilePos(s.file, preservedPos)
 
-    endPos - preservedPos + runway
+    some Natural(endPos - preservedPos + runway)
   ,
   closeSync: proc (s: InputStream)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
@@ -266,7 +296,7 @@ proc fileInput*(filename: string,
     setFilePos(file, offset)
 
   makeHandle FileInputStream(
-    vtable: vtableAddr FileInputVTable,
+    vtable: vtableAddr fileInputVTable,
     buffers: initPageBuffers(pageSize),
     file: file)
 
@@ -284,20 +314,44 @@ proc unsafeMemoryInput*(str: string): InputStreamHandle =
 
 proc len*(s: InputStream): Option[Natural] {.raises: [Defect, IOError].} =
   if s.vtable == nil:
-    some s.span.len
+    some s.totalUnconsumedBytes
   elif s.vtable.getLenSync != nil:
-    some s.vtable.getLenSync(s)
+    s.vtable.getLenSync(s)
   else:
     none Natural
 
-template len*(s: AsyncInputStream): int =
+template len*(s: AsyncInputStream): Option[Natural] =
   len InputStream(s)
 
-proc flipPage(s: InputStream) =
+func memoryInput*(buffers: PageBuffers): InputStreamHandle =
-  doAssert s.buffers.len > 1
+  var span = if buffers.len == 0: default(PageSpan)
-  discard s.buffers.popFirst
+             else: obtainReadableSpan buffers.queue[0]
-  s.span = s.buffers[0].span
+
-  s.spanEndPos += s.span.len
+  makeHandle InputStream(buffers: buffers,
+                         span: span,
+                         spanEndPos: span.len)
+
+func memoryInput*(data: openarray[byte]): InputStreamHandle =
+  let
+    buffers = initPageBuffers(data.len)
+    page = buffers.addWritablePage(data.len)
+    pageSpan = page.fullSpan
+
+  copyMem(pageSpan.startAddr, unsafeAddr data[0], data.len)
+
+  makeHandle InputStream(buffers: buffers,
+                         span: pageSpan,
+                         spanEndPos: data.len)
+
+func memoryInput*(data: openarray[char]): InputStreamHandle =
+  memoryInput charsToBytes(data)
+
+proc resetBuffers*(s: InputStream, buffers: PageBuffers) =
+  # This should be used only on safe memory input streams
+  doAssert s.vtable == nil and s.buffers != nil and buffers.len > 0
+  s.buffers = buffers
+  s.span = obtainReadableSpan buffers.queue[0]
+  s.spanEndPos = s.span.len
 
 proc continueAfterRead(s: InputStream, bytesRead: Natural): bool =
   # Please note that this is extracted into a proc only to reduce the code
@@ -308,11 +362,11 @@ proc continueAfterRead(s: InputStream, bytesRead: Natural): bool =
   # The read might have been incomplete which signals the EOF of the stream.
   # If this is the case, we disconnect the input device which prevents any
   # further attempts to read from it:
-  if wasEofReached(s.buffers):
+  if s.buffers.eofReached:
     disconnectInputDevice(s)
 
   if bytesRead > 0:
-    s.span = s.buffers.getReadableSpan()
+    s.buffers.nextReadableSpan(s.span)
     s.spanEndPos += s.span.len
     return true
   else:
@@ -395,21 +449,20 @@ template readable*(sp: AsyncInputStream): bool =
   ## Async version of `readable`.
   ## The intended API usage is the same. Instead of blocking, an async
   ## stream will use `await` while waiting for more data.
-  let s = sp
+  let s = InputStream sp
   if hasRunway(s.span):
     true
   else:
-    bufferMoreDataImpl(s, fsAsync, readAsync)
+    bufferMoreDataImpl(s, fsAwait, readAsync)
 
 func continueAfterReadN(s: InputStream,
                         runwayBeforeRead, bytesRead: Natural) =
   if runwayBeforeRead == 0 and bytesRead > 0:
-    s.span = s.buffers.getReadableSpan()
+    s.buffers.nextReadableSpan(s.span)
     s.spanEndPos += s.span.len
 
 template readableNImpl(s, n, awaiter, readOp: untyped): bool =
   let runway = totalUnconsumedBytes(s)
-
   if runway >= n:
     true
   elif s.buffers == nil or s.vtable == nil or s.vtable.readOp == nil:
@@ -423,7 +476,7 @@ template readableNImpl(s, n, awaiter, readOp: untyped): bool =
     while true:
       bytesRead += awaiter s.vtable.readOp(s, nil, bytesDeficit)
 
-      if wasEofReached(s.buffers):
+      if s.buffers.eofReached:
         disconnectInputDevice(s)
         res = bytesRead >= bytesDeficit
         break
@@ -472,7 +525,7 @@ template readable*(sp: AsyncInputStream, np: int): bool =
   ## The intended API usage is the same. Instead of blocking, an async
   ## stream will use `await` while waiting for more data.
   let
-    s = sp
+    s = InputStream sp
     n = np
 
   readableNImpl(s, n, fsAwait, readAsync)
@@ -491,7 +544,7 @@ proc peekAt*(s: InputStream, pos: int): byte {.inline.} =
   return peekHead[]
 
 template peekAt*(s: AsyncInputStream, pos: int): byte =
-  peekAt InputStream(s)
+  peekAt InputStream(s), pos
 
 proc advance*(s: InputStream) =
   if hasRunway(s.span):
@@ -509,7 +562,7 @@ proc read*(s: InputStream): byte =
 template read*(s: AsyncInputStream): byte =
   read InputStream(s)
 
-proc drainBuffersInto(s: InputStream, dstAddr: ptr byte, dstLen: Natural): Natural =
+proc drainBuffersInto*(s: InputStream, dstAddr: ptr byte, dstLen: Natural): Natural =
   var
     dst = dstAddr
     remainingBytes = dstLen
@@ -527,25 +580,36 @@ proc drainBuffersInto(s: InputStream, dstAddr: ptr byte, dstLen: Natural): Natur
   if s.buffers != nil:
     # Since we reached the end of the current page,
     # we have to do the equivalent of `flipPage`:
+
+    # TODO: what if the page was extended?
     if s.buffers.len > 0:
       discard s.buffers.popFirst()
 
     for page in consumePages(s.buffers):
       let
-        pageStart = page.pageStartAddr
+        pageStart = page.readableStart
-        pageLen = page.endOffset - page.startOffset
+        pageLen = page.writtenTo - page.consumedTo
+
+      # There are two possible scenarios ahead:
+      # 1) We'll either stop at this page in which case our span will
+      #    point to the end of the page (so, it's fully consumed)
+      # 2) We are going to copy the entire page to the destination
+      #    buffer and we'll continue (so, it's fully consumed again)
+      page.consumedTo = page.writtenTo
 
       if pageLen > remainingBytes:
         # This page has enough data to fill the rest of the buffer:
         copyMem(dst, pageStart, remainingBytes)
-        page.startOffset += remainingBytes
 
         # This page is partially consumed now and we must set our
-        # span to point to its remaining contents. We also need to
+        # span to point to its remaining contents:
-        # know how much our position in the stream has advanced:
+        s.span = PageSpan(startAddr: offset(pageStart, remainingBytes),
-        let bytesDrainedFromBufers = dstLen - runway
+                          endAddr: page.readableEnd)
-        s.span = page.span
+
-        s.spanEndPos += bytesDrainedFromBufers + s.span.len
+        # We also need to know how much our position in the stream
+        # has advanced:
+        let bytesDrainedFromBuffers = dstLen - runway
+        s.spanEndPos += bytesDrainedFromBuffers + s.span.len
 
         # We return the length of the buffer, which means that is
         # has been fully populated:
@@ -575,7 +639,7 @@ template readIntoExImpl(s: InputStream,
 
     bytesRead += awaiter s.vtable.readOp(s, adjustedDst, bytesDeficit)
 
-    if wasEofReached(s.buffers):
+    if s.buffers.eofReached:
       disconnectInputDevice(s)
       break
 
@@ -630,13 +694,6 @@ proc readOnce*(sp: AsyncInputStream): Future[Natural] =
   doAssert s.buffers != nil and s.vtable != nil
   s.vtable.readAsync(s, nil, 0)
 
-proc getBestRunway(s: InputStream): Natural =
-  result = s.span.len
-  if result == 0:
-    if s.buffers != nil and s.buffers.len > 1:
-      s.flipPage
-      result = s.span.len
-
 when defined(windows):
   proc alloca(n: int): ptr byte {.importc, header: "<malloc.h>".}
 else:
@@ -655,7 +712,7 @@ template readNImpl(sp: InputStream,
   let
     s = sp
     n = np
-    runway = getBestRunway(s)
+    runway = getBestContiguousRunway(s)
 
   # Since Nim currently doesn't allow the `makeOpenArray` calls bellow
   # to appear in different branches of an if statement, the code must

faststreams/multisync.nim

@@ -21,12 +21,12 @@ macro fsMultiSync*(body: untyped) =
   for i in 1 ..< asyncProcParams.len:
     let paramsDef = asyncProcParams[i]
     let typ = paramsDef[^2]
-    if sameType(typ, bindSym"InputStream"):
+    if eqIdent(typ, "InputStream"):
       paramsDef[^2] = bindSym "AsyncInputStream"
-    elif sameType(typ, bindSym"OutputStream"):
+    elif eqIdent(typ, "OutputStream"):
       paramsDef[^2] = bindSym "AsyncOutputStream"
 
   result = newStmtList(body, asyncProcBody)
-  if defined(debugSupportAsync):
+  when defined(debugSupportAsync):
     echo result.repr
 

faststreams/outputs.nim

@@ -16,19 +16,22 @@ export
 
 type
   OutputStream* = ref object of RootObj
-    vtable*: ptr OutputStreamVTable   # This is nil for any memory output
+    vtable*: ptr OutputStreamVTable # This is nil for any memory output
-    buffers*: PageBuffers              # This is nil for unsafe memory outputs
+    buffers*: PageBuffers           # This is nil for unsafe memory outputs
-    span: PageSpan
+    span*: PageSpan
-    spanEndPos: Natural
+    spanEndPos*: Natural
     extCursorsCount: int
-    closeFut: Future[void]
+    closeFut: Future[void]          # This is nil before `close` is called
+    when debugHelpers:
+      name*: string
 
   WriteCursor* = object
     span: PageSpan
     stream: OutputStream
 
   LayeredOutputStream* = ref object of OutputStream
-    subStream*: OutputStream
+    destination*: OutputStream
+    allowWaitFor*: bool
 
   OutputStreamHandle* = object
     s*: OutputStream
@@ -66,10 +69,11 @@ type
   FileOutputStream = ref object of OutputStream
     file: File
 
-const
+template Async*(s: OutputStream): AsyncOutputStream =
-  nimAllocatorMetadataSize* = 0
+  AsyncOutputStream(s)
-    # TODO: Get this from Nim's allocator.
+
-    # The goal is to make perfect page-aligned allocations
+template Sync*(s: AsyncOutputStream): OutputStream =
+  OuputStream(s)
 
 proc disconnectOutputDevice(s: OutputStream) =
   if s.vtable != nil:
@@ -82,9 +86,30 @@ proc disconnectOutputDevice(s: OutputStream) =
 template disconnectOutputDevice(s: AsyncOutputStream) =
   disconnectOutputDevice OutputStream(s)
 
+template flushImpl(s: OutputStream, awaiter, writeOp, flushOp: untyped) =
+  doAssert s.extCursorsCount == 0
+  if s.vtable != nil:
+    if s.buffers != nil:
+      trackWrittenTo(s.buffers, s.span.startAddr)
+      awaiter s.vtable.writeOp(s, nil, 0)
+
+    if s.vtable.flushOp != nil:
+      awaiter s.vtable.flushOp(s)
+
+proc flush*(s: OutputStream) =
+  flushImpl(s, noAwait, writeSync, flushSync)
+
+template flush*(sp: AsyncOutputStream) =
+  let s = OutputStream sp
+  flushImpl(s, fsAwait, writeAsync, flushAsync)
+
+proc flushAsync*(s: AsyncOutputStream) {.async.} =
+  flush s
+
 proc close*(s: OutputStream,
             behavior = dontWaitAsyncClose)
            {.raises: [IOError, Defect].} =
+  flush s
   disconnectOutputDevice(s)
   if s.closeFut != nil:
     fsTranslateErrors "Stream closing failed":
@@ -93,11 +118,15 @@ proc close*(s: OutputStream,
       else:
         asyncCheck s.closeFut
 
-proc close*(s: AsyncOutputStream): Future[void]
+template close*(sp: AsyncOutputStream) =
-           {.raises: [IOError, Defect].} =
+  let s = OutputStream sp
+  flush(Async s)
   disconnectOutputDevice(s)
-  result = OutputStream(s).closeFut
+  if s.closeFut != nil:
-  doAssert result != nil
+    await s.closeFut
+
+proc closeAsync*(s: AsyncOutputStream) {.async.} =
+  close s
 
 template closeNoWait*(sp: AsyncOutputStream|OutputStream) =
   ## Close the stream without waiting even if's async.
@@ -135,8 +164,11 @@ template isExternalCursor(c: var WriteCursor): bool =
   addr(c) != addr(c.stream.cursor)
 
 proc addPage(s: OutputStream) =
-  s.span = s.buffers.addWritablePage().writableSpan
+  let
-  s.spanEndPos += s.span.len
+    nextPageSize = s.buffers.pageSize
+    nextPage = s.buffers.addWritablePage(nextPageSize)
+  s.span = nextPage.fullSpan
+  s.spanEndPos += nextPageSize
 
 template makeHandle*(sp: OutputStream): OutputStreamHandle =
   let s = sp
@@ -144,6 +176,8 @@ template makeHandle*(sp: OutputStream): OutputStreamHandle =
 
 proc memoryOutput*(pageSize = defaultPageSize): OutputStreamHandle =
   doAssert pageSize > 0
+  # We are not creating an initial output page, because `ensureRunway`
+  # can determine the most appropriate size.
   makeHandle OutputStream(buffers: initPageBuffers(pageSize))
 
 proc unsafeMemoryOutput*(buffer: pointer, len: Natural): OutputStreamHandle =
@@ -158,22 +192,18 @@ proc ensureRunway*(s: OutputStream, neededRunway: Natural) =
   ## hint specified at stream creation with `pageSize`.
   let runway = s.span.len
 
-  # This is a temporary requirement.
-  # ensureRunway should be called immediately after creating the OutputStream
-  # In the future, we'll relax this by implementing more logic in buffers.nim
-  doAssert runway == 0, "call ensureRunway immediately after stream creation"
-
   if neededRunway > runway:
     # If you use an unsafe memory output, you must ensure that
     # it will have a large enough size to hold the data you are
     # feeding to it.
     doAssert s.buffers != nil, "Unsafe memory output of insufficient size"
-    s.span = s.buffers.ensureRunway(neededRunway - runway)
+    s.buffers.ensureRunway(s.span, neededRunway)
+    s.spanEndPos += (s.span.len - runway)
 
 template ensureRunway*(s: AsyncOutputStream, neededRunway: Natural) =
-  ensureRunway OutputStream(s, neededRunway)
+  ensureRunway OutputStream(s), neededRunway
 
-let FileOutputVTable = OutputStreamVTable(
+let fileOutputVTable = OutputStreamVTable(
   writeSync: proc (s: OutputStream, src: pointer, srcLen: Natural)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
     var file = FileOutputStream(s).file
@@ -205,7 +235,7 @@ proc fileOutput*(filename: string,
   let f = open(filename, fileMode)
 
   makeHandle FileOutputStream(
-    vtable: vtableAddr FileOutputVTable,
+    vtable: vtableAddr fileOutputVTable,
     buffers: initPageBuffers(pageSize),
     file: f)
 
@@ -215,6 +245,33 @@ proc pos*(s: OutputStream): int =
 template pos*(s: AsyncOutputStream): int =
   pos OutputStream(s)
 
+proc getBuffers*(s: OutputStream): PageBuffers =
+  doAssert s.buffers != nil
+  s.buffers.trackWrittenTo s.span.startAddr
+  return s.buffers
+
+proc recycleBuffers*(s: OutputStream, buffers: PageBuffers) =
+  if buffers != nil:
+    s.buffers = buffers
+
+    let len = buffers.queue.len
+    if len > 0:
+      if len > 1:
+        buffers.queue.shrink(fromLast = len - 1)
+
+      let bufferPage = buffers.queue[0]
+      bufferPage.writtenTo = 0
+      bufferPage.consumedTo = 0
+
+      s.span = bufferPage.fullSpan
+      s.spanEndPos = s.span.len
+      return
+  else:
+    s.buffers = initPageBuffers(defaultPageSize)
+
+  s.span = default(PageSpan)
+  s.spanEndPos = 0
+
 #
 # Pre-conditions for `drainAllBuffers(Sync/Async)`
 #  * The cursor has reached the current span end
@@ -264,13 +321,13 @@ proc delayFixedSizeWrite*(s: OutputStream, size: Natural): WriteCursor =
       runwayDeficit = size - runway
       nextPageSize = nextAlignedSize(runwayDeficit, s.buffers.pageSize)
       nextPage = s.buffers.addWritablePage(nextPageSize)
-      nextPageSpan = nextPage.writableSpan
+      nextPageSpan = nextPage.fullSpan
 
     s.span = PageSpan(startAddr: offset(nextPageSpan.startAddr, runwayDeficit),
                       endAddr: nextPageSpan.endAddr)
 
     # See the explanation about split cursors above
-    nextPage.startOffset = -runwayDeficit
+    nextPage.consumedTo = -runwayDeficit
 
     s.spanEndPos += nextPageSize
 
@@ -293,12 +350,14 @@ proc delayVarSizeWrite*(s: OutputStream, maxSize: Natural): VarSizeWriteCursor =
     s.span.startAddr = endAddr
 
   else:
-    s.buffers.endLastPageAt(s.span.startAddr)
     let
       nextPageSize = nextAlignedSize(maxSize, s.buffers.pageSize)
-      nextPageSpan = s.buffers.addWritablePage(nextPageSize).writableSpan
+      nextPage = s.buffers.addWritablePage(nextPageSize)
+      nextPageSpan = nextPage.fullSpan
       cursorEndAddr = offset(nextPageSpan.startAddr, maxSize)
 
+    nextPage.consumedTo = -maxSize
+
     result = VarSizeWriteCursor WriteCursor(
       stream: s,
       span: PageSpan(startAddr: nextPageSpan.startAddr,
@@ -324,17 +383,17 @@ proc finalWrite*(c: var VarSizeWriteCursor, data: openArray[byte]) =
   doAssert overestimatedBytes >= 0
 
   for page in items(cursor.stream.buffers.queue):
-    let baseAddr = page.pageBaseAddr
+    let baseAddr = page.allocationStart
-    if page.pageEndAddr == cursor.span.endAddr:
+    if page.allocationEnd == cursor.span.endAddr:
       # This is a page ending cursor
-      page.endOffset = distance(baseAddr, cursor.span.startAddr) + data.len
+      page.writtenTo = distance(baseAddr, cursor.span.startAddr) + data.len
       copyMem(cursor.span.startAddr, unsafeAddr data[0], data.len)
       finalize cursor
       return
 
     if cursor.span.startAddr == baseAddr:
       # This is page starting cursor
-      page.startOffset = overestimatedBytes
+      page.consumedTo = overestimatedBytes
       copyMem(offset(baseAddr, overestimatedBytes), unsafeAddr data[0], data.len)
       finalize cursor
       return
@@ -351,10 +410,10 @@ proc tryMovingToNextPage(c: var WriteCursor) =
   # page is big enough to hold all the data. When we created the cursor,
   # we've taken a note regarding the number of bytes on the second page
   # that are reserved by writing them as a negative value for the page
-  # `startOffset`.
+  # `consumedTo`.
   #
   # All we need to do here is update the cursor span to point to the next
-  # page and set the now final `endAddr`. The page `startOffset` is updated
+  # page and set the now final `endAddr`. The page `consumedTo` is updated
   # to 0 to indicate that the cursor has made the flip.
   #
   # If you are wondering, var-sized cursors cannot be split, because our
@@ -363,13 +422,13 @@ proc tryMovingToNextPage(c: var WriteCursor) =
   # When we try to create a var-sized cursor, we check if there are enough
   # bytes on the current page to contain the worst case scenario (the var
   # sized cursor has an upper size limit). If there are enough bytes, we
-  # end the page prematurely (it will end up with an `endOffset`). We can
+  # end the page prematurely (it will end up with an `writtenTo`). We can
   # then recycle the same memory for the next page that will use an adjusted
-  # `startOffset`. The `endOffset` of the first page will be written when
+  # `consumedTo`. The `writtenTo` of the first page will be written when
   # the cursor is finalized and its final size becomes known.
   #
   # If there weren't enough bytes (a much more rare event), we allocate a
-  # new page. We adjust the `endOffset` of the current page to mark it's
+  # new page. We adjust the `writtenTo` of the current page to mark it's
   # premature end and we mark the cursor as special by writing a
 
   # The split cursor is definetely not on the last page, so we can iterate
@@ -377,11 +436,11 @@ proc tryMovingToNextPage(c: var WriteCursor) =
   var prevPage = c.stream.buffers.queue[0]
   for i in 1 ..< c.stream.buffers.queue.len:
     let page = c.stream.buffers.queue[i]
-    if c.span.endAddr == prevPage.pageEndAddr and page.startOffset < 0:
+    if c.span.endAddr == prevPage.allocationEnd and page.consumedTo < 0:
       # We found what we need, so let's get to business:
-      c.span.startAddr = page.pageBaseAddr
+      c.span.startAddr = page.allocationStart
-      c.span.endAddr = offset(c.span.startAddr, -page.startOffset)
+      c.span.endAddr = offset(c.span.startAddr, -page.consumedTo)
-      page.startOffset = 0
+      page.consumedTo = 0
       return
     prevPage = page
 
@@ -390,27 +449,8 @@ proc tryMovingToNextPage(c: var WriteCursor) =
   # pre-allocated cursor span, which is considered a Defect (a range error)
   doAssert false, "Attempt to write past the end of a cursor"
 
-template flushImpl(s: OutputStream, awaiter, writeOp, flushOp: untyped) =
-  doAssert s.extCursorsCount == 0
-  if s.vtable != nil:
-    if s.buffers != nil:
-      s.buffers.endLastPageAt s.span.startAddr
-      awaiter s.vtable.writeOp(s, nil, 0)
-      s.span = s.buffers.getWritableSpan()
-      s.spanEndPos += s.span.len
-
-    if s.vtable.flushOp != nil:
-      awaiter s.vtable.flushOp(s)
-
-proc flush*(s: OutputStream) =
-  flushImpl(s, noAwait, writeSync, flushSync)
-
-template flush*(s: AsyncOutputStream) =
-  let s = sp
-  flushImpl(s, fsAwait, writeAsync, flushAsync)
-
 template writeByteImpl(s: OutputStream, b: byte, awaiter, writeOp, drainOp: untyped) =
-  if s.span.atEnd:
+  if atEnd(s.span):
     # Unsafe memory outputs don't use pages at all, so if our cursor
     # reached here, this is a range violation defect:
     doAssert canExtendOutput(s)
@@ -424,12 +464,13 @@ template writeByteImpl(s: OutputStream, b: byte, awaiter, writeOp, drainOp: unty
     elif s.buffers == nil:
       awaiter s.vtable.writeOp(nil, unsafeAddr b, 1)
     else:
+      trackWrittenToEnd(s.buffers)
       awaiter drainOp(s, nil, 0)
 
   writeByte(s.span, b)
 
 proc write*(c: var WriteCursor, b: byte) =
-  if c.span.atEnd:
+  if atEnd(c.span):
     # The cursor has reached the end of its buffer, but it may be a
     # split cursor. If that's the case, the following function will
     # succeed. If that's not a split cursor, we'll raise a Defect.
@@ -440,16 +481,20 @@ proc write*(c: var WriteCursor, b: byte) =
 proc write*(s: OutputStream, b: byte) =
   writeByteImpl(s, b, noAwait, writeSync, drainAllBuffersSync)
 
-template write*(s: AsyncOutputStream, b: byte) =
+proc write*(sp: AsyncOutputStream, b: byte) =
-  # TODO: I should do something with the write async Futures
+  let s = OutputStream sp
-  bind write
+  if atEnd(s.span):
-  write OutputStream(s)
+    addPage(s)
+  writeByte(s.span, b)
 
 template writeAndWait*(sp: AsyncOutputStream, b: byte) =
   let s = sp
   writeByteImpl(s, b, fsAwait, writeAsync, drainAllBuffersAsync)
 
-template write*(s: OutputStream|AsyncOutputStream|var WriteCursor, x: char) =
+template write*(s: AsyncOutputStream, x: char) =
+  write s, byte(x)
+
+template write*(s: OutputStream|var WriteCursor, x: char) =
   bind write
   write s, byte(x)
 
@@ -472,8 +517,8 @@ proc writeToANewPage(s: OutputStream, bytes: openArray[byte]) =
   let nextPageSize = nextAlignedSize(inputLen, s.buffers.pageSize)
   let nextPage = s.buffers.addWritablePage(nextPageSize)
 
-  s.span = nextPage.writableSpan
+  s.span = nextPage.fullSpan
-  s.spanEndPos += s.span.len
+  s.spanEndPos += nextPageSize
 
   copyMem(s.span.startAddr, inputPos, inputLen)
   s.span.startAddr = offset(s.span.startAddr, inputLen)
@@ -496,7 +541,7 @@ template writeBytesImpl(s: OutputStream,
     # We'll try to create them as large as possible:
     s.writeToANewPage(bytes)
   else:
-    s.buffers.endLastPageAt(s.span.startAddr)
+    trackWrittenTo(s.buffers, s.span.startAddr)
     drainOp
 
 proc write*(s: OutputStream, bytes: openArray[byte]) =
@@ -506,7 +551,7 @@ proc write*(s: OutputStream, bytes: openArray[byte]) =
 proc write*(s: OutputStream, chars: openArray[char]) =
   write s, charsToBytes(chars)
 
-proc write*(s: OutputStream, value: string) {.inline.} =
+proc write*(s: OutputStream|AsyncOutputStream, value: string) {.inline.} =
   write s, value.toOpenArrayByte(0, value.len - 1)
 
 template memCopyToBytes(value: auto): untyped =
@@ -519,30 +564,30 @@ proc writeMemCopy*(s: OutputStream, value: auto) =
   bind write
   write s, memCopyToBytes(value)
 
-proc writeBytesAsyncImpl(sp: AsyncOutputStream,
+proc writeBytesAsyncImpl(sp: OutputStream,
                          bytes: openarray[byte]): Future[void] =
-  let s = OutputStream(sp)
+  let s = sp
   writeBytesImpl(s, bytes):
     return s.vtable.writeAsync(s, unsafeAddr bytes[0], bytes.len)
 
-proc writeBytesAsyncImpl(s: AsyncOutputStream,
+proc writeBytesAsyncImpl(s: OutputStream,
                          chars: openarray[char]): Future[void] =
   writeBytesAsyncImpl s, charsToBytes(chars)
 
-proc writeBytesAsyncImpl(s: AsyncOutputStream,
+proc writeBytesAsyncImpl(s: OutputStream,
                          str: string): Future[void] =
   writeBytesAsyncImpl s, toOpenArray(str, 0, str.len - 1)
 
-template writeAndWait*(sp: AsyncOutputStream, value: auto) =
+template writeAndWait*(sp: AsyncOutputStream, value: untyped) =
   bind writeBytesAsyncImpl
 
   let
-    s = sp
+    s = OutputStream sp
     f = writeBytesAsyncImpl(s, value)
 
   if f != nil:
     fsAwait(f)
-    s.span = s.buffers.getWritableSpan()
+    s.span = getWritableSpan s.buffers
     s.spanEndPos += s.span.len
 
 template writeMemCopyAndWait*(sp: AsyncOutputStream, value: auto) =
@@ -601,9 +646,9 @@ template consumeOutputs*(sp: OutputStream, bytesVar, body: untyped) =
   let s = sp
   doAssert s.extCursorsCount == 0 and s.buffers != nil
 
-  for pageStartAddr, pageLen in consumePageBuffers(s.buffers):
+  for pageReadableStart, pageLen in consumePageBuffers(s.buffers):
     template bytesVar: untyped =
-      makeOpenArray(pageStartAddr, pageLen)
+      makeOpenArray(pageReadableStart, pageLen)
 
     body
 
@@ -632,16 +677,16 @@ template consumeContiguousOutput*(sp: OutputStream, bytesVar, body: untyped) =
 
   if s.buffers.queue.len == 1:
     let page = s.buffers.queue[0]
-    bytesPtr = page.pageStartAddr
+    bytesPtr = page.readableStart
-    bytesLen = page.endOffset - pageStartOffset
+    bytesLen = page.writtenTo - page.consumedTo
     # We need to reset the page to an empty state, so it can be reused
-    page.startOffset = 0
+    page.consumedTo = 0
-    page.endOffset = 0
+    page.writtenTo = 0
   else:
     contigiousBytes = newStringOfCap(s.pos)
 
-    for pageStartAddr, pageLen in consumePageBuffers(s.buffers):
+    for pageReadableStart, pageLen in consumePageBuffers(s.buffers):
-      contigiousBytes.add makeOpenArray(cast[ptr char](pageStartAddr), pageLen)
+      contigiousBytes.add makeOpenArray(cast[ptr char](pageReadableStart), pageLen)
 
     bytesPtr = addr contigiousBytes[0]
     bytesLen = contigiousBytes.len
@@ -658,13 +703,13 @@ proc getOutput*(s: OutputStream, T: type string): string =
   ## Before consuming the output, all outstanding delayed writes must be finalized.
   ##
   doAssert s.extCursorsCount == 0 and s.buffers != nil
-  s.buffers.endLastPageAt s.span.startAddr
+  s.buffers.trackWrittenTo s.span.startAddr
 
   if s.buffers.queue.len == 1:
     let page = s.buffers.queue[0]
-    if page.startOffset == 0:
+    if page.consumedTo == 0:
       result.swap page.data[]
-      result.setLen page.endOffset
+      result.setLen page.writtenTo
       # We clear the buffers, so the stream will be in pristine state.
       # The next write is going to create a fresh new starting page.
       s.buffers.queue.clear()
@@ -680,3 +725,9 @@ template getOutput*(s: OutputStream, T: type seq[byte]): seq[byte] =
 template getOutput*(s: OutputStream): seq[byte] =
   cast[seq[byte]](s.getOutput(string))
 
+template getOutput*(s: AsyncOutputStream): seq[byte] =
+  getOutput OutputStream(s)
+
+template getOutput*(s: AsyncOutputStream, T: type): untyped =
+  getOutput OutputStream(s), T
+

faststreams/pipelines.nim

@@ -1,28 +1,340 @@
 import
   macros,
-  inputs, outputs
+  inputs, outputs, buffers, async_backend
 
 export
-  inputs, outputs
+  inputs, outputs, async_backend
 
-macro executePipeline*(start: InputStream, steps: varargs[untyped]) =
+template clearAndWait(ep: AsyncEvent) =
-  var input = start
+  let e = ep
-  result = newStmtList()
+  clear e
+  await e.wait()
 
-  for i in 0 .. steps.len - 2:
+type
+  FsAsyncPipe* = ref object
+    # TODO: Make these stream handles
+    input*: AsyncInputStream
+    output*: AsyncOutputStream
+    buffers*: PageBuffers
+
+template enterWait(fut: var Future, context: static string) =
+  let wait = newFuture[void](context)
+  fut = wait
+  try: await wait
+  finally: fut = nil
+
+template awake(fp: Future) =
+  let f = fp
+  if f != nil and not finished(f):
+    complete f
+
+proc pipeRead(s: LayeredInputStream,
+              dst: pointer, dstLen: Natural): Future[Natural] {.async.} =
+  let buffers = s.buffers
+  if buffers.eofReached: return 0
+
+  var
+    bytesInBuffersAtStart = buffers.totalBufferedBytes
+    minBytesExpected = max(1, dstLen)
+    bytesInBuffersNow = bytesInBuffersAtStart
+
+  describeBuffers "at start", buffers
+
+  while bytesInBuffersNow < minBytesExpected:
+    awake buffers.waitingWriter
+    echo "About to wait for writer"
+    buffers.waitingReader.enterWait "waiting for writer to buffer more data"
+    echo "Awaken from wait"
+
+    bytesInBuffersNow = buffers.totalBufferedBytes
+    if buffers.eofReached:
+      echo "read bytes ", bytesInBuffersNow - bytesInBuffersAtStart
+      describeBuffers "at end", buffers
+      return bytesInBuffersNow - bytesInBuffersAtStart
+
+  if dst != nil:
+    doAssert drainBuffersInto(s, cast[ptr byte](dst), dstLen) == dstLen
+
+  awake buffers.waitingWriter
+
+  return bytesInBuffersNow - bytesInBuffersAtStart
+
+proc pipeWrite(s: LayeredOutputStream, src: pointer, srcLen: Natural) {.async.} =
+  let buffers = s.buffers
+  echo "pipe write"
+  while buffers.canAcceptWrite(srcLen) == false:
+    buffers.waitingWriter.enterWait "waiting for reader to drain the buffers"
+
+  if src != nil:
+    buffers.appendUnbufferedWrite(src, srcLen)
+
+  awake buffers.waitingReader
+  describeBuffers "pipeWrite", buffers
+
+template completedFuture(name: static string): untyped =
+  let fut = newFuture[void](name)
+  complete fut
+  fut
+
+let pipeInputVTable = InputStreamVTable(
+  readSync: proc (s: InputStream, dst: pointer, dstLen: Natural): Natural
+                 {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    fsTranslateErrors "Failed to read from pipe":
+      let ls = LayeredInputStream(s)
+      doAssert ls.allowWaitFor
+      return waitFor pipeRead(ls, dst, dstLen)
+  ,
+  readAsync: proc (s: InputStream, dst: pointer, dstLen: Natural): Future[Natural]
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    fsTranslateErrors "Unexpected error from the async macro":
+      let ls = LayeredInputStream(s)
+      return pipeRead(ls, dst, dstLen)
+  ,
+  getLenSync: proc (s: InputStream): Option[Natural]
+                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    let source = LayeredInputStream(s).source
+    if source != nil:
+      return source.len
+  ,
+  closeSync: proc (s: InputStream)
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    let source = LayeredInputStream(s).source
+    if source != nil:
+      close source
+  ,
+  closeAsync: proc (s: InputStream): Future[void]
+                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    fsTranslateErrors "Unexpected error from the async macro":
+      let source = LayeredInputStream(s).source
+      if source != nil:
+        return closeAsync(Async source)
+      else:
+        return completedFuture("pipeInput.closeAsync")
+)
+
+let pipeOutputVTable = OutputStreamVTable(
+  writeSync: proc (s: OutputStream, src: pointer, srcLen: Natural)
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    fsTranslateErrors "Failed to write all bytes to pipe":
+      var ls = LayeredOutputStream(s)
+      doAssert ls.allowWaitFor
+      waitFor pipeWrite(ls, src, srcLen)
+  ,
+  writeAsync: proc (s: OutputStream, src: pointer, srcLen: Natural): Future[void]
+                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    # TODO: The async macro is raising exceptions even when
+    #       merely forwarding a future:
+    fsTranslateErrors "Unexpected error from the async macro":
+      return pipeWrite(LayeredOutputStream s, src, srcLen)
+  ,
+  flushSync: proc (s: OutputStream)
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    let destination = LayeredOutputStream(s).destination
+    if destination != nil:
+      flush destination
+  ,
+  flushAsync: proc (s: OutputStream): Future[void]
+                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    fsTranslateErrors "Unexpected error from the async macro":
+      let destination = LayeredOutputStream(s).destination
+      if destination != nil:
+        return flushAsync(Async destination)
+      else:
+        return completedFuture("pipeOutput.flushAsync")
+  ,
+  closeSync: proc (s: OutputStream)
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
+
+    s.buffers.eofReached = true
+    echo "writer closes the stream"
+
+    fsTranslateErrors "Unexpected error from Future.complete":
+      awake s.buffers.waitingReader
+
+    let destination = LayeredOutputStream(s).destination
+    if destination != nil:
+      close destination
+  ,
+  closeAsync: proc (s: OutputStream): Future[void]
+                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    s.buffers.eofReached = true
+
+    fsTranslateErrors "Unexpected error from Future.complete":
+      awake s.buffers.waitingReader
+
+    fsTranslateErrors "Unexpected error from the async macro":
+      let destination = LayeredOutputStream(s).destination
+      if destination != nil:
+        return closeAsync(Async destination)
+      else:
+        return completedFuture("pipeOutput.closeAsync")
+)
+
+func pipeInput*(source: InputStream,
+                pageSize = defaultPageSize,
+                allowWaitFor = false): AsyncInputStream =
+  doAssert pageSize > 0
+
+  AsyncInputStream LayeredInputStream(
+    vtable: vtableAddr pipeInputVTable,
+    buffers: initPageBuffers pageSize,
+    allowWaitFor: allowWaitFor,
+    source: source)
+
+func pipeInput*(buffers: PageBuffers,
+                allowWaitFor = false,
+                source: InputStream = nil): AsyncInputStream =
+  var span = if buffers.len == 0: default(PageSpan)
+             else: obtainReadableSpan buffers.queue[0]
+
+  AsyncInputStream LayeredInputStream(
+    vtable: vtableAddr pipeInputVTable,
+    buffers: buffers,
+    span: span,
+    spanEndPos: span.len,
+    allowWaitFor: allowWaitFor,
+    source: source)
+
+proc pipeOutput*(destination: OutputStream,
+                 pageSize = defaultPageSize,
+                 maxBufferedBytes = defaultPageSize * 4,
+                 allowWaitFor = false): AsyncOutputStream =
+  doAssert pageSize > 0
+
+  var
+    buffers = initPageBuffers pageSize
+    span = buffers.getWritableSpan()
+
+  AsyncOutputStream LayeredOutputStream(
+    vtable: vtableAddr pipeOutputVTable,
+    buffers: buffers,
+    span: span,
+    spanEndPos: span.len,
+    allowWaitFor: allowWaitFor,
+    destination: destination)
+
+proc pipeOutput*(buffers: PageBuffers,
+                 allowWaitFor = false,
+                 destination: OutputStream = nil): AsyncOutputStream =
+  var span = buffers.getWritableSpan()
+
+  AsyncOutputStream LayeredOutputStream(
+    vtable: vtableAddr pipeOutputVTable,
+    buffers: buffers,
+    span: span,
+    # TODO What if the buffers are partially populated?
+    #      Should we adjust the spanEndPos? This would
+    #      need the old buffers.totalBytesWritten var.
+    spanEndPos: span.len,
+    allowWaitFor: allowWaitFor,
+    destination: destination)
+
+func asyncPipe*(pageSize = defaultPageSize,
+                maxBufferedBytes = defaultPageSize * 4): FsAsyncPipe =
+  doAssert pageSize > 0
+  FsAsyncPipe(buffers: initPageBuffers(pageSize, maxBufferedBytes))
+
+func initReader*(pipe: FsAsyncPipe): AsyncInputStream =
+  result = pipeInput(pipe.buffers)
+  pipe.input = result
+
+func initWriter*(pipe: FsAsyncPipe): AsyncOutputStream =
+
+  result = pipeOutput(pipe.buffers)
+  pipe.output = result
+
+proc exchangeBuffersAfterPipilineStep(input: InputStream, output: OutputStream) =
+  let formerInputBuffers = input.buffers
+  let formerOutputBuffers = output.getBuffers
+
+  input.resetBuffers formerOutputBuffers
+  output.recycleBuffers formerInputBuffers
+
+macro executePipeline*(start: InputStream, steps: varargs[untyped]): untyped =
+  result = newTree(nnkStmtListExpr)
+
+  var
+    inputVal = start
+    outputVal = newCall(bindSym"memoryOutput")
+
+    inputVar = genSym(nskVar, "input")
+    outputVar = genSym(nskVar, "output")
+
+    step0 = steps[0]
+
+  result.add quote do:
     var
-      step = steps[i]
+      `inputVar` = `inputVal`
-      outputVar = genSym(nskVar, "out")
+      `outputVar` = OutputStream `outputVal`
-      output = if i == steps.len - 2: steps[^1]
-               else: newCall(bindSym"memoryOutput")
 
+    `step0`(`inputVar`, `outputVar`)
+
+  if steps.len > 2:
+    let step1 = steps[1]
     result.add quote do:
-      var `outputVar` = `output`
+      let formerInputBuffers = `inputVar`.buffers
-      `step`(`input`, `outputVar`)
+      `inputVar` = memoryInput(getBuffers `outputVar`)
+      recycleBuffers(`outputVar`, formerInputBuffers)
+      `step1`(`inputVar`, `outputVar`)
 
-    input = quote do:
+  for i in 2 .. steps.len - 2:
-      unsafeMemoryInput(getOutput(`outputVar`))
+    let step = steps[i]
+    result.add quote do:
+      exchangeBuffersAfterPipilineStep(`inputVar`, `outputVar`)
+      `step`(`inputVar`, `outputVar`)
+
+  var closingCall = steps[^1]
+  closingCall.insert(1, outputVar)
+  result.add closingCall
 
   if defined(debugMacros) or defined(debugPipelines):
     echo result.repr
 
+macro executePipeline*(start: AsyncInputStream, steps: varargs[untyped]): untyped =
+  var
+    stream = ident "stream"
+    pipelineSteps = ident "pipelineSteps"
+    pipelineBody = newTree(nnkStmtList)
+
+    step0 = steps[0]
+    stepOutput = genSym(nskVar, "pipe")
+
+  pipelineBody.add quote do:
+    var `pipelineSteps` = newSeq[Future[void]]()
+    var `stepOutput` = asyncPipe()
+    add `pipelineSteps`, `step0`(`stream`, initWriter(`stepOutput`))
+
+  var
+    stepInput = stepOutput
+
+  for i in 1 .. steps.len - 2:
+    var step = steps[i]
+    stepOutput = genSym(nskVar, "pipe")
+
+    pipelineBody.add quote do:
+      var `stepOutput` = asyncPipe()
+      add `pipelineSteps`, `step`(initReader(`stepInput`), initWriter(`stepOutput`))
+
+    stepInput = stepOutput
+
+  var RetTypeExpr = copy steps[^1]
+  RetTypeExpr.insert(1, newCall("default", ident"AsyncOutputStream"))
+
+  var closingCall = steps[^1]
+  closingCall.insert(1, newDotExpr(stepInput, ident"output"))
+
+  pipelineBody.add quote do:
+    await allFutures(`pipelineSteps`)
+    return `closingCall`
+
+  result = quote do:
+    type RetType = type(`RetTypeExpr`)
+
+    proc pipelineProc(`stream`: AsyncInputStream): Future[RetType] {.async.} =
+      `pipelineBody`
+
+    pipelineProc(`start`)
+
+  when defined(debugMacros):
+    echo result.repr
+

tests/base64.nim

@@ -13,7 +13,12 @@ const
   invalidChar = 255
   paddingByte = byte('=')
 
-proc base64encode*(i: InputStream, o: OutputStream) =
+template encodeSize(size: int): int = (size * 4 div 3) + 6
+
+import
+  ../faststreams/buffers
+
+proc base64encode*(i: InputStream, o: OutputStream) {.fsMultiSync.} =
   var
     n: uint32
     b: uint32
@@ -25,6 +30,10 @@ proc base64encode*(i: InputStream, o: OutputStream) =
   template outputChar(x: typed) =
     o.write cb64[x and 63]
 
+  let inputLen = i.len
+  if inputLen.isSome:
+    o.ensureRunway encodeSize(inputLen.get)
+
   while i.readable(3):
     inputByte(b shl 16)
     inputByte(n or b shl 8)
@@ -48,6 +57,8 @@ proc base64encode*(i: InputStream, o: OutputStream) =
       o.write paddingByte
       o.write paddingByte
 
+  close o
+
 proc initDecodeTable*(): array[256, char] =
   # computes a decode table at compile time
   for i in 0 ..< 256:
@@ -63,7 +74,7 @@ proc initDecodeTable*(): array[256, char] =
 const
   decodeTable = initDecodeTable()
 
-proc base64decode*(i: InputStream, o: OutputStream) =
+proc base64decode*(i: InputStream, o: OutputStream) {.fsMultiSync.} =
   proc decodeSize(size: int): int =
     return (size * 3 div 4) + 6
 
@@ -112,3 +123,5 @@ proc base64decode*(i: InputStream, o: OutputStream) =
   elif i.readable:
     raise newException(ValueError, "The input stream has insufficient nymber of bytes for base64 decoding")
 
+  close o
+

tests/test_inputs.nim

@@ -74,6 +74,11 @@ procSuite "input stream":
       finally:
         close input
 
+  # TODO: fileInput with offset
+  #        - in the middle of the
+  #        - right at the end of the file
+  #        - past the end of the file
+
   asciiTableFileTest "count lines":
     check countLines(input) == 34
 

tests/test_outputs.nim

@@ -81,10 +81,7 @@ suite "output stream":
 
   template checkOutputsMatch(showResults = false,
                              skipUnbufferedFile = false) =
-    flush fileStream
     close fileStream
-
-    flush unbufferedFileStream
     close unbufferedFileStream
 
     check fileExists(fileOutputPath) and
@@ -122,15 +119,15 @@ suite "output stream":
 
     let outputsMatch =
       nimSeq == makeOpenArray(cast[ptr byte](buffer),
-                              streamWritingToExistingBuffer.pos) and
+                              streamWritingToExistingBuffer.pos)
-      nimSeq == memStreamRes and
-      nimSeq == readFileRes and
-      nimSeq == fileInputRes and
-      nimSeq == memFileInputRes and
-      nimSeq == fileInputWithSmallPagesRes
-
     check outputsMatch
 
+    check nimSeq == memStreamRes
+    check nimSeq == readFileRes
+    check nimSeq == fileInputRes
+    check nimSeq == memFileInputRes
+    check nimSeq == fileInputWithSmallPagesRes
+
     when not skipUnbufferedFile:
       let unbufferedFileRes = readFile(unbufferedFileOutputPath).string.bytes
       check nimSeq == unbufferedFileRes
@@ -236,5 +233,6 @@ suite "output stream":
       finalize dw.cursor
 
     # The final outputs are the same
-    check nimSeq == memStream.getOutput
+    let resultsAreEqual = nimSeq == memStream.getOutput
+    check resultsAreEqual
 

tests/test_pipelines.nim

@@ -1,8 +1,13 @@
 {.used.}
 
 import
-  std/[unittest, strutils, base64],
+  # Std lib:
-  ../faststreams/pipelines,
+  std/[strutils, random, base64, terminal],
+  # Other packages:
+  testutils/unittests,
+  # FastStreams modules:
+  ../faststreams/[pipelines, multisync],
+  # Testing modules:
   ./base64 as fsBase64
 
 include system/timers
@@ -13,17 +18,29 @@ type
     fsAsyncPipeline: Nanos
     stdFunctionCalls: Nanos
 
-proc upcaseAllCharacters(i: InputStream, o: OutputStream) =
+proc upcaseAllCharacters(i: InputStream, o: OutputStream) {.fsMultiSync.} =
+  let inputLen = i.len
+  if inputLen.isSome:
+    o.ensureRunway inputLen.get
+
   while i.readable:
-    o.write toUpperAscii(char i.read())
+    o.write toUpperAscii(i.read.char)
+
+  echo "closing upcase"
+  close o
+
+proc printTimes(t: TestTimes) =
+  styledEcho "  cpu time [FS Sync  ]: ", styleBright, $t.fsPipeline, "ms"
+  styledEcho "  cpu time [FS Async ]: ", styleBright, $t.fsAsyncPipeline, "ms"
+  styledEcho "  cpu time [Std Lib  ]: ", styleBright, $t.stdFunctionCalls, "ms"
 
 template timeit(timerVar: var Nanos, code: untyped) =
   let t0 = getTicks()
   code
   timerVar = int(getTicks() - t0) div 1000000
 
-suite "pipelines":
+procSuite "pipelines":
-  var loremIpsum = """
+  let loremIpsum = """
     Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod
     tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim
     veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex
@@ -32,25 +49,83 @@ suite "pipelines":
     cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id
     est laborum.
 
-  """.repeat(100)
+  """
 
-  test "upper-case/base64 pipeline":
+  #[
+  test "upper-case/base64 pipeline benchmark":
     var
       times: TestTimes
       stdRes: string
       fsRes: string
+      fsAsyncRes: string
+
+    let inputText = loremIpsum.repeat(5000)
+
+    when debugHelpers:
+      echo "Input len: ", inputText.len
+      echo "Base 64 len: ", base64.encode(inputText).len
 
     timeIt times.fsPipeline:
-      var memOut = memoryOutput()
+      fsRes = executePipeline(unsafeMemoryInput(inputText),
-      executePipeline(unsafeMemoryInput(loremIpsum),
+                              upcaseAllCharacters,
-                      upcaseAllCharacters,
+                              base64encode,
-                      base64encode,
+                              base64decode,
-                      base64decode,
+                              getOutput string)
-                      memOut)
+
-      fsRes = memOut.getOutput(string)
+    timeIt times.fsAsyncPipeline:
+      fsAsyncRes = waitFor executePipeline(Async unsafeMemoryInput(inputText),
+                                           upcaseAllCharacters,
+                                           base64encode,
+                                           base64decode,
+                                           getOutput string)
 
     timeIt times.stdFunctionCalls:
-      stdRes = base64.decode(base64.encode(toUpperAscii(loremIpsum)))
+      stdRes = base64.decode(base64.encode(toUpperAscii(inputText)))
 
+    check fsAsyncRes == stdRes
     check fsRes == stdRes
 
+    printTimes times
+  ]#
+
+  asyncTest "upper-case/base64 async pipeline":
+    let pipe = asyncPipe()
+    let inputText = repeat(loremIpsum, 8)
+
+    when debugHelpers:
+      echo "Input len: ", inputText.len
+
+    proc pipeFeeder(s: AsyncOutputStream) {.gcsafe, async.} =
+      randomize 1234
+      var pos = 0
+
+      while pos != inputText.len:
+        let bytesToWrite = rand(15)
+
+        if bytesToWrite == 0:
+          s.write inputText[pos]
+          inc pos
+        else:
+          let endPos = min(pos + bytesToWrite, inputText.len)
+          s.writeAndWait inputText[pos ..< endPos]
+          pos = endPos
+
+        let sleep = rand(50) - 45
+        if sleep > 0:
+          echo "written ", pos
+          await sleepAsync(sleep.milliseconds)
+
+      close s
+
+    asyncCheck pipeFeeder(pipe.initWriter)
+
+    let f = executePipeline(pipe.initReader,
+                            upcaseAllCharacters,
+                            base64encode,
+                            base64decode,
+                            getOutput string)
+
+    let fsAsyncres = await f
+
+    check fsAsyncRes == toUpperAscii(inputText)
+