diff --git a/README.md b/README.md index 3bd8026..9773f14 100644 --- a/README.md +++ b/README.md @@ -473,6 +473,10 @@ bytes before continuing. The rationale here is that introducing an interruption point at every `write` produces less optimal code, but if this is desired you can use the `stream.writeAndWait` API. +If you have existing algorithms that output data to an `openArray`, you can use +the `stream.getWritableBytes` API to continue using them without introducing any +intermediate buffers. + #### Delayed Writes Many protocols and formats employ fixed-size and variable-size length prefixes diff --git a/faststreams/outputs.nim b/faststreams/outputs.nim index 35c7162..cfe8bcf 100644 --- a/faststreams/outputs.nim +++ b/faststreams/outputs.nim @@ -136,7 +136,7 @@ template flushImpl(s: OutputStream, awaiter, writeOp, flushOp: untyped) = trackWrittenTo(s.buffers, s.span.startAddr) awaiter s.vtable.writeOp(s, nil, 0) s.span = getWritableSpan(s.buffers) - s.spanEndPos += s.span.len - runway + s.spanEndPos = s.spanEndPos - runway + s.span.len if s.vtable.flushOp != nil: awaiter s.vtable.flushOp(s) @@ -223,7 +223,7 @@ proc ensureRunway*(s: OutputStream, neededRunway: Natural) = # feeding to it. fsAssert s.buffers != nil, "Unsafe memory output of insufficient size" s.buffers.ensureRunway(s.span, neededRunway) - s.spanEndPos += (s.span.len - runway) + s.spanEndPos = s.spanEndPos - runway + s.span.len when fsAsyncSupport: template ensureRunway*(s: AsyncOutputStream, neededRunway: Natural) = @@ -472,7 +472,55 @@ proc delayVarSizeWrite*(s: OutputStream, maxSize: Natural): VarSizeWriteCursor = s.span = PageSpan(startAddr: cursorEndAddr, endAddr: nextPageSpan.endAddr) - s.spanEndPos += nextPageSize - runway + s.spanEndPos = s.spanEndPos - runway + nextPageSize + +proc getWritableBytesOnANewPage(s: OutputStream, spanSize: Natural): ptr byte = + fsAssert s.buffers != nil + + trackWrittenTo(s.buffers, s.span.startAddr) + + let + nextPageSize = nextAlignedSize(spanSize, s.buffers.pageSize) + nextPage = allocWritablePage(nextPageSize, spanSize) + + s.buffers.queue.addLast nextPage + + let retiredSpanRunway = s.span.len + s.span = nextPage.fullSpan + s.spanEndPos = s.spanEndPos - retiredSpanRunway + spanSize + s.span.startAddr + +template getWritableBytes*(sp: OutputStream, spanSizeParam: Natural): openArray[byte] = + ## Returns a contiguous range of memory that the caller is free to populate fully + ## or partially. The caller indicates how many bytes were written to the span by + ## calling `advance(numberOfBytes)` once or multiple times. Advancing the stream + ## past the allocated span size is considered a defect. The typical usage pattern + ## of this API looks as follows: + ## + ## stream.advance(myComponent.writeBlock(stream.getWritableBytes(maxBlockSize))) + ## + ## In the example, `writeBlock` would be a function returning the number of bytes + ## written to the span. + ## + ## While it's not illegal to issue other writing operations to the stream during + ## the `getWritetableSpan` -> `advance` sequence, doing this is not recommended + ## because it will result in overwriting the same range of bytes. + let + s = sp + spanSize = spanSizeParam + runway = s.span.len + startAddr = if spanSize <= runway: + s.span.startAddr + else: + getWritableBytesOnANewPage(s, spanSize) + + makeOpenArray(startAddr, spanSize) + +proc advance*(s: OutputStream, bytesWrittenToWritableSpan: Natural) = + ## Advance the stream write cursor. + ## Typically used after a previous call to `getWritableBytes`. + fsAssert bytesWrittenToWritableSpan <= s.span.len + s.span.startAddr = offset(s.span.startAddr, bytesWrittenToWritableSpan) proc finalize*(cursor: var WriteCursor) = fsAssert cursor.stream.extCursorsCount > 0 @@ -776,7 +824,7 @@ proc consumeOutputsImpl(s: OutputStream, consumer: OutputConsumingProc) = consumer(makeOpenArray(pageReadableStart, pageLen)) s.span = getWritableSpan(s.buffers) - s.spanEndPos += s.span.len - runway + s.spanEndPos = s.spanEndPos - runway + s.span.len template consumeOutputs*(s: OutputStream, bytesVar, body: untyped) = ## Please note that calling `consumeOutputs` on an unbuffered stream @@ -821,7 +869,7 @@ proc consumeContiguousOutputImpl(s: OutputStream, consumer: OutputConsumingProc) consumer(makeOpenArray(bytesPtr, bytesLen)) s.span = s.buffers.getWritableSpan() - s.spanEndPos += s.span.len - runway + s.spanEndPos = s.spanEndPos - runway + s.span.len template consumeContiguousOutput*(s: OutputStream, bytesVar, body: untyped) = ## Please note that calling `consumeContiguousOutput` on an unbuffered stream diff --git a/tests/test_outputs.nim b/tests/test_outputs.nim index 4ce3f0f..509ff0a 100644 --- a/tests/test_outputs.nim +++ b/tests/test_outputs.nim @@ -213,6 +213,11 @@ suite "output stream": checkOutputsMatch() +proc writeBlock(data: openArray[byte], output: openArray[byte]): int = + doAssert data.len <= output.len + copyMem(unsafeAddr output[0], unsafeAddr data[0], data.len) + data.len + suite "randomized tests": type WriteTypes = enum @@ -357,7 +362,14 @@ suite "randomized tests": for i in 0 .. 3000: let bytes = randomBytes(rand(9999) + 1) referenceBytes.add bytes - output.write bytes + + var openArraySize = rand(12000) + if openArraySize >= bytes.len: + # Make sure that sometimes `writeBlock` populates the entire span + if i < 100: openArraySize = bytes.len + output.advance writeBlock(bytes, output.getWritableBytes(openArraySize)) + else: + output.write bytes close output