Implement readInto and read(n)

faststreams.nimble

@@ -9,6 +9,7 @@ skipDirs      = @["tests"]
 
 requires "nim >= 1.2.0",
          "stew",
+         "testutils",
          "chronos"
 
 task test, "Run all tests":

faststreams/async_backend.nim

@@ -37,6 +37,8 @@ else:
 template fsTranslateErrors*(errMsg: string, body: untyped) =
   try:
     body
+  except IOError as err:
+    raise err
   except Exception as err:
     if err[] of Defect:
       raise (ref Defect)(err)

faststreams/buffers.nim

@@ -4,22 +4,13 @@ import
   async_backend
 
 type
-  PageKind* = enum
-    userPage
-    stringPage
-    mallocPage
-
   PageSpan* = object
     startAddr*, endAddr*: ptr byte
 
   Page* = object
     startOffset*: Natural
     endOffset*: Natural
-    case kind*: PageKind
+    data*: ref string
-    of userPage, mallocPage:
-      bufferStart, bufferEnd: ptr byte
-    of stringPage:
-      data*: ref string
 
   PageRef* = ref Page
 
@@ -47,38 +38,30 @@ const
   defaultPageSize* = 4096 - (pageMetadataSize + nimAllocatorMetadataSize)
   maxStackUsage* = 16384
 
-func pageBaseAddr*(page: PageRef): ptr byte =
+proc openArrayToPair*(a: var openarray[byte]): (ptr byte, Natural) =
-  if page.kind == stringPage:
+  (addr a[0], Natural(a.len))
-    cast[ptr byte](addr page.data[][0])
+
-  else:
+template pageBaseAddr*(page: PageRef): ptr byte =
-    page.bufferStart
+  cast[ptr byte](addr page.data[][0])
 
 func pageStartAddr*(page: PageRef): ptr byte =
-  if page.kind == stringPage:
+  offset(cast[ptr byte](addr page.data[][0]), page.startOffset)
-    offset(cast[ptr byte](addr page.data[][0]), page.startOffset)
-  else:
-    offset(page.bufferStart, page.startOffset)
 
 func pageEndAddr*(page: PageRef): ptr byte =
-  if page.kind == stringPage:
+  offset(cast[ptr byte](addr page.data[][0]), page.endOffset)
-    offset(cast[ptr byte](addr page.data[][0]), page.endOffset)
+
-  else:
+func pageLen*(page: PageRef): Natural =
-    offset(page.bufferStart, page.endOffset)
+  page.endOffset - page.startOffset
 
 template pageChars*(page: PageRef): untyped =
   let baseAddr = cast[ptr UncheckedArray[char]](pageBaseAddr(page))
   toOpenArray(baseAddr, page.startOffset, page.endOffset - 1)
 
 func span*(page: PageRef, writable: static[bool] = false): PageSpan =
-  if page.kind == stringPage:
+  let baseAddr = page.pageBaseAddr
-    let baseAddr = cast[ptr byte](addr page.data[][0])
+  PageSpan(startAddr: offset(baseAddr, page.startOffset),
-    PageSpan(startAddr: offset(baseAddr, page.startOffset),
+           endAddr: offset(baseAddr, when writable: page.data[].len
-             endAddr: offset(baseAddr, when writable: page.data[].len
+                                     else: page.endOffset))
-                                       else: page.endOffset))
-  else:
-    PageSpan(startAddr: offset(page.bufferStart, page.startOffset),
-             endAddr: when writable: page.bufferEnd
-                      else: offset(page.bufferStart, page.endOffset))
 
 template writableSpan*(page: PageRef): PageSpan =
   span(page, writable = true)
@@ -94,37 +77,46 @@ template allocRef[T: not ref](x: T): ref T =
   res[] = x
   res
 
-func getWritablePage*(buffers: PageBuffers): PageRef =
-  # TODO: The semantics of this func are quite unusual
-  #       I should find a more appropriate name
-  if buffers.queue.len == 0:
-    result = PageRef(kind: stringPage,
-                     data: allocRef newString(buffers.pageSize),
-                     endOffset: buffers.pageSize)
-    buffers.queue.addLast result
-  else:
-    result = buffers.queue[0]
-
 func addWritablePage*(buffers: PageBuffers, pageSize: Natural): PageRef =
-  result = PageRef(kind: stringPage,
+  result = PageRef(data: allocRef newString(pageSize),
-                   data: allocRef newString(pageSize),
                    endOffset: pageSize)
   buffers.queue.addLast result
 
+func getWritablePage*(buffers: PageBuffers,
+                      preferredSize: Natural): PageRef =
+  if buffers.queue.len == 1:
+    let recycledPage = buffers.queue.peekLast
+    if recycledPage.endOffset == 0 and
+       recycledPage.data[].len == preferredSize:
+      recycledPage.endOffset = recycledPage.data[].len
+      return recycledPage
+
+  return addWritablePage(buffers, preferredSize)
+
 func addWritablePage*(buffers: PageBuffers): PageRef =
   buffers.addWritablePage(buffers.pageSize)
 
 template getWritableSpan*(buffers: PageBuffers): PageSpan =
-  getWritablePage(buffers).span(writable = true)
+  getWritablePage(buffers, buffers.pageSize).span(writable = true)
+
+proc getReadableSpan*(buffers: PageBuffers): PageSpan =
+  if buffers.queue.len > 1:
+    discard buffers.queue.popFirst
+
+  buffers.queue[0].span
 
 func ensureRunway*(buffers: PageBuffers, neededRunway: Natural): PageSpan =
   doAssert buffers.queue.len == 0
   buffers.pageSize = neededRunway
   getWritableSpan(buffers)
 
-template len*(buffers: PageBuffers): int =
+template len*(buffers: PageBuffers): Natural =
   buffers.queue.len
 
+func totalBufferredBytes*(buffers: PageBuffers): Natural =
+  for i in 1 ..< buffers.queue.len:
+    result += buffers.queue[i].pageLen
+
 template popFirst*(buffers: PageBuffers): PageRef =
   buffers.queue.popFirst
 
@@ -148,54 +140,36 @@ func endLastPageAt*(buffers: PageBuffers, address: ptr byte) =
     var topPage = buffers.queue.peekLast
     topPage.endOffset = distance(topPage.pageBaseAddr, address)
 
-func trackPageWrite*(page: PageRef, bytesWritten: Natural) {.inline.} =
-  page.endOffset = page.startOffset + bytesWritten
-
-template writeToSpan*(buffersParam: PageBuffers,
-                      spanVarName, writeExpr: untyped) =
-  var
-    buffers = buffersParam
-    page = buffers.getWritablePage
-    spanVarName = page.writableSpan
-
-  # TODO: what if we exit with an exception here?
-  # Are the side-effects of `getWritablePage` above OK to keep?
-
-  let bytesWritten = writeExpr
-  trackPageWrite(page, bytesWritten)
-
-  if bytesWritten == 0:
-    buffers.eofReached = true
-
 func nextAlignedSize*(minSize, pageSize: Natural): Natural =
   # TODO: This is not perfectly accurate. Revisit later
   ((minSize div pageSize) + 1) * pageSize
 
-template consumeAllPages*(buffersParam: PageBuffers,
+iterator consumePages*(buffers: PageBuffers): PageRef =
-                          pageAddrVar, pageLenVar, body: untyped) =
-  let buffers = buffersParam
   doAssert buffers != nil
 
   var recycledPage: PageRef
-  for page in buffers.queue:
+  while buffers.queue.len > 0:
-    let
+    var page = peekFirst(buffers.queue)
-      pageAddrVar = page.pageStartAddr
-      pageLenVar = page.endOffset - page.startOffset
-
-    if page.kind == stringPage and page.data[].len == buffers.pageSize:
-      recycledPage = page
 
     # TODO: what if the body throws an exception?
-    # Should we do anything with the remaining pages?
+    # Should we do anything with the consumed page?
-    body
+    yield page
 
-  buffers.queue.clear()
+    if page.data[].len == buffers.pageSize:
+      recycledPage = page
+
+    discard buffers.queue.popFirst
 
   if recycledPage != nil:
     recycledPage.startOffset = 0
     recycledPage.endOffset = 0
     buffers.queue.addLast recycledPage
 
+iterator consumePageBuffers*(buffers: PageBuffers): (ptr byte, Natural) =
+  for page in consumePages(buffers):
+    yield (page.pageStartAddr,
+           Natural(page.endOffset - page.startOffset))
+
 template wasEofReached*(buffers: PageBuffers): bool =
   buffers.eofReached
 
@@ -224,3 +198,74 @@ template charsToBytes*(chars: openArray[char]): untyped =
   var charsStart = unsafeAddr chars[0]
   makeOpenArray(cast[ptr byte](charsStart), chars.len)
 
+template implementWrites*(buffersParam: PageBuffers,
+                          srcParam: pointer,
+                          srcLenParam: Natural,
+                          dstDesc: static string,
+                          writeStartVar, writeLenVar,
+                          writeBlock: untyped) =
+  let
+    buffers = buffersParam
+    writeStartVar = srcParam
+    writeLenVar = srcLenParam
+
+  template raiseError =
+    raise newException(IOError, "Failed to write all bytes to " & dstDesc)
+
+  if buffers != nil:
+    for writeStartVar, writeLenVar in consumePageBuffers(s.buffers):
+      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
+    partialReadIsEof
+    zeroReadIsNotEof
+
+  ReadFlags* = set[ReadFlag]
+
+template implementSingleRead*(buffersParam: PageBuffers,
+                              dstParam: pointer,
+                              dstLenParam: Natural,
+                              flags: static ReadFlags,
+                              readStartVar, readLenVar,
+                              readBlock: untyped): Natural =
+  var
+    buffers = buffersParam
+    readStartVar = dstParam
+    readLenVar = dstLenParam
+    bytesRead: Natural
+
+  if readStartVar != nil:
+    bytesRead = readBlock
+  else:
+    let
+      bestPageSize = nextAlignedSize(readLenVar, buffers.pageSize)
+      page = getWritablePage(buffers, bestPageSize)
+
+    readStartVar = page.pageStartAddr
+    readLenVar = page.endOffset - page.startOffset
+
+    # 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
+
+  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,40 +20,44 @@ const
   closingErrMsg = "Failed to close Chronos transport"
   writeIncompleteErrMsg = "Failed to write all bytes to Chronos transport"
 
-proc fsCloseWait(t: StreamTransport) {.async, raises: [Defect, IOError].} =
+proc chronosCloseWait(t: StreamTransport) {.async, raises: [Defect, IOError].} =
   fsTranslateErrors closingErrMsg:
     await t.closeWait()
 
-proc fsReadOnce(t: StreamTransport,
+proc chronosReadOnce(s: ChronosInputStream, dst: pointer, dstLen: Natural): Future[Natural]
-                buffer: ptr byte, bufSize: int)
+                    {.raises: [IOError, Defect], async.} =
-               {.raises: [Defect, IOError], async.} =
   fsTranslateErrors readingErrMsg:
-    buffers.writeToSpan(span):
+    return implementSingleRead(s.buffers, dst, dstLen, {},
-      await t.readOnce(span.startAddr, span.len)
+                               readStartAddr, readLen):
+      await s.transport.readOnce(readStartAddr, readLen)
+
+proc chronosWrites(s: ChronosOutputStream, src: pointer, srcLen: Natural)
+                  {.raises: [IOError, Defect], async.} =
+  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(
-  readSync: proc (s: InputStream, buffers: PageBuffers)
+  readSync: proc (s: InputStream, dst: pointer, dstLen: Natural): Natural
                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
     var cs = ChronosInputStream(s)
     doAssert cs.allowWaitFor
-
+    waitFor chronosReadOnce(cs, dst, dstLen)
-    fsTranslateErrors readingErrMsg:
-      buffers.writeToSpan(span):
-        waitFor cs.transport.readOnce(span.startAddr, span.len)
   ,
-  readAsync: proc (s: InputStream, buffers: PageBuffers): Future[Natural]
+  readAsync: proc (s: InputStream, dst: pointer, dstLen: Natural): Future[Natural]
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
-    ChronosInputStream(s).transport.fsReadOnce(buffers)
+    chronosReadOnce(ChronosInputStream s, dst, dstLen)
   ,
   closeSync: proc (s: InputStream)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
     fsTranslateErrors closingErrMsg:
-      ChronosInputStream(s).transport.close()
+      s.closeFut = ChronosInputStream(s).transport.close()
   ,
   closeAsync: proc (s: InputStream): Future[void]
                    {.nimcall, gcsafe, raises: [IOError, Defect].} =
-    ChronosInputStream(s).transport.fsCloseWait()
+    chronosCloseWait ChronosInputStream(s).transport
 )
 
 func chronosInput*(s: StreamTransport,
@@ -65,49 +69,24 @@ func chronosInput*(s: StreamTransport,
     allowWaitFor: allowWaitFor))
 
 let ChronosOutputStreamVTable = OutputStreamVTable(
-  writePageSync: proc (s: OutputStream, page: openarray[byte])
+  writeSync: proc (s: OutputStream, src: pointer, srcLen: Natural)
-                      {.nimcall, gcsafe, raises: [IOError, Defect].} =
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
     var cs = ChronosOutputStream(s)
     doAssert cs.allowWaitFor
-    let bytesWritten = fsTranslateErrors writingErrMsg:
+    waitFor chronosWrites(cs, src, srcLen)
-      waitFor cs.transport.write(unsafeAddr page[0], page.len)
-    if bytesWritten != page.len:
-      raise newException(IOError, writeIncompleteErrMsg)
   ,
-  writePageAsync: proc (s: OutputStream, buf: pointer, bufLen: int): Future[void]
+  writeAsync: proc (s: OutputStream, src: pointer, srcLen: Natural): Future[void]
-                       {.nimcall, gcsafe, raises: [IOError, Defect].} =
-    var
-      cs = ChronosOutputStream(s)
-      retFuture = newFuture[void]("ChronosOutputStream.writePageAsync")
-      writeFut: Future[int]
-
-    proc continuation(udata: pointer) {.gcsafe.} =
-      if writeFut.error != nil:
-        retFuture.fail newException(IOError, writingErrMsg, writeFut.error)
-      elif writeFut.read != bufLen:
-        retFuture.fail newException(IOError, writeIncompleteErrMsg)
-      else:
-        retFuture.complete()
-
-    var writeFut = cs.transport.write(unsafeAddr page[0], page.len)
-    writeFut.addCallback(continuation, nil)
-
-    retFuture.cancelCallback = proc (udata: pointer) {.gcsafe.} =
-      writeFut.removeCallback(continuation, nil)
-
-    return retFuture
-  ,
-  flushSync: proc (s: OutputStream)
-                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
-    discard
-  ,
-  flushAsync: proc (s: OutputStream): Future[void]
                    {.nimcall, gcsafe, raises: [IOError, Defect].} =
-    result = newFuture[void]("ChronosOutputStream.flushAsync")
+    chronosWrites(ChronosOutputStream s, src, srcLen)
-    result.complete()
   ,
-  closeSync: ChronosInputStreamVTable.closeSync,
+  closeSync: proc (s: OutputStream)
-  closeAsyncProc: ChronosInputStreamVTable.closeAsyncProc
+                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    fsTranslateErrors closingErrMsg:
+      s.closeFut = close ChronosOutputStream(s).transport
+  ,
+  closeAsync: proc (s: OutputStream): Future[void]
+                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
+    chronosCloseWait ChronosOutputtream(s).transport
 )
 
 func chronosOutput*(s: StreamTransport,

faststreams/inputs.nim

@@ -22,10 +22,10 @@ type
 
   AsyncInputStream* {.borrow: `.`.} = distinct InputStream
 
-  ReadSyncProc* = proc (s: InputStream)
+  ReadSyncProc* = proc (s: InputStream, dst: pointer, dstLen: Natural): Natural
                        {.nimcall, gcsafe, raises: [IOError, Defect].}
 
-  ReadAsyncProc* = proc (s: InputStream): Future[void]
+  ReadAsyncProc* = proc (s: InputStream, dst: pointer, dstLen: Natural): Future[Natural]
                         {.nimcall, gcsafe, raises: [IOError, Defect].}
 
   CloseSyncProc* = proc (s: InputStream)
@@ -208,18 +208,21 @@ func totalUnconsumedBytes*(s: InputStream): Natural =
   ## buffers and that can be consumed with `read` or `advance`.
   result = s.span.len
   if s.buffers != nil:
-    result += s.buffers.totalBytesRead - s.spanEndPos
+    result += s.buffers.totalBufferredBytes
+    # result += s.buffers.totalBytesRead - s.spanEndPos
 
 template totalUnconsumedBytes*(s: AsyncInputStream): Natural =
   ## Alias for InputStream.totalUnconsumedBytes
   totalUnconsumedBytes InputStream(s)
 
 let FileInputVTable = InputStreamVTable(
-  readSync: proc (s: InputStream)
+  readSync: proc (s: InputStream, dst: pointer, dstLen: Natural): Natural
                  {.nimcall, gcsafe, raises: [IOError, Defect].} =
     let file = FileInputStream(s).file
-    s.buffers.writeToSpan(span):
+    implementSingleRead(s.buffers, dst, dstLen,
-      file.readBuffer(span.startAddr, span.len)
+                        {partialReadIsEof},
+                        readStartAddr, readLen):
+      file.readBuffer(readStartAddr, readLen)
   ,
   getLenSync: proc (s: InputStream): Natural
                    {.nimcall, gcsafe, raises: [IOError, Defect].} =
@@ -296,26 +299,24 @@ proc flipPage(s: InputStream) =
   s.span = s.buffers[0].span
   s.spanEndPos += s.span.len
 
-proc continueAfterRead(s: InputStream): bool =
+proc continueAfterRead(s: InputStream, bytesRead: Natural): bool =
   # Please note that this is extracted into a proc only to reduce the code
   # that ends up inlined into async procs by `bufferMoreDataImpl`.
   # The inlining itself is required to support the await-free operation of
   # the `readable` APIs.
-  let firstReadPage = s.buffers[0]
-
-  s.span = firstReadPage.span
-  let bytesRead = s.span.len
-  s.spanEndPos += bytesRead
 
   # 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):
-    s.disconnectInputDevice()
+    disconnectInputDevice(s)
 
-  # If we read some bytes anyway, we tell the user code that our buffers
+  if bytesRead > 0:
-  # contain some unconsumed data:
+    s.span = s.buffers.getReadableSpan()
-  bytesRead > 0
+    s.spanEndPos += s.span.len
+    return true
+  else:
+    return false
 
 template bufferMoreDataImpl(s, awaiter, readOp: untyped): bool =
   # This template is always called when the current page has been
@@ -339,8 +340,7 @@ template bufferMoreDataImpl(s, awaiter, readOp: untyped): bool =
       # the read was successful. In `continueAfterRead`, we examine if
       # EOF was reached, but please note that some data might have been
       # read anyway:
-      awaiter s.vtable.readOp(s)
+      continueAfterRead(s, awaiter s.vtable.readOp(s, nil, 0))
-      continueAfterRead(s)
 
 proc bufferMoreDataSync(s: InputStream): bool =
   # This proc exists only to avoid inlining of the code of
@@ -401,8 +401,14 @@ template readable*(sp: AsyncInputStream): bool =
   else:
     bufferMoreDataImpl(s, fsAsync, readAsync)
 
+func continueAfterReadN(s: InputStream,
+                        runwayBeforeRead, bytesRead: Natural) =
+  if runwayBeforeRead == 0 and bytesRead > 0:
+    s.span = s.buffers.getReadableSpan()
+    s.spanEndPos += s.span.len
+
 template readableNImpl(s, n, awaiter, readOp: untyped): bool =
-  let runway = s.totalUnconsumedBytes
+  let runway = totalUnconsumedBytes(s)
 
   if runway >= n:
     true
@@ -410,22 +416,23 @@ template readableNImpl(s, n, awaiter, readOp: untyped): bool =
     false
   else:
     var
-      bytesDeficit = n - runway
-      targetBytesRead = s.buffers.totalBytesRead + bytesDeficit
       res = false
+      bytesRead = Natural 0
+      bytesDeficit = n - runway
 
     while true:
-      awaiter s.vtable.readOp(s)
+      bytesRead += awaiter s.vtable.readOp(s, nil, bytesDeficit)
 
       if wasEofReached(s.buffers):
-        s.disconnectInputDevice()
+        disconnectInputDevice(s)
-        res = s.buffers.totalBytesRead >= targetBytesRead
+        res = bytesRead >= bytesDeficit
         break
 
-      if s.buffers.totalBytesRead >= targetBytesRead:
+      if bytesRead >= bytesDeficit:
         res = true
         break
 
+    continueAfterReadN(s, runway, bytesRead)
     res
 
 proc readable*(s: InputStream, n: int): bool =
@@ -502,14 +509,89 @@ proc read*(s: InputStream): byte =
 template read*(s: AsyncInputStream): byte =
   read InputStream(s)
 
-proc readIntoEx*(s: InputStream, target: var openarray[byte]): int =
+proc drainBuffersInto(s: InputStream, dstAddr: ptr byte, dstLen: Natural): Natural =
+  var
+    dst = dstAddr
+    remainingBytes = dstLen
+    runway = s.span.len
+
+  if runway >= remainingBytes:
+    copyMem(dst, s.span.startAddr, remainingBytes)
+    s.span.bumpPointer remainingBytes
+    return dstLen
+  elif runway > 0:
+    copyMem(dst, s.span.startAddr, runway)
+    dst = offset(dst, runway)
+    remainingBytes -= runway
+
+  if s.buffers != nil:
+    # Since we reached the end of the current page,
+    # we have to do the equivalent of `flipPage`:
+    if s.buffers.len > 0:
+      discard s.buffers.popFirst()
+
+    for page in consumePages(s.buffers):
+      let
+        pageStart = page.pageStartAddr
+        pageLen = page.endOffset - page.startOffset
+
+      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
+        # know how much our position in the stream has advanced:
+        let bytesDrainedFromBufers = dstLen - runway
+        s.span = page.span
+        s.spanEndPos += bytesDrainedFromBufers + s.span.len
+
+        # We return the length of the buffer, which means that is
+        # has been fully populated:
+        return dstLen
+      else:
+        copyMem(dst, pageStart, pageLen)
+        remainingBytes -= pageLen
+        dst = offset(dst, pageLen)
+
+  # We've completerly drained the current span and all the buffers,
+  # so we set the span to a pristine state that will trigger a new
+  # read on the next interaction with the stream.
+  s.span = default(PageSpan)
+
+  # We failed to populate the entire buffer
+  return dstLen - remainingBytes
+
+template readIntoExImpl(s: InputStream,
+                        dst: ptr byte, dstLen: Natural,
+                        awaiter, readOp: untyped): Natural =
+  var bytesRead = drainBuffersInto(s, dst, dstLen)
+
+  while bytesRead < dstLen:
+    let
+      bytesDeficit = dstLen - bytesRead
+      adjustedDst = offset(dst, bytesRead)
+
+    bytesRead += awaiter s.vtable.readOp(s, adjustedDst, bytesDeficit)
+
+    if wasEofReached(s.buffers):
+      disconnectInputDevice(s)
+      break
+
+  s.spanEndPos += bytesRead
+  bytesRead
+
+proc readIntoEx*(s: InputStream, dst: var openarray[byte]): int =
   ## Read data into the destination buffer.
   ##
   ## Returns the number of bytes that were successfully
   ## written to the buffer. The function will return a
   ## number smaller than the buffer length only if EOF
   ## was reached before the buffer was fully populated.
-  discard
+  let dstAddr = addr dst[0]
+  let dstLen = dst.len
+  readIntoExImpl(s, dstAddr, dstLen, noAwait, readSync)
 
 proc readInto*(s: InputStream, target: var openarray[byte]): bool =
   ## Read data into the destination buffer.
@@ -519,7 +601,14 @@ proc readInto*(s: InputStream, target: var openarray[byte]): bool =
   ## regarding the number of bytes read, see `readIntoEx`.
   s.readIntoEx(target) == target.len
 
-template readInto*(s: AsyncInputStream, target: var openarray[byte]): bool =
+template readIntoEx*(sp: AsyncInputStream, dst: var openarray[byte]): int =
+  let s = sp
+  # BEWARE! `openArrayToPair` here is needed to avoid
+  # double evaluation of the `dst` expression:
+  let (dstAddr, dstLen) = openArrayToPair(dst)
+  readIntoExImpl(s, dstAddr, dstLen, fsAwait, readAsync)
+
+template readInto*(sp: AsyncInputStream, dst: var openarray[byte]): bool =
   ## Asynchronously read data into the destination buffer.
   ##
   ## Returns `false` if EOF was reached before the buffer
@@ -529,16 +618,84 @@ template readInto*(s: AsyncInputStream, target: var openarray[byte]): bool =
   ## If there are enough bytes already buffered by the stream,
   ## the expression will complete immediately.
   ## Otherwise, it will await more bytes to become available.
-  discard
 
-proc checkReadAhead(s: InputStream, n: Natural): ptr byte =
+  let s = sp
-  # TODO: handle multi-page
+  # BEWARE! `openArrayToPair` here is needed to avoid
-  result = s.span.startAddr
+  # double evaluation of the `dst` expression:
-  doAssert s.span.len >= n
+  let (dstAddr, dstLen) = openArrayToPair(dst)
-  bumpPointer s.span, n
+  readIntoExImpl(s, dstAddr, dstLen, fsAwait, readAsync) == dstLen
 
-template read*(s: InputStream, n: Natural): auto =
+proc readOnce*(sp: AsyncInputStream): Future[Natural] =
-  makeOpenArray(checkReadAhead(s, n), n)
+  let s = InputStream(sp)
+  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:
+  proc alloca(n: int): ptr byte {.importc, header: "<alloca.h>".}
+
+template allocHeapMem(tmpSeq: var seq[byte], n, _: Natural): ptr byte =
+  tmpSeq.setLen(n)
+  addr tmpSeq[0]
+
+template allocStackMem(tmpSeq: var seq[byte], _, n: Natural): ptr byte =
+  alloca(n)
+
+template readNImpl(sp: InputStream,
+                   np: Natural,
+                   allocMem: untyped): openarray[byte] =
+  let
+    s = sp
+    n = np
+    runway = getBestRunway(s)
+
+  # Since Nim currently doesn't allow the `makeOpenArray` calls bellow
+  # to appear in different branches of an if statement, the code must
+  # be written in this branch-free linear fashion. The `dataCopy` seq
+  # may remain empty in the case where we use stack memory or return
+  # an `openarray` from the existing span.
+  var tmpSeq: seq[byte]
+  var startAddr: ptr byte
+
+  if n > runway:
+    startAddr = allocMem(tmpSeq, n, np)
+    doAssert drainBuffersInto(s, startAddr, n) == n
+  else:
+    startAddr = s.span.startAddr
+    bumpPointer s.span, n
+
+  makeOpenArray(startAddr, n)
+
+template read*(sp: InputStream, np: static Natural): openarray[byte] =
+  const n = np
+  when n < maxStackUsage:
+    readNImpl(sp, n, allocStackMem)
+  else:
+    readNImpl(sp, n, allocHeapMem)
+
+template read*(s: InputStream, n: Natural): openarray[byte] =
+  readNImpl(s, n, allocHeapMem)
+
+template read*(s: AsyncInputStream, n: Natural): openarray[byte] =
+  read InputStream(s), n
+
+proc lookAheadMatch*(s: InputStream, data: openarray[byte]): bool =
+  for i in 0 ..< data.len:
+    if s.peekAt(i) != data[i]:
+      return false
+
+  return true
+
+template lookAheadMatch*(s: AsyncInputStream, data: openarray[byte]): bool =
+  lookAheadMatch InputStream(s)
 
 proc next*(s: InputStream): Option[byte] =
   if readable(s):

faststreams/multisync.nim

@@ -7,15 +7,15 @@ macro fsMultiSync*(body: untyped) =
   # but taking async parameters and having the async pragma.
   var
     asyncProcBody = copy body
-    asyncProcParams = asyncBody[3]
+    asyncProcParams = asyncProcBody[3]
 
   asyncProcBody.addPragma(bindSym"async")
 
   # The return types becomes Future[T]
-  if asyncProcParams[0].kind == nnkEmpty
+  if asyncProcParams[0].kind == nnkEmpty:
-    asyncProcParams[0] = newBracketExpr(ident"Future", ident"void")
+    asyncProcParams[0] = newTree(nnkBracketExpr, ident"Future", ident"void")
   else:
-    asyncProcParams[0] = newBracketExpr(ident"Future", asyncProcParams[0])
+    asyncProcParams[0] = newTree(nnkBracketExpr, ident"Future", asyncProcParams[0])
 
   # We replace all stream inputs with their async counterparts
   for i in 1 ..< asyncProcParams.len:
@@ -26,7 +26,7 @@ macro fsMultiSync*(body: untyped) =
     elif sameType(typ, bindSym"OutputStream"):
       paramsDef[^2] = bindSym "AsyncOutputStream"
 
-  result = newStmtList(body, asyncBody)
+  result = newStmtList(body, asyncProcBody)
   if defined(debugSupportAsync):
     echo result.repr
 

faststreams/outputs.nim

@@ -35,10 +35,10 @@ type
 
   AsyncOutputStream* {.borrow: `.`.} = distinct OutputStream
 
-  WriteSyncProc* = proc (s: OutputStream, buf: pointer, bufLen: Natural)
+  WriteSyncProc* = proc (s: OutputStream, src: pointer, srcLen: Natural)
                         {.nimcall, gcsafe, raises: [IOError, Defect].}
 
-  WriteAsyncProc* = proc (s: OutputStream, buf: pointer, bufLen: Natural): Future[void]
+  WriteAsyncProc* = proc (s: OutputStream, src: pointer, srcLen: Natural): Future[void]
                          {.nimcall, gcsafe, raises: [IOError, Defect].}
 
   FlushSyncProc* = proc (s: OutputStream)
@@ -174,22 +174,13 @@ template ensureRunway*(s: AsyncOutputStream, neededRunway: Natural) =
   ensureRunway OutputStream(s, neededRunway)
 
 let FileOutputVTable = OutputStreamVTable(
-  writeSync: proc (s: OutputStream, buf: pointer, bufLen: Natural)
+  writeSync: proc (s: OutputStream, src: pointer, srcLen: Natural)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
     var file = FileOutputStream(s).file
 
-    template fail =
+    implementWrites(s.buffers, src, srcLen, "FILE",
-      raise newException(IOError, "Failed to write OutputStream page.")
+                    writeStartAddr, writeLen):
-
+      file.writeBuffer(writeStartAddr, writeLen)
-    if s.buffers != nil:
-      s.buffers.consumeAllPages(pageAddr, pageLen):
-        let written = file.writeBuffer(pageAddr, pageLen)
-        if written != pageLen: fail()
-
-    if bufLen > 0:
-      doAssert buf != nil
-      var written = file.writeBuffer(buf, bufLen)
-      if written != bufLen: fail()
   ,
   flushSync: proc (s: OutputStream)
                   {.nimcall, gcsafe, raises: [IOError, Defect].} =
@@ -605,11 +596,12 @@ template consumeOutputs*(sp: OutputStream, bytesVar, body: untyped) =
   ## Please note that calling `consumeOutputs` on an unbuffered stream
   ## or an unsafe memory stream is considered a Defect.
   ##
-  ## Before consuming the outputs, all outstanding delayed writes must be finalized.
+  ## Before consuming the outputs, all outstanding delayed writes must
+  ## be finalized.
   let s = sp
   doAssert s.extCursorsCount == 0 and s.buffers != nil
 
-  consumeAllPages(s.buffers, pageStartAddr, pageLen):
+  for pageStartAddr, pageLen in consumePageBuffers(s.buffers):
     template bytesVar: untyped =
       makeOpenArray(pageStartAddr, pageLen)
 
@@ -619,15 +611,20 @@ template consumeContiguousOutput*(sp: OutputStream, bytesVar, body: untyped) =
   ## Please note that calling `consumeContiguousOutput` on an unbuffered stream
   ## or an unsafe memory stream is considered a Defect.
   ##
-  ## Before consuming the output, all outstanding delayed writes must be finalized.
+  ## Before consuming the output, all outstanding delayed writes must
+  ## be finalized.
   ##
 
-  # TODO: This code is a bit too much to be inlined. Maybe this should be a proc
+  # TODO: This code is a bit too much to be inlined. Maybe this should be
-  # with a callback, but this will restrict the types of variables it can write to.
+  # a proc with a callback, but this will restrict the types of variables
-  # OTOH, perhaps only `consumeAllPages` is the offending part.
+  # it can write to. OTOH, perhaps only `consumePageBuffers` is the offending
+  # part.
   var
     s = sp
-    contigiousBytes: string # this may remain null
+    contigiousBytes: string
+      # this may remain null
+      # We are using a string, because `newStringOfCap` doesn't zero out
+      # the memory. TODO check if the same is true for `newSeqOfCap`.
     bytesPtr: ptr byte
     bytesLen: int
 
@@ -643,7 +640,7 @@ template consumeContiguousOutput*(sp: OutputStream, bytesVar, body: untyped) =
   else:
     contigiousBytes = newStringOfCap(s.pos)
 
-    consumeAllPages(s.buffers, pageStartAddr, pageLen):
+    for pageStartAddr, pageLen in consumePageBuffers(s.buffers):
       contigiousBytes.add makeOpenArray(cast[ptr char](pageStartAddr), pageLen)
 
     bytesPtr = addr contigiousBytes[0]
@@ -665,7 +662,7 @@ proc getOutput*(s: OutputStream, T: type string): string =
 
   if s.buffers.queue.len == 1:
     let page = s.buffers.queue[0]
-    if page.kind == stringPage and page.startOffset == 0:
+    if page.startOffset == 0:
       result.swap page.data[]
       result.setLen page.endOffset
       # We clear the buffers, so the stream will be in pristine state.

faststreams/textio.nim

@@ -1,16 +1,34 @@
 import
   stew/ptrops,
-  inputs, outputs, buffers
+  inputs, outputs, buffers, multisync
+
+template matchingIntType(T: type int64): type = uint64
+template matchingIntType(T: type int32): type = uint32
+template matchingIntType(T: type uint64): type = int64
+template matchingIntType(T: type uint32): type = int32
+
+# To reduce the produce code bloat, we will compile the integer
+# handling functions only for the native type of the platform.
+# Smaller int types will be automatically promoted to the native.
+# On a 32-bit platforms, we'll also compile support for 64-bit types.
+when sizeof(int) == sizeof(int64):
+  type
+    CompiledIntTypes = int64
+    PromotedIntTypes = int8|int16|int32
+    CompiledUIntTypes = uint64
+    PromotedUintTypes = uint8|uint16|uint32
+else:
+  type
+    CompiledIntTypes = int|int64
+    PromotedIntTypes = int8|int16
+    CompiledUIntTypes = uint|uint64
+    PromotedUintTypes = uint8|uint16
 
 # The following code implements writing numbers to a stream without going
 # through Nim's `$` operator which will allocate memory.
 # It's based on some speed comparisons of different methods presented here:
 # http://www.zverovich.net/2013/09/07/integer-to-string-conversion-in-cplusplus.html
 
-# TODO Maybe the `writeText` proc shouldn't be instantiated for every integer
-# type, but only for the largest "native" one. We can promote the rest with
-# a template.
-
 const
   digitsTable = block:
     var s = ""
@@ -21,7 +39,7 @@ const
 
   maxLen = ($BiggestInt.high).len + 4 # null terminator, sign
 
-proc writeText*(s: OutputStream, x: SomeUnsignedInt) =
+proc writeText*(s: OutputStream, x: CompiledUIntTypes) =
   var
     num: array[maxLen, char]
     pos = num.len
@@ -60,9 +78,8 @@ proc writeText*(s: OutputStream, x: SomeUnsignedInt) =
 
   write s, num.toOpenArray(pos, static(num.len - 1))
 
-proc writeText*(s: OutputStream, x: SomeSignedInt) =
+proc writeText*(s: OutputStream, x: CompiledIntTypes) =
-  # TODO: Determine this accurately
+  type MatchingUInt = matchingIntType typeof(x)
-  type MatchingUInt = BiggestUInt
 
   if x < 0:
     s.write '-'
@@ -90,3 +107,89 @@ proc writeHex*(s: OutputStream, bytes: openarray[byte]) =
 proc writeHex*(s: OutputStream, chars: openarray[char]) =
   writeHex s, charsToBytes(chars)
 
+const
+  NewLines* = {'\r', '\n'}
+  Digits* = {'0'..'9'}
+
+proc readLine*(s: InputStream, keepEol = false): TaintedString =
+  doAssert readableNow(s)
+
+  while s.readable:
+    let c = s.peek.char
+    if c in NewLines:
+      if keepEol:
+        result.add c
+        if c == '\r' and s.readable and s.peek.char == '\n':
+          result.add s.read.char
+      else:
+        advance s
+        if c == '\r' and s.readable and s.peek.char == '\n':
+          advance s
+      return
+
+    result.add s.read.char
+
+proc readUntil*(s: InputStream,
+                sep: openarray[char]): Option[TaintedString] =
+  doAssert readableNow(s)
+  var res = ""
+  while s.readable(sep.len):
+    if s.lookAheadMatch(charsToBytes(sep)):
+      return some(res)
+    res.add s.read.char
+
+template nextLine*(sp: InputStream, keepEol = false): Option[TaintedString] =
+  let s = sp
+  if s.readable:
+    some readLine(s, keepEol)
+  else:
+    none string
+
+iterator lines*(s: InputStream, keepEol = false): TaintedString =
+  while s.readable:
+    yield readLine(s, keepEol)
+
+proc readUnsignedInt*(s: InputStream, T: type[CompiledUIntTypes]): T =
+  doAssert s.readable and s.peek.char in Digits
+
+  template eatDigitAndPeek: char =
+    advance s
+    if not s.readable: return
+    s.peek.char
+
+  var c = s.peek.char
+  result = T(ord(c) - ord('0'))
+  c = eatDigitAndPeek()
+  while c.isDigit:
+    # TODO: How do we handle the possible overflow here?
+    result = result * 10 + T(ord(c) - ord('0'))
+    c = eatDigitAndPeek()
+
+template readUnsignedInt*(s: InputStream): uint =
+  readUnsignedInt(s, uint)
+
+proc readSignedInt*(s: InputStream, T: type[CompiledIntTypes]): Option[T] =
+  if s.readable:
+    let maybeSign = s.read.peek
+    if maybeSign in {'-', '+'}:
+      if not s.readable(2) or s.peekAt(1).char notin Digits:
+        return
+      else:
+        advance s
+    elif maybeSign notin Digits:
+      return
+
+    type UIntType = matchingIntType T
+    let uintVal = readUnsignedInt(s, UIntType)
+
+    if maybeSign == '-':
+      if uintVal > UIntType(max(T)) + 1:
+        return # Overflow. We've consumed part of the stream though.
+               # TODO: Should we rewind it to a previous state?
+      return some cast[T](UIntType(0) - uintVal)
+    else:
+      if uintVal > UIntType(max(T)):
+        return # Overflow. We've consumed part of the stream though.
+               # TODO: Should we rewind it to a previous state?
+      return some T(uintVal)
+

tests/files/ascii_table.txt

@@ -0,0 +1,34 @@
+|000 nul|001 soh|002 stx|003 etx|004 eot|005 enq|006 ack|007 bel|
+|010 bs |011 ht |012 nl |013 vt |014 np |015 cr |016 so |017 si |
+|020 dle|021 dc1|022 dc2|023 dc3|024 dc4|025 nak|026 syn|027 etb|
+|030 can|031 em |032 sub|033 esc|034 fs |035 gs |036 rs |037 us |
+|040 sp |041  ! |042  " |043  # |044  $ |045  % |046  & |047  ' |
+|050  ( |051  ) |052  * |053  + |054  , |055  - |056  . |057  / |
+|060  0 |061  1 |062  2 |063  3 |064  4 |065  5 |066  6 |067  7 |
+|070  8 |071  9 |072  : |073  ; |074  < |075  = |076  > |077  ? |
+|100  @ |101  A |102  B |103  C |104  D |105  E |106  F |107  G |
+|110  H |111  I |112  J |113  K |114  L |115  M |116  N |117  O |
+|120  P |121  Q |122  R |123  S |124  T |125  U |126  V |127  W |
+|130  X |131  Y |132  Z |133  [ |134  \ |135  ] |136  ^ |137  _ |
+|140  ` |141  a |142  b |143  c |144  d |145  e |146  f |147  g |
+|150  h |151  i |152  j |153  k |154  l |155  m |156  n |157  o |
+|160  p |161  q |162  r |163  s |164  t |165  u |166  v |167  w |
+|170  x |171  y |172  z |173  { |174  | |175  } |176  ~ |177 del|
+
+| 00 nul| 01 soh| 02 stx| 03 etx| 04 eot| 05 enq| 06 ack| 07 bel|
+| 08 bs | 09 ht | 0a nl | 0b vt | 0c np | 0d cr | 0e so | 0f si |
+| 10 dle| 11 dc1| 12 dc2| 13 dc3| 14 dc4| 15 nak| 16 syn| 17 etb|
+| 18 can| 19 em | 1a sub| 1b esc| 1c fs | 1d gs | 1e rs | 1f us |
+| 20 sp | 21  ! | 22  " | 23  # | 24  $ | 25  % | 26  & | 27  ' |
+| 28  ( | 29  ) | 2a  * | 2b  + | 2c  , | 2d  - | 2e  . | 2f  / |
+| 30  0 | 31  1 | 32  2 | 33  3 | 34  4 | 35  5 | 36  6 | 37  7 |
+| 38  8 | 39  9 | 3a  : | 3b  ; | 3c  < | 3d  = | 3e  > | 3f  ? |
+| 40  @ | 41  A | 42  B | 43  C | 44  D | 45  E | 46  F | 47  G |
+| 48  H | 49  I | 4a  J | 4b  K | 4c  L | 4d  M | 4e  N | 4f  O |
+| 50  P | 51  Q | 52  R | 53  S | 54  T | 55  U | 56  V | 57  W |
+| 58  X | 59  Y | 5a  Z | 5b  [ | 5c  \ | 5d  ] | 5e  ^ | 5f  _ |
+| 60  ` | 61  a | 62  b | 63  c | 64  d | 65  e | 66  f | 67  g |
+| 68  h | 69  i | 6a  j | 6b  k | 6c  l | 6d  m | 6e  n | 6f  o |
+| 70  p | 71  q | 72  r | 73  s | 74  t | 75  u | 76  v | 77  w |
+| 78  x | 79  y | 7a  z | 7b  { | 7c  | | 7d  } | 7e  ~ | 7f del|
+

tests/test_inputs.nim

@@ -1,23 +1,121 @@
 {.used.}
 
 import
-  os, unittest, strutils, stew/ranges/ptr_arith,
+  os, unittest, strutils, random,
-  ../faststreams
+  stew/ranges/ptr_arith, testutils,
+  ../faststreams, ../faststreams/textio
 
-suite "input stream":
+setCurrentDir getAppDir()
-  test "empty input":
+
+proc bytes(s: string): seq[byte] =
+  result = newSeqOfCap[byte](s.len)
+  for c in s: result.add byte(c)
+
+proc str(bytes: openarray[byte]): string =
+  result = newStringOfCap(bytes.len)
+  for b in bytes:
+    result.add b.char
+
+proc countLines(s: InputStream): Natural =
+  for s in lines(s):
+    inc result
+
+const
+  asciiTableFile = "files" / "ascii_table.txt"
+  asciiTableContents = slurp(asciiTableFile)
+
+procSuite "input stream":
+  template emptyInputTests(suiteName, setupCode: untyped) =
+    suite suiteName & " empty inputs":
+      setup setupCode
+
+      test "input is not readable with read":
+        check not input.readable
+        expect Defect:
+          echo "This read should not complete: ", input.read
+
+      test "input is not readable with read(n)":
+        check not input.readable(10)
+        expect Defect:
+          echo "This read should not complete: ", input.read(10)
+
+      test "next returns none":
+        check input.next.isNone
+
+  emptyInputTests "unsafeMemoryInput":
     var str = ""
-    var i = unsafeMemoryInput(str)
+    var input = unsafeMemoryInput(str)
 
-    check:
+  emptyInputTests "fileInput":
-      i.readable == false
+    var input = memFileInput("files" / "empty_file")
-      i.next.isNone
 
-    expect Defect:
+  emptyInputTests "memFileInput":
-      echo i.read
+    var input = fileInput("files" / "empty_file")
+
+  template asciiTableFileTest(name: string, body: untyped) =
+    test name & " of ascii table with regular pageSize":
+      var input {.inject.} = fileInput(asciiTableFile)
+      try:
+        body
+      finally:
+        close input
+
+    test name & " of ascii table with pageSize = 10":
+      var input {.inject.} = fileInput(asciiTableFile, pageSize = 10)
+      try:
+        body
+      finally:
+        close input
+
+    test name & " of ascii table with pageSize = 1":
+      var input {.inject.} = fileInput(asciiTableFile, pageSize = 1)
+      try:
+        body
+      finally:
+        close input
+
+  asciiTableFileTest "count lines":
+    check countLines(input) == 34
+
+  asciiTableFileTest "mixed read types":
+    randomize(10000)
+
+    var fileContents = ""
+
+    while true:
+      let r = rand(100)
+      if r < 20:
+        let readSize = 1 + rand(10)
+
+        var buf = newSeq[byte](readSize)
+        let bytesRead = input.readIntoEx(buf)
+        fileContents.add buf.toOpenArray(0, bytesRead - 1).str
+
+        if bytesRead < buf.len:
+          break
+
+      elif r < 50:
+        let readSize = 6 + rand(10)
+
+        if input.readable(readSize):
+          fileContents.add input.read(readSize).str
+        else:
+          while input.readable:
+            fileContents.add input.read.char
+          break
+
+      else:
+        if input.readable:
+          fileContents.add input.read.char
+
+      # You can uncomment this to get earlier failure in the test:
+      when false:
+        require fileContents == asciiTableContents[0 ..< fileContents.len]
+
+    check fileContents == asciiTableContents
 
   test "missing file input":
-    const fileName = "there-is-no-such-faststreams-file-1"
+    const fileName = "there-is-no-such-faststreams-file"
 
     check not fileExists(fileName)
     expect CatchableError: discard fileInput(fileName)

tests/test_outputs.nim

@@ -166,11 +166,11 @@ suite "output stream":
     fileStream.write delayedWriteContent
     streamWritingToExistingBuffer.write delayedWriteContent
 
-    var totalBytesWritten = 0
+    var bytesWritten = 0
     for i, count in [2]: # 12, 342, 2121, 23, 1, 34012, 932]:
       output repeat(byte(i), count)
-      totalBytesWritten += count
+      bytesWritten += count
-      check memStream.pos - cursorStart == totalBytesWritten
+      check memStream.pos - cursorStart == bytesWritten
 
     cursor.finalWrite delayedWriteContent