From 73c44c6e9b55f7fb2f5b902c78086f46c1b118dd Mon Sep 17 00:00:00 2001 From: Dominik Picheta Date: Wed, 12 Jun 2019 16:07:05 +0100 Subject: [bugfix] Fixes async IO operations stalling even after socket is closed. (#11232) --- lib/pure/asyncdispatch.nim | 117 +++++++++++++++++++++++++++----------------- lib/pure/asyncfutures.nim | 12 +++-- lib/pure/includes/oserr.nim | 15 ++++-- 3 files changed, 92 insertions(+), 52 deletions(-) (limited to 'lib/pure') diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim index ea35a444d..382d9d44a 100644 --- a/lib/pure/asyncdispatch.nim +++ b/lib/pure/asyncdispatch.nim @@ -568,7 +568,7 @@ when defined(windows) or defined(nimdoc): if flags.isDisconnectionError(errcode): retFuture.complete() else: - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) ) let ret = WSASend(socket.SocketHandle, addr dataBuf, 1, addr bytesReceived, @@ -1134,11 +1134,6 @@ else: var data = newAsyncData() p.selector.registerHandle(fd.SocketHandle, {}, data) - proc closeSocket*(sock: AsyncFD) = - let disp = getGlobalDispatcher() - disp.selector.unregister(sock.SocketHandle) - sock.SocketHandle.close() - proc unregister*(fd: AsyncFD) = getGlobalDispatcher().selector.unregister(fd.SocketHandle) @@ -1174,7 +1169,9 @@ else: let p = getGlobalDispatcher() not p.selector.isEmpty() or p.timers.len != 0 or p.callbacks.len != 0 - template processBasicCallbacks(ident, rwlist: untyped) = + proc processBasicCallbacks( + fd: AsyncFD, event: Event + ): tuple[readCbListCount, writeCbListCount: int] = # Process pending descriptor and AsyncEvent callbacks. # # Invoke every callback stored in `rwlist`, until one @@ -1187,32 +1184,46 @@ else: # or it can be possible to fall into endless cycle. var curList: seq[Callback] - withData(p.selector, ident, adata) do: - shallowCopy(curList, adata.rwlist) - adata.rwlist = newSeqOfCap[Callback](InitCallbackListSize) + let selector = getGlobalDispatcher().selector + withData(selector, fd.int, fdData): + case event + of Event.Read: + shallowCopy(curList, fdData.readList) + fdData.readList = newSeqOfCap[Callback](InitCallbackListSize) + of Event.Write: + shallowCopy(curList, fdData.writeList) + fdData.writeList = newSeqOfCap[Callback](InitCallbackListSize) + else: + assert false, "Cannot process callbacks for " & $event let newLength = max(len(curList), InitCallbackListSize) var newList = newSeqOfCap[Callback](newLength) for cb in curList: - if len(newList) > 0: - # A callback has already returned with EAGAIN, don't call any others - # until next `poll`. + if not cb(fd): + # Callback wants to be called again. newList.add(cb) + # This callback has returned with EAGAIN, so we don't need to + # call any other callbacks as they are all waiting for the same event + # on the same fd. + break + + withData(selector, fd.int, fdData) do: + # Descriptor is still present in the queue. + case event + of Event.Read: + fdData.readList = newList & fdData.readList + of Event.Write: + fdData.writeList = newList & fdData.writeList else: - if not cb(fd.AsyncFD): - # Callback wants to be called again. - newList.add(cb) + assert false, "Cannot process callbacks for " & $event - withData(p.selector, ident, adata) do: - # descriptor still present in queue. - adata.rwlist = newList & adata.rwlist - rLength = len(adata.readList) - wLength = len(adata.writeList) + result.readCbListCount = len(fdData.readList) + result.writeCbListCount = len(fdData.writeList) do: - # descriptor was unregistered in callback via `unregister()`. - rLength = -1 - wLength = -1 + # Descriptor was unregistered in callback via `unregister()`. + result.readCbListCount = -1 + result.writeCbListCount = -1 template processCustomCallbacks(ident: untyped) = # Process pending custom event callbacks. Custom events are @@ -1221,7 +1232,7 @@ else: # so there is no need to iterate over list. var curList: seq[Callback] - withData(p.selector, ident, adata) do: + withData(p.selector, ident.int, adata) do: shallowCopy(curList, adata.readList) adata.readList = newSeqOfCap[Callback](InitCallbackListSize) @@ -1232,16 +1243,33 @@ else: if not cb(fd.AsyncFD): newList.add(cb) - withData(p.selector, ident, adata) do: + withData(p.selector, ident.int, adata) do: # descriptor still present in queue. adata.readList = newList & adata.readList if len(adata.readList) == 0: # if no callbacks registered with descriptor, unregister it. - p.selector.unregister(fd) + p.selector.unregister(fd.int) do: # descriptor was unregistered in callback via `unregister()`. discard + proc closeSocket*(sock: AsyncFD) = + let selector = getGlobalDispatcher().selector + if sock.SocketHandle notin selector: + raise newException(ValueError, "File descriptor not registered.") + + let data = selector.getData(sock.SocketHandle) + sock.unregister() + sock.SocketHandle.close() + # We need to unblock the read and write callbacks which could still be + # waiting for the socket to become readable and/or writeable. + for cb in data.readList & data.writeList: + if not cb(sock): + raise newException( + ValueError, "Expecting async operations to stop when fd has closed." + ) + + proc runOnce(timeout = 500): bool = let p = getGlobalDispatcher() when ioselSupportedPlatform: @@ -1257,41 +1285,42 @@ else: let nextTimer = processTimers(p, result) var count = p.selector.selectInto(adjustTimeout(timeout, nextTimer), keys) for i in 0.. 0: incl(newEvents, Event.Read) - if wLength > 0: incl(newEvents, Event.Write) - p.selector.updateHandle(SocketHandle(fd), newEvents) + if readCbListCount > 0: incl(newEvents, Event.Read) + if writeCbListCount > 0: incl(newEvents, Event.Write) + p.selector.updateHandle(SocketHandle(fd), newEvents) # Timer processing. discard processTimers(p, result) @@ -1370,7 +1399,7 @@ else: if flags.isDisconnectionError(lastError): retFuture.complete() else: - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) else: result = false # We still want this callback to be called. else: diff --git a/lib/pure/asyncfutures.nim b/lib/pure/asyncfutures.nim index 9af72f8b3..e86a34d81 100644 --- a/lib/pure/asyncfutures.nim +++ b/lib/pure/asyncfutures.nim @@ -393,11 +393,13 @@ proc asyncCheck*[T](future: Future[T]) = ## This should be used instead of ``discard`` to discard void futures, ## or use ``waitFor`` if you need to wait for the future's completion. assert(not future.isNil, "Future is nil") - future.callback = - proc () = - if future.failed: - injectStacktrace(future) - raise future.error + # TODO: We can likely look at the stack trace here and inject the location + # where the `asyncCheck` was called to give a better error stack message. + proc asyncCheckCallback() = + if future.failed: + injectStacktrace(future) + raise future.error + future.callback = asyncCheckCallback proc `and`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] = ## Returns a future which will complete once both ``fut1`` and ``fut2`` diff --git a/lib/pure/includes/oserr.nim b/lib/pure/includes/oserr.nim index 68ce5d95f..34d8d0085 100644 --- a/lib/pure/includes/oserr.nim +++ b/lib/pure/includes/oserr.nim @@ -57,8 +57,10 @@ proc osErrorMsg*(errorCode: OSErrorCode): string = if errorCode != OSErrorCode(0'i32): result = $c_strerror(errorCode.int32) -proc raiseOSError*(errorCode: OSErrorCode; additionalInfo = "") {.noinline.} = - ## Raises an `OSError exception `_. +proc newOSError*( + errorCode: OSErrorCode, additionalInfo = "" +): owned(ref OSError) {.noinline.} = + ## Creates a new `OSError exception `_. ## ## The ``errorCode`` will determine the ## message, `osErrorMsg proc <#osErrorMsg,OSErrorCode>`_ will be used @@ -82,7 +84,14 @@ proc raiseOSError*(errorCode: OSErrorCode; additionalInfo = "") {.noinline.} = e.msg.addQuoted additionalInfo if e.msg == "": e.msg = "unknown OS error" - raise e + return e + +proc raiseOSError*(errorCode: OSErrorCode, additionalInfo = "") {.noinline.} = + ## Raises an `OSError exception `_. + ## + ## Read the description of the `newOSError proc <#newOSError,OSErrorCode,string>`_ to learn + ## how the exception object is created. + raise newOSError(errorCode, additionalInfo) {.push stackTrace:off.} proc osLastError*(): OSErrorCode {.sideEffect.} = -- cgit 1.4.1-2-gfad0