From bf930b6f1b7fc1a381f6a97c2b1c45566bfb88c1 Mon Sep 17 00:00:00 2001 From: Zahary Karadjov Date: Fri, 1 May 2020 21:39:08 +0300 Subject: [PATCH] Implement readInto and read(n) --- faststreams.nimble | 1 + faststreams/async_backend.nim | 2 + faststreams/buffers.nim | 203 ++++++++++++++++----------- faststreams/chronos_adapters.nim | 85 +++++------- faststreams/inputs.nim | 229 ++++++++++++++++++++++++++----- faststreams/multisync.nim | 10 +- faststreams/outputs.nim | 45 +++--- faststreams/textio.nim | 121 ++++++++++++++-- tests/files/ascii_table.txt | 34 +++++ tests/files/empty_file | 0 tests/test_inputs.nim | 120 ++++++++++++++-- tests/test_outputs.nim | 6 +- 12 files changed, 636 insertions(+), 220 deletions(-) create mode 100644 tests/files/ascii_table.txt create mode 100644 tests/files/empty_file diff --git a/faststreams.nimble b/faststreams.nimble index 628719a..be3a4c5 100644 --- a/faststreams.nimble +++ b/faststreams.nimble @@ -9,6 +9,7 @@ skipDirs = @["tests"] requires "nim >= 1.2.0", "stew", + "testutils", "chronos" task test, "Run all tests": diff --git a/faststreams/async_backend.nim b/faststreams/async_backend.nim index 30de7b5..90b2185 100644 --- a/faststreams/async_backend.nim +++ b/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) diff --git a/faststreams/buffers.nim b/faststreams/buffers.nim index 9c91a8c..df07d25 100644 --- a/faststreams/buffers.nim +++ b/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 - of userPage, mallocPage: - bufferStart, bufferEnd: ptr byte - of stringPage: - data*: ref string + data*: ref string PageRef* = ref Page @@ -47,38 +38,30 @@ const defaultPageSize* = 4096 - (pageMetadataSize + nimAllocatorMetadataSize) maxStackUsage* = 16384 -func pageBaseAddr*(page: PageRef): ptr byte = - if page.kind == stringPage: - cast[ptr byte](addr page.data[][0]) - else: - page.bufferStart +proc openArrayToPair*(a: var openarray[byte]): (ptr byte, Natural) = + (addr a[0], Natural(a.len)) + +template pageBaseAddr*(page: PageRef): ptr byte = + 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) - else: - offset(page.bufferStart, page.startOffset) + offset(cast[ptr byte](addr page.data[][0]), page.startOffset) func pageEndAddr*(page: PageRef): ptr byte = - if page.kind == stringPage: - offset(cast[ptr byte](addr page.data[][0]), page.endOffset) - else: - offset(page.bufferStart, page.endOffset) + offset(cast[ptr byte](addr page.data[][0]), page.endOffset) + +func pageLen*(page: PageRef): Natural = + 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 = cast[ptr byte](addr page.data[][0]) - PageSpan(startAddr: offset(baseAddr, page.startOffset), - endAddr: offset(baseAddr, when writable: page.data[].len - else: page.endOffset)) - else: - PageSpan(startAddr: offset(page.bufferStart, page.startOffset), - endAddr: when writable: page.bufferEnd - else: offset(page.bufferStart, page.endOffset)) + let baseAddr = page.pageBaseAddr + PageSpan(startAddr: offset(baseAddr, page.startOffset), + endAddr: offset(baseAddr, when writable: page.data[].len + else: 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, - data: allocRef newString(pageSize), + result = PageRef(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, - pageAddrVar, pageLenVar, body: untyped) = - let buffers = buffersParam +iterator consumePages*(buffers: PageBuffers): PageRef = doAssert buffers != nil var recycledPage: PageRef - for page in buffers.queue: - let - pageAddrVar = page.pageStartAddr - pageLenVar = page.endOffset - page.startOffset - - if page.kind == stringPage and page.data[].len == buffers.pageSize: - recycledPage = page + while buffers.queue.len > 0: + var page = peekFirst(buffers.queue) # TODO: what if the body throws an exception? - # Should we do anything with the remaining pages? - body + # Should we do anything with the consumed page? + 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 + diff --git a/faststreams/chronos_adapters.nim b/faststreams/chronos_adapters.nim index 450bb76..870f1a0 100644 --- a/faststreams/chronos_adapters.nim +++ b/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, - buffer: ptr byte, bufSize: int) - {.raises: [Defect, IOError], async.} = +proc chronosReadOnce(s: ChronosInputStream, dst: pointer, dstLen: Natural): Future[Natural] + {.raises: [IOError, Defect], async.} = fsTranslateErrors readingErrMsg: - buffers.writeToSpan(span): - await t.readOnce(span.startAddr, span.len) + 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.} = + 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 - - fsTranslateErrors readingErrMsg: - buffers.writeToSpan(span): - waitFor cs.transport.readOnce(span.startAddr, span.len) + waitFor chronosReadOnce(cs, dst, dstLen) , - 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]) - {.nimcall, gcsafe, raises: [IOError, Defect].} = + writeSync: proc (s: OutputStream, src: pointer, srcLen: Natural) + {.nimcall, gcsafe, raises: [IOError, Defect].} = var cs = ChronosOutputStream(s) doAssert cs.allowWaitFor - let bytesWritten = fsTranslateErrors writingErrMsg: - waitFor cs.transport.write(unsafeAddr page[0], page.len) - if bytesWritten != page.len: - raise newException(IOError, writeIncompleteErrMsg) + waitFor chronosWrites(cs, src, srcLen) , - writePageAsync: proc (s: OutputStream, buf: pointer, bufLen: int): 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] + writeAsync: proc (s: OutputStream, src: pointer, srcLen: Natural): Future[void] {.nimcall, gcsafe, raises: [IOError, Defect].} = - result = newFuture[void]("ChronosOutputStream.flushAsync") - result.complete() + chronosWrites(ChronosOutputStream s, src, srcLen) , - closeSync: ChronosInputStreamVTable.closeSync, - closeAsyncProc: ChronosInputStreamVTable.closeAsyncProc + closeSync: proc (s: OutputStream) + {.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, diff --git a/faststreams/inputs.nim b/faststreams/inputs.nim index 9809b83..c6678e3 100644 --- a/faststreams/inputs.nim +++ b/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): - file.readBuffer(span.startAddr, span.len) + implementSingleRead(s.buffers, dst, dstLen, + {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 - # contain some unconsumed data: - bytesRead > 0 + if bytesRead > 0: + s.span = s.buffers.getReadableSpan() + 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) + continueAfterRead(s, awaiter s.vtable.readOp(s, nil, 0)) 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() - res = s.buffers.totalBytesRead >= targetBytesRead + disconnectInputDevice(s) + 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 = - # TODO: handle multi-page - result = s.span.startAddr - doAssert s.span.len >= n - bumpPointer s.span, n + 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) == dstLen -template read*(s: InputStream, n: Natural): auto = - makeOpenArray(checkReadAhead(s, n), n) +proc readOnce*(sp: AsyncInputStream): Future[Natural] = + 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: "".} +else: + proc alloca(n: int): ptr byte {.importc, header: "".} + +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): diff --git a/faststreams/multisync.nim b/faststreams/multisync.nim index 6bf1a01..6c470b6 100644 --- a/faststreams/multisync.nim +++ b/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 - asyncProcParams[0] = newBracketExpr(ident"Future", ident"void") + if asyncProcParams[0].kind == nnkEmpty: + 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 diff --git a/faststreams/outputs.nim b/faststreams/outputs.nim index 213ef1f..997a070 100644 --- a/faststreams/outputs.nim +++ b/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 = - raise newException(IOError, "Failed to write OutputStream page.") - - 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() + implementWrites(s.buffers, src, srcLen, "FILE", + writeStartAddr, writeLen): + file.writeBuffer(writeStartAddr, writeLen) , 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 - # with a callback, but this will restrict the types of variables it can write to. - # OTOH, perhaps only `consumeAllPages` is the offending part. + # TODO: This code is a bit too much to be inlined. Maybe this should be + # a proc with a callback, but this will restrict the types of variables + # 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. diff --git a/faststreams/textio.nim b/faststreams/textio.nim index 900e3cc..6e201fe 100644 --- a/faststreams/textio.nim +++ b/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) = - # TODO: Determine this accurately - type MatchingUInt = BiggestUInt +proc writeText*(s: OutputStream, x: CompiledIntTypes) = + type MatchingUInt = matchingIntType typeof(x) 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) + diff --git a/tests/files/ascii_table.txt b/tests/files/ascii_table.txt new file mode 100644 index 0000000..5bba9d8 --- /dev/null +++ b/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| + diff --git a/tests/files/empty_file b/tests/files/empty_file new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_inputs.nim b/tests/test_inputs.nim index ceb1110..01acf6f 100644 --- a/tests/test_inputs.nim +++ b/tests/test_inputs.nim @@ -1,23 +1,121 @@ {.used.} import - os, unittest, strutils, stew/ranges/ptr_arith, - ../faststreams + os, unittest, strutils, random, + stew/ranges/ptr_arith, testutils, + ../faststreams, ../faststreams/textio -suite "input stream": - test "empty input": +setCurrentDir getAppDir() + +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: - i.readable == false - i.next.isNone + emptyInputTests "fileInput": + var input = memFileInput("files" / "empty_file") - expect Defect: - echo i.read + emptyInputTests "memFileInput": + 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) diff --git a/tests/test_outputs.nim b/tests/test_outputs.nim index 8d842cf..2699ebf 100644 --- a/tests/test_outputs.nim +++ b/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 - check memStream.pos - cursorStart == totalBytesWritten + bytesWritten += count + check memStream.pos - cursorStart == bytesWritten cursor.finalWrite delayedWriteContent