diff options
Diffstat (limited to 'lib/pure/asyncdispatch.nim')
| -rw-r--r-- | lib/pure/asyncdispatch.nim | 254 |
1 files changed, 179 insertions, 75 deletions
diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim index 55a20270d..126db7a7f 100644 --- a/lib/pure/asyncdispatch.nim +++ b/lib/pure/asyncdispatch.nim @@ -41,13 +41,13 @@ ## requested amount of data is read **or** an exception occurs. ## ## Code to read some data from a socket may look something like this: -## -## .. code-block::nim -## var future = socket.recv(100) -## future.addCallback( -## proc () = -## echo(future.read) -## ) +## ```Nim +## var future = socket.recv(100) +## future.addCallback( +## proc () = +## echo(future.read) +## ) +## ``` ## ## All asynchronous functions returning a `Future` will not block. They ## will not however return immediately. An asynchronous function will have @@ -61,8 +61,8 @@ ## callback on this future which will be called once the future completes. ## All the callback does is write the data stored in the future to `stdout`. ## The `read` function is used for this and it checks whether the future -## completes with an error for you (if it did it will simply raise the -## error), if there is no error however it returns the value of the future. +## completes with an error for you (if it did, it will simply raise the +## error), if there is no error, however, it returns the value of the future. ## ## Asynchronous procedures ## ======================= @@ -98,40 +98,71 @@ ## `await`. The following section shows different ways that you can handle ## exceptions in async procs. ## +## .. caution:: +## Procedures marked {.async.} do not support mutable parameters such +## as `var int`. References such as `ref int` should be used instead. +## ## Handling Exceptions ## ------------------- ## -## The most reliable way to handle exceptions is to use `yield` on a future -## then check the future's `failed` property. For example: +## You can handle exceptions in the same way as in ordinary Nim code; +## by using the try statement: ## -## .. code-block:: Nim -## var future = sock.recv(100) -## yield future -## if future.failed: -## # Handle exception +## ```Nim +## try: +## let data = await sock.recv(100) +## echo("Received ", data) +## except: +## # Handle exception +## ``` ## -## The `async` procedures also offer limited support for the try statement. -## -## .. code-block:: Nim -## try: -## let data = await sock.recv(100) -## echo("Received ", data) -## except: -## # Handle exception +## An alternative approach to handling exceptions is to use `yield` on a future +## then check the future's `failed` property. For example: ## -## Unfortunately the semantics of the try statement may not always be correct, -## and occasionally the compilation may fail altogether. -## As such it is better to use the former style when possible. +## ```Nim +## var future = sock.recv(100) +## yield future +## if future.failed: +## # Handle exception +## ``` ## ## ## Discarding futures ## ================== ## -## Futures should **never** be discarded. This is because they may contain -## errors. If you do not care for the result of a Future then you should -## use the `asyncCheck` procedure instead of the `discard` keyword. Note -## however that this does not wait for completion, and you should use -## `waitFor` for that purpose. +## Futures should **never** be discarded directly because they may contain +## errors. If you do not care for the result of a Future then you should use +## the `asyncCheck` procedure instead of the `discard` keyword. Note that this +## does not wait for completion, and you should use `waitFor` or `await` for that purpose. +## +## .. note:: `await` also checks if the future fails, so you can safely discard +## its result. +## +## Handling futures +## ================ +## +## There are many different operations that apply to a future. +## The three primary high-level operations are `asyncCheck`, +## `waitFor`, and `await`. +## +## * `asyncCheck`: Raises an exception if the future fails. It neither waits +## for the future to finish nor returns the result of the future. +## * `waitFor`: Polls the event loop and blocks the current thread until the +## future finishes. This is often used to call an async procedure from a +## synchronous context and should never be used in an `async` proc. +## * `await`: Pauses execution in the current async procedure until the future +## finishes. While the current procedure is paused, other async procedures will +## continue running. Should be used instead of `waitFor` in an async +## procedure. +## +## Here is a handy quick reference chart showing their high-level differences: +## ============== ===================== ======================= +## Procedure Context Blocking +## ============== ===================== ======================= +## `asyncCheck` non-async and async non-blocking +## `waitFor` non-async blocks current thread +## `await` async suspends current proc +## ============== ===================== ======================= ## ## Examples ## ======== @@ -165,19 +196,49 @@ ## ================ ## ## * The effect system (`raises: []`) does not work with async procedures. +## * Mutable parameters are not supported by async procedures. +## +## +## Multiple async backend support +## ============================== +## +## Thanks to its powerful macro support, Nim allows ``async``/``await`` to be +## implemented in libraries with only minimal support from the language - as +## such, multiple ``async`` libraries exist, including ``asyncdispatch`` and +## ``chronos``, and more may come to be developed in the future. +## +## Libraries built on top of async/await may wish to support multiple async +## backends - the best way to do so is to create separate modules for each backend +## that may be imported side-by-side. +## +## An alternative way is to select backend using a global compile flag - this +## method makes it difficult to compose applications that use both backends as may +## happen with transitive dependencies, but may be appropriate in some cases - +## libraries choosing this path should call the flag `asyncBackend`, allowing +## applications to choose the backend with `-d:asyncBackend=<backend_name>`. +## +## Known `async` backends include: +## +## * `-d:asyncBackend=none`: disable `async` support completely +## * `-d:asyncBackend=asyncdispatch`: https://nim-lang.org/docs/asyncdispatch.html +## * `-d:asyncBackend=chronos`: https://github.com/status-im/nim-chronos/ +## +## ``none`` can be used when a library supports both a synchronous and +## asynchronous API, to disable the latter. + +import std/[os, tables, strutils, times, heapqueue, options, asyncstreams] +import std/[math, monotimes] +import std/asyncfutures except callSoon -import os, tables, strutils, times, heapqueue, options, asyncstreams -import options, math, std/monotimes -import asyncfutures except callSoon +import std/[nativesockets, net, deques] -import nativesockets, net, deques +when defined(nimPreviewSlimSystem): + import std/[assertions, syncio] export Port, SocketFlag export asyncfutures except callSoon export asyncstreams -#{.injectStmt: newGcInvariant().} - # TODO: Check if yielded future is nil and throw a more meaningful exception type @@ -220,6 +281,8 @@ proc adjustTimeout( result = max(nextTimer.get(), 0) result = min(pollTimeout, result) +proc runOnce(timeout: int): bool {.gcsafe.} + proc callSoon*(cbproc: proc () {.gcsafe.}) {.gcsafe.} ## Schedule `cbproc` to be called as soon as possible. ## The callback is called when control returns to the event loop. @@ -239,7 +302,7 @@ template implementSetInheritable() {.dirty.} = fd.FileHandle.setInheritable(inheritable) when defined(windows) or defined(nimdoc): - import winlean, sets, hashes + import std/[winlean, sets, hashes] type CompletionKey = ULONG_PTR @@ -329,7 +392,7 @@ when defined(windows) or defined(nimdoc): let p = getGlobalDispatcher() p.handles.len != 0 or p.timers.len != 0 or p.callbacks.len != 0 - proc runOnce(timeout = 500): bool = + proc runOnce(timeout: int): bool = let p = getGlobalDispatcher() if p.handles.len == 0 and p.timers.len == 0 and p.callbacks.len == 0: raise newException(ValueError, @@ -470,7 +533,7 @@ when defined(windows) or defined(nimdoc): if flags.isDisconnectionError(errcode): retFuture.complete("") else: - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) if dataBuf.buf != nil: dealloc dataBuf.buf dataBuf.buf = nil @@ -488,7 +551,7 @@ when defined(windows) or defined(nimdoc): if flags.isDisconnectionError(err): retFuture.complete("") else: - retFuture.fail(newException(OSError, osErrorMsg(err))) + retFuture.fail(newOSError(err)) elif ret == 0: # Request completed immediately. if bytesReceived != 0: @@ -540,7 +603,7 @@ when defined(windows) or defined(nimdoc): if flags.isDisconnectionError(errcode): retFuture.complete(0) else: - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) if dataBuf.buf != nil: dataBuf.buf = nil ) @@ -556,7 +619,7 @@ when defined(windows) or defined(nimdoc): if flags.isDisconnectionError(err): retFuture.complete(0) else: - retFuture.fail(newException(OSError, osErrorMsg(err))) + retFuture.fail(newOSError(err)) elif ret == 0: # Request completed immediately. if bytesReceived != 0: @@ -604,7 +667,7 @@ when defined(windows) or defined(nimdoc): if flags.isDisconnectionError(err): retFuture.complete() else: - retFuture.fail(newException(OSError, osErrorMsg(err))) + retFuture.fail(newOSError(err)) else: retFuture.complete() # We don't deallocate `ol` here because even though this completed @@ -639,7 +702,7 @@ when defined(windows) or defined(nimdoc): if errcode == OSErrorCode(-1): retFuture.complete() else: - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) ) let ret = WSASendTo(socket.SocketHandle, addr dataBuf, 1, addr bytesSent, @@ -649,7 +712,7 @@ when defined(windows) or defined(nimdoc): let err = osLastError() if err.int32 != ERROR_IO_PENDING: GC_unref(ol) - retFuture.fail(newException(OSError, osErrorMsg(err))) + retFuture.fail(newOSError(err)) else: retFuture.complete() # We don't deallocate `ol` here because even though this completed @@ -683,7 +746,7 @@ when defined(windows) or defined(nimdoc): else: # datagram sockets don't have disconnection, # so we can just raise an exception - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) ) let res = WSARecvFrom(socket.SocketHandle, addr dataBuf, 1, @@ -694,7 +757,7 @@ when defined(windows) or defined(nimdoc): let err = osLastError() if err.int32 != ERROR_IO_PENDING: GC_unref(ol) - retFuture.fail(newException(OSError, osErrorMsg(err))) + retFuture.fail(newOSError(err)) else: # Request completed immediately. if bytesReceived != 0: @@ -707,7 +770,7 @@ when defined(windows) or defined(nimdoc): proc acceptAddr*(socket: AsyncFD, flags = {SocketFlag.SafeDisconn}, inheritable = defined(nimInheritHandles)): - owned(Future[tuple[address: string, client: AsyncFD]]) = + owned(Future[tuple[address: string, client: AsyncFD]]) {.gcsafe.} = ## Accepts a new connection. Returns a future containing the client socket ## corresponding to that connection and the remote address of the client. ## The future will complete when the connection is successfully accepted. @@ -745,7 +808,7 @@ when defined(windows) or defined(nimdoc): else: retFuture.complete(newAcceptFut.read) else: - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) template completeAccept() {.dirty.} = var listenSock = socket @@ -774,7 +837,7 @@ when defined(windows) or defined(nimdoc): var ol = newCustom() ol.data = CompletionData(fd: socket, cb: - proc (fd: AsyncFD, bytesCount: DWORD, errcode: OSErrorCode) = + proc (fd: AsyncFD, bytesCount: DWORD, errcode: OSErrorCode) {.gcsafe.} = if not retFuture.finished: if errcode == OSErrorCode(-1): completeAccept() @@ -1024,7 +1087,7 @@ when defined(windows) or defined(nimdoc): raiseOSError(osLastError()) var pcd = cast[PostCallbackDataPtr](allocShared0(sizeof(PostCallbackData))) - var flags = WT_EXECUTEINWAITTHREAD.DWORD + var flags = WT_EXECUTEINWAITTHREAD.DWORD or WT_EXECUTEONLYONCE.DWORD proc proccb(fd: AsyncFD, bytesCount: DWORD, errcode: OSErrorCode) = closeWaitable(hProcess) @@ -1103,11 +1166,14 @@ when defined(windows) or defined(nimdoc): initAll() else: - import selectors - from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK, + import std/selectors + from std/posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK, MSG_NOSIGNAL when declared(posix.accept4): - from posix import accept4, SOCK_CLOEXEC + from std/posix import accept4, SOCK_CLOEXEC + when defined(genode): + import genode/env # get the implicit Genode env + import genode/signals const InitCallbackListSize = 4 # initial size of callbacks sequence, @@ -1126,6 +1192,8 @@ else: PDispatcher* = ref object of PDispatcherBase selector: Selector[AsyncData] + when defined(genode): + signalHandler: SignalHandler proc `==`*(x, y: AsyncFD): bool {.borrow.} proc `==`*(x, y: AsyncEvent): bool {.borrow.} @@ -1141,9 +1209,22 @@ else: result.selector = newSelector[AsyncData]() result.timers.clear() result.callbacks = initDeque[proc () {.closure, gcsafe.}](InitDelayedCallbackListSize) + when defined(genode): + let entrypoint = ep(cast[GenodeEnv](runtimeEnv)) + result.signalHandler = newSignalHandler(entrypoint): + discard runOnce(0) var gDisp{.threadvar.}: owned PDispatcher ## Global dispatcher + when defined(nuttx): + import std/exitprocs + + proc cleanDispatcher() {.noconv.} = + gDisp = nil + + proc addFinalyzer() = + addExitProc(cleanDispatcher) + proc setGlobalDispatcher*(disp: owned PDispatcher) = if not gDisp.isNil: assert gDisp.callbacks.len == 0 @@ -1153,6 +1234,8 @@ else: proc getGlobalDispatcher*(): PDispatcher = if gDisp.isNil: setGlobalDispatcher(newDispatcher()) + when defined(nuttx): + addFinalyzer() result = gDisp proc getIoHandler*(disp: PDispatcher): Selector[AsyncData] = @@ -1310,10 +1393,11 @@ else: ValueError, "Expecting async operations to stop when fd has closed." ) - - proc runOnce(timeout = 500): bool = + proc runOnce(timeout: int): bool = let p = getGlobalDispatcher() if p.selector.isEmpty() and p.timers.len == 0 and p.callbacks.len == 0: + when defined(genode): + if timeout == 0: return raise newException(ValueError, "No handles or timers registered in dispatcher.") @@ -1384,7 +1468,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. elif res == 0: @@ -1413,7 +1497,7 @@ else: if flags.isDisconnectionError(lastError): retFuture.complete(0) else: - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) else: result = false # We still want this callback to be called. else: @@ -1479,7 +1563,7 @@ else: let lastError = osLastError() if lastError.int32 != EINTR and lastError.int32 != EWOULDBLOCK and lastError.int32 != EAGAIN: - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) else: result = false # We still want this callback to be called. else: @@ -1505,7 +1589,7 @@ else: let lastError = osLastError() if lastError.int32 != EINTR and lastError.int32 != EWOULDBLOCK and lastError.int32 != EAGAIN: - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) else: result = false else: @@ -1518,7 +1602,7 @@ else: owned(Future[tuple[address: string, client: AsyncFD]]) = var retFuture = newFuture[tuple[address: string, client: AsyncFD]]("acceptAddr") - proc cb(sock: AsyncFD): bool = + proc cb(sock: AsyncFD): bool {.gcsafe.} = result = true var sockAddress: Sockaddr_storage var addrLen = sizeof(sockAddress).SockLen @@ -1546,7 +1630,7 @@ else: if flags.isDisconnectionError(lastError): return false else: - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) else: try: let address = getAddrString(cast[ptr SockAddr](addr sockAddress)) @@ -1675,9 +1759,11 @@ when defined(windows) or defined(nimdoc): proc (fd: AsyncFD, bytesCount: DWORD, errcode: OSErrorCode) = if not retFuture.finished: if errcode == OSErrorCode(-1): + const SO_UPDATE_CONNECT_CONTEXT = 0x7010 + socket.SocketHandle.setSockOptInt(SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, 1) # 15022 retFuture.complete() else: - retFuture.fail(newException(OSError, osErrorMsg(errcode))) + retFuture.fail(newOSError(errcode)) ) let ret = connectEx(socket.SocketHandle, addrInfo.ai_addr, @@ -1695,7 +1781,7 @@ when defined(windows) or defined(nimdoc): # With ERROR_IO_PENDING `ol` will be deallocated in `poll`, # and the future will be completed/failed there, too. GC_unref(ol) - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) else: proc doConnect(socket: AsyncFD, addrInfo: ptr AddrInfo): owned(Future[void]) = let retFuture = newFuture[void]("doConnect") @@ -1712,7 +1798,7 @@ else: # interrupted, keep waiting return false else: - retFuture.fail(newException(OSError, osErrorMsg(OSErrorCode(ret)))) + retFuture.fail(newOSError(OSErrorCode(ret))) return true let ret = connect(socket.SocketHandle, @@ -1726,7 +1812,7 @@ else: if lastError.int32 == EINTR or lastError.int32 == EINPROGRESS: addWrite(socket, cb) else: - retFuture.fail(newException(OSError, osErrorMsg(lastError))) + retFuture.fail(newOSError(lastError)) template asyncAddrInfoLoop(addrInfo: ptr AddrInfo, fd: untyped, protocol: Protocol = IPPROTO_RAW) = @@ -1765,7 +1851,7 @@ template asyncAddrInfoLoop(addrInfo: ptr AddrInfo, fd: untyped, curAddrInfo = curAddrInfo.ai_next if curAddrInfo == nil: - freeaddrinfo(addrInfo) + freeAddrInfo(addrInfo) when shouldCreateFd: closeUnusedFds() if lastException != nil: @@ -1781,7 +1867,7 @@ template asyncAddrInfoLoop(addrInfo: ptr AddrInfo, fd: untyped, try: curFd = createAsyncNativeSocket(domain, sockType, protocol) except: - freeaddrinfo(addrInfo) + freeAddrInfo(addrInfo) closeUnusedFds() raise getCurrentException() when defined(windows): @@ -1791,7 +1877,7 @@ template asyncAddrInfoLoop(addrInfo: ptr AddrInfo, fd: untyped, doConnect(curFd, curAddrInfo).callback = tryNextAddrInfo curAddrInfo = curAddrInfo.ai_next else: - freeaddrinfo(addrInfo) + freeAddrInfo(addrInfo) when shouldCreateFd: closeUnusedFds(ord(domain)) retFuture.complete(curFd) @@ -1908,7 +1994,8 @@ proc send*(socket: AsyncFD, data: string, return retFuture # -- Await Macro -include asyncmacro +import std/asyncmacro +export asyncmacro proc readAll*(future: FutureStream[string]): owned(Future[string]) {.async.} = ## Returns a future that will complete when all the string data from the @@ -1945,17 +2032,34 @@ proc activeDescriptors*(): int {.inline.} = result = getGlobalDispatcher().selector.count when defined(posix): - import posix + import std/posix -when defined(linux) or defined(windows) or defined(macosx) or defined(bsd): +when defined(linux) or defined(windows) or defined(macosx) or defined(bsd) or + defined(solaris) or defined(zephyr) or defined(freertos) or defined(nuttx) or defined(haiku): proc maxDescriptors*(): int {.raises: OSError.} = ## Returns the maximum number of active file descriptors for the current ## process. This involves a system call. For now `maxDescriptors` is - ## supported on the following OSes: Windows, Linux, OSX, BSD. + ## supported on the following OSes: Windows, Linux, OSX, BSD, Solaris. when defined(windows): result = 16_700_000 + elif defined(zephyr) or defined(freertos): + result = FD_MAX else: var fdLim: RLimit if getrlimit(RLIMIT_NOFILE, fdLim) < 0: raiseOSError(osLastError()) result = int(fdLim.rlim_cur) - 1 + +when defined(genode): + proc scheduleCallbacks*(): bool {.discardable.} = + ## *Genode only.* + ## Schedule callback processing and return immediately. + ## Returns `false` if there is nothing to schedule. + ## RPC servers should call this to dispatch `callSoon` + ## bodies after retiring an RPC to its client. + ## This is effectively a non-blocking `poll(…)` and is + ## equivalent to scheduling a momentary no-op timeout + ## but faster and with less overhead. + let dis = getGlobalDispatcher() + result = dis.callbacks.len > 0 + if result: submit(dis.signalHandler.cap) |