status-im/nim-chronos
2
0
Fork 0
mirror of https://github.com/status-im/nim-chronos.git synced 2025-02-21 15:38:20 +00:00
Code Issues Projects Releases Wiki Activity

Remove timeout handling for streams.

Browse Source
This commit is contained in:
cheatfate 2019-06-06 14:00:47 +03:00
parent 2ba6e5a412
commit 77b4ff0534
No known key found for this signature in database
GPG Key ID: 46ADD633A7201F95
2 changed files with 16 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
chronos/streams/asyncstream.nim
View File

@ -91,7 +91,6 @@ type
par*: ref Exception par*: ref Exception
AsyncStreamWriteError* = object of AsyncStreamError AsyncStreamWriteError* = object of AsyncStreamError
par*: ref Exception par*: ref Exception
AsyncStreamTimeoutError* = object of AsyncStreamError
proc init*(t: typedesc[AsyncBuffer], size: int): AsyncBuffer = proc init*(t: typedesc[AsyncBuffer], size: int): AsyncBuffer =
result.buffer = newSeq[byte](size) result.buffer = newSeq[byte](size)

92
chronos/streams/chunkstream.nim
View File

@ -13,50 +13,18 @@ export asyncstream, stream, timer, common
const const
ChunkBufferSize = 4096 ChunkBufferSize = 4096
ChunkHeaderSize = 8 ChunkHeaderSize = 8
ChunkDefaultTimeout = 10.seconds
CRLF = @[byte(0x0D), byte(0x0A)] CRLF = @[byte(0x0D), byte(0x0A)]
type type
ChunkedStreamReader* = ref object of AsyncStreamReader ChunkedStreamReader* = ref object of AsyncStreamReader
timeout: Duration
ChunkedStreamWriter* = ref object of AsyncStreamWriter ChunkedStreamWriter* = ref object of AsyncStreamWriter
ChunkedStreamError* = object of CatchableError ChunkedStreamError* = object of CatchableError
ChunkedStreamTimeoutError* = object of ChunkedStreamError
ChunkedStreamProtocolError* = object of ChunkedStreamError ChunkedStreamProtocolError* = object of ChunkedStreamError
proc newTimeoutError(): ref Exception {.inline.} =
newException(ChunkedStreamTimeoutError, "Timeout exceeded!")
proc newProtocolError(): ref Exception {.inline.} = proc newProtocolError(): ref Exception {.inline.} =
newException(ChunkedStreamProtocolError, "Protocol error!") newException(ChunkedStreamProtocolError, "Protocol error!")
proc oneOf*[A, B, C](fut1: Future[A], fut2: Future[B],
fut3: Future[C]): Future[void] =
## Wait for completion of one future from list [`fut1`, `fut2`, `fut3`],
## resulting future do not care about futures error, so you need to check
## `fut1`, `fut2`, `fut3` for error manually.
var retFuture = newFuture[void]("chunked.oneOf()")
proc cb(data: pointer) {.gcsafe.} =
var index: int
if not retFuture.finished:
var fut = cast[FutureBase](data)
if cast[pointer](fut1) == data:
fut2.removeCallback(cb)
fut3.removeCallback(cb)
elif cast[pointer](fut2) == data:
fut1.removeCallback(cb)
fut3.removeCallback(cb)
elif cast[pointer](fut3) == data:
fut2.removeCallback(cb)
fut3.removeCallback(cb)
retFuture.complete()
fut1.callback = cb
fut2.callback = cb
fut3.callback = cb
return retFuture
proc oneOf*[A, B](fut1: Future[A], fut2: Future[B]): Future[void] = proc oneOf*[A, B](fut1: Future[A], fut2: Future[B]): Future[void] =
## Wait for completion of `fut1` or `fut2`, resulting future do not care about ## Wait for completion of `fut1` or `fut2`, resulting future do not care about
## error, so you need to check `fut1` and `fut2` for error. ## error, so you need to check `fut1` and `fut2` for error.
@ -121,25 +89,14 @@ proc chunkedReadLoop(stream: AsyncStreamReader) {.async.} =
var rstream = cast[ChunkedStreamReader](stream) var rstream = cast[ChunkedStreamReader](stream)
var exitFut = rstream.exevent.wait() var exitFut = rstream.exevent.wait()
var buffer = newSeq[byte](1024) var buffer = newSeq[byte](1024)
var timerFut: Future[void]
if rstream.timeout == InfiniteDuration:
timerFut = newFuture[void]()
while true: while true:
if rstream.timeout != InfiniteDuration:
timerFut = sleepAsync(rstream.timeout)
# Reading chunk size # Reading chunk size
var ruFut1 = rstream.rsource.readUntil(addr buffer[0], 1024, CRLF) var ruFut1 = rstream.rsource.readUntil(addr buffer[0], 1024, CRLF)
await oneOf(ruFut1, timerFut, exitFut) await oneOf(ruFut1, exitFut)
if exitFut.finished or timerFut.finished: if exitFut.finished:
if timerFut.finished: rstream.state = AsyncStreamState.Stopped
rstream.error = newTimeoutError()
rstream.state = AsyncStreamState.Error
else:
rstream.state = AsyncStreamState.Stopped
break break
if ruFut1.failed: if ruFut1.failed:
@ -159,14 +116,10 @@ proc chunkedReadLoop(stream: AsyncStreamReader) {.async.} =
let toRead = min(int(chunksize), rstream.buffer.bufferLen()) let toRead = min(int(chunksize), rstream.buffer.bufferLen())
var reFut2 = rstream.rsource.readExactly(rstream.buffer.getBuffer(), var reFut2 = rstream.rsource.readExactly(rstream.buffer.getBuffer(),
toRead) toRead)
await oneOf(reFut2, timerFut, exitFut) await oneOf(reFut2, exitFut)
if exitFut.finished or timerFut.finished: if exitFut.finished:
if timerFut.finished: rstream.state = AsyncStreamState.Stopped
rstream.error = newTimeoutError()
rstream.state = AsyncStreamState.Error
else:
rstream.state = AsyncStreamState.Stopped
break break
if reFut2.failed: if reFut2.failed:
@ -187,13 +140,9 @@ proc chunkedReadLoop(stream: AsyncStreamReader) {.async.} =
# Reading chunk trailing CRLF # Reading chunk trailing CRLF
var reFut3 = rstream.rsource.readExactly(addr buffer[0], 2) var reFut3 = rstream.rsource.readExactly(addr buffer[0], 2)
await oneOf(reFut3, timerFut, exitFut) await oneOf(reFut3, exitFut)
if exitFut.finished or timerFut.finished: if exitFut.finished:
if timerFut.finished: rstream.state = AsyncStreamState.Stopped
rstream.error = newTimeoutError()
rstream.state = AsyncStreamState.Error
else:
rstream.state = AsyncStreamState.Stopped
break break
if reFut3.failed: if reFut3.failed:
@ -208,14 +157,10 @@ proc chunkedReadLoop(stream: AsyncStreamReader) {.async.} =
else: else:
# Reading trailing line for last chunk # Reading trailing line for last chunk
var ruFut4 = rstream.rsource.readUntil(addr buffer[0], len(buffer), CRLF) var ruFut4 = rstream.rsource.readUntil(addr buffer[0], len(buffer), CRLF)
await oneOf(ruFut4, timerFut, exitFut) await oneOf(ruFut4, exitFut)
if exitFut.finished or timerFut.finished: if exitFut.finished:
if timerFut.finished: rstream.state = AsyncStreamState.Stopped
rstream.error = newTimeoutError()
rstream.state = AsyncStreamState.Error
else:
rstream.state = AsyncStreamState.Stopped
break break
if ruFut4.failed: if ruFut4.failed:
@ -336,30 +281,25 @@ proc chunkedWriteLoop(stream: AsyncStreamWriter) {.async.} =
untrackAsyncStreamWriter(wstream) untrackAsyncStreamWriter(wstream)
proc init*[T](child: ChunkedStreamReader, rsource: AsyncStreamReader, proc init*[T](child: ChunkedStreamReader, rsource: AsyncStreamReader,
bufferSize = ChunkBufferSize, timeout = ChunkDefaultTimeout, bufferSize = ChunkBufferSize, udata: ref T) =
udata: ref T) =
child.timeout = timeout
init(cast[AsyncStreamReader](child), rsource, chunkedReadLoop, bufferSize, init(cast[AsyncStreamReader](child), rsource, chunkedReadLoop, bufferSize,
udata) udata)
proc init*(child: ChunkedStreamReader, rsource: AsyncStreamReader, proc init*(child: ChunkedStreamReader, rsource: AsyncStreamReader,
bufferSize = ChunkBufferSize, timeout = ChunkDefaultTimeout) = bufferSize = ChunkBufferSize) =
child.timeout = timeout
init(cast[AsyncStreamReader](child), rsource, chunkedReadLoop, bufferSize) init(cast[AsyncStreamReader](child), rsource, chunkedReadLoop, bufferSize)
proc newChunkedStreamReader*[T](rsource: AsyncStreamReader, proc newChunkedStreamReader*[T](rsource: AsyncStreamReader,
bufferSize = AsyncStreamDefaultBufferSize, bufferSize = AsyncStreamDefaultBufferSize,
timeout = ChunkDefaultTimeout,
udata: ref T): ChunkedStreamReader = udata: ref T): ChunkedStreamReader =
result = new ChunkedStreamReader result = new ChunkedStreamReader
result.init(rsource, bufferSize, timeout, udata) result.init(rsource, bufferSize, udata)
proc newChunkedStreamReader*(rsource: AsyncStreamReader, proc newChunkedStreamReader*(rsource: AsyncStreamReader,
bufferSize = AsyncStreamDefaultBufferSize, bufferSize = AsyncStreamDefaultBufferSize,
timeout = ChunkDefaultTimeout,
): ChunkedStreamReader = ): ChunkedStreamReader =
result = new ChunkedStreamReader result = new ChunkedStreamReader
result.init(rsource, bufferSize, timeout) result.init(rsource, bufferSize)
proc init*[T](child: ChunkedStreamWriter, wsource: AsyncStreamWriter, proc init*[T](child: ChunkedStreamWriter, wsource: AsyncStreamWriter,
queueSize = AsyncStreamDefaultQueueSize, udata: ref T) = queueSize = AsyncStreamDefaultQueueSize, udata: ref T) =