2018-05-16 08:22:34 +00:00
|
|
|
#
|
2019-02-06 14:49:11 +00:00
|
|
|
# Chronos
|
2018-05-16 08:22:34 +00:00
|
|
|
#
|
2019-02-06 14:49:11 +00:00
|
|
|
# (c) Copyright 2015 Dominik Picheta
|
|
|
|
# (c) Copyright 2018-Present Status Research & Development GmbH
|
2018-05-16 08:22:34 +00:00
|
|
|
#
|
|
|
|
# Licensed under either of
|
|
|
|
# Apache License, version 2.0, (LICENSE-APACHEv2)
|
|
|
|
# MIT license (LICENSE-MIT)
|
|
|
|
|
|
|
|
include "system/inclrtl"
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
import os, tables, strutils, heapqueue, lists, options, nativesockets, net,
|
2020-01-08 01:06:27 +00:00
|
|
|
deques
|
2020-08-15 22:45:41 +00:00
|
|
|
import timer
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
export Port, SocketFlag
|
2019-09-10 17:19:49 +00:00
|
|
|
export timer
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
#{.injectStmt: newGcInvariant().}
|
|
|
|
|
|
|
|
## AsyncDispatch
|
|
|
|
## *************
|
|
|
|
##
|
|
|
|
## This module implements asynchronous IO. This includes a dispatcher,
|
|
|
|
## a ``Future`` type implementation, and an ``async`` macro which allows
|
|
|
|
## asynchronous code to be written in a synchronous style with the ``await``
|
|
|
|
## keyword.
|
|
|
|
##
|
|
|
|
## The dispatcher acts as a kind of event loop. You must call ``poll`` on it
|
|
|
|
## (or a function which does so for you such as ``waitFor`` or ``runForever``)
|
|
|
|
## in order to poll for any outstanding events. The underlying implementation
|
|
|
|
## is based on epoll on Linux, IO Completion Ports on Windows and select on
|
|
|
|
## other operating systems.
|
|
|
|
##
|
|
|
|
## The ``poll`` function will not, on its own, return any events. Instead
|
|
|
|
## an appropriate ``Future`` object will be completed. A ``Future`` is a
|
|
|
|
## type which holds a value which is not yet available, but which *may* be
|
|
|
|
## available in the future. You can check whether a future is finished
|
|
|
|
## by using the ``finished`` function. When a future is finished it means that
|
|
|
|
## either the value that it holds is now available or it holds an error instead.
|
|
|
|
## The latter situation occurs when the operation to complete a future fails
|
|
|
|
## with an exception. You can distinguish between the two situations with the
|
|
|
|
## ``failed`` function.
|
|
|
|
##
|
|
|
|
## Future objects can also store a callback procedure which will be called
|
|
|
|
## automatically once the future completes.
|
|
|
|
##
|
|
|
|
## Futures therefore can be thought of as an implementation of the proactor
|
|
|
|
## pattern. In this
|
|
|
|
## pattern you make a request for an action, and once that action is fulfilled
|
|
|
|
## a future is completed with the result of that action. Requests can be
|
|
|
|
## made by calling the appropriate functions. For example: calling the ``recv``
|
|
|
|
## function will create a request for some data to be read from a socket. The
|
|
|
|
## future which the ``recv`` function returns will then complete once the
|
|
|
|
## 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)
|
|
|
|
## )
|
|
|
|
##
|
|
|
|
## All asynchronous functions returning a ``Future`` will not block. They
|
|
|
|
## will not however return immediately. An asynchronous function will have
|
|
|
|
## code which will be executed before an asynchronous request is made, in most
|
|
|
|
## cases this code sets up the request.
|
|
|
|
##
|
|
|
|
## In the above example, the ``recv`` function will return a brand new
|
|
|
|
## ``Future`` instance once the request for data to be read from the socket
|
|
|
|
## is made. This ``Future`` instance will complete once the requested amount
|
|
|
|
## of data is read, in this case it is 100 bytes. The second line sets a
|
|
|
|
## 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.
|
|
|
|
##
|
|
|
|
## Asynchronous procedures
|
|
|
|
## -----------------------
|
|
|
|
##
|
|
|
|
## Asynchronous procedures remove the pain of working with callbacks. They do
|
|
|
|
## this by allowing you to write asynchronous code the same way as you would
|
|
|
|
## write synchronous code.
|
|
|
|
##
|
|
|
|
## An asynchronous procedure is marked using the ``{.async.}`` pragma.
|
|
|
|
## When marking a procedure with the ``{.async.}`` pragma it must have a
|
|
|
|
## ``Future[T]`` return type or no return type at all. If you do not specify
|
|
|
|
## a return type then ``Future[void]`` is assumed.
|
|
|
|
##
|
|
|
|
## Inside asynchronous procedures ``await`` can be used to call any
|
|
|
|
## procedures which return a
|
|
|
|
## ``Future``; this includes asynchronous procedures. When a procedure is
|
|
|
|
## "awaited", the asynchronous procedure it is awaited in will
|
|
|
|
## suspend its execution
|
|
|
|
## until the awaited procedure's Future completes. At which point the
|
|
|
|
## asynchronous procedure will resume its execution. During the period
|
|
|
|
## when an asynchronous procedure is suspended other asynchronous procedures
|
|
|
|
## will be run by the dispatcher.
|
|
|
|
##
|
|
|
|
## The ``await`` call may be used in many contexts. It can be used on the right
|
|
|
|
## hand side of a variable declaration: ``var data = await socket.recv(100)``,
|
|
|
|
## in which case the variable will be set to the value of the future
|
|
|
|
## automatically. It can be used to await a ``Future`` object, and it can
|
|
|
|
## be used to await a procedure returning a ``Future[void]``:
|
|
|
|
## ``await socket.send("foobar")``.
|
|
|
|
##
|
|
|
|
## If an awaited future completes with an error, then ``await`` will re-raise
|
|
|
|
## this error. To avoid this, you can use the ``yield`` keyword instead of
|
|
|
|
## ``await``. The following section shows different ways that you can handle
|
|
|
|
## exceptions in async procs.
|
|
|
|
##
|
|
|
|
## 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:
|
|
|
|
##
|
|
|
|
## .. code-block:: Nim
|
|
|
|
## var future = sock.recv(100)
|
|
|
|
## yield future
|
|
|
|
## if future.failed:
|
|
|
|
## # 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
|
|
|
|
##
|
|
|
|
## 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.
|
|
|
|
##
|
|
|
|
##
|
|
|
|
## 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.
|
|
|
|
##
|
|
|
|
## Examples
|
|
|
|
## --------
|
|
|
|
##
|
|
|
|
## For examples take a look at the documentation for the modules implementing
|
|
|
|
## asynchronous IO. A good place to start is the
|
|
|
|
## `asyncnet module <asyncnet.html>`_.
|
|
|
|
##
|
|
|
|
## Limitations/Bugs
|
|
|
|
## ----------------
|
|
|
|
##
|
|
|
|
## * The effect system (``raises: []``) does not work with async procedures.
|
|
|
|
## * Can't await in a ``except`` body
|
|
|
|
## * Forward declarations for async procs are broken,
|
|
|
|
## link includes workaround: https://github.com/nim-lang/Nim/issues/3182.
|
|
|
|
|
|
|
|
# TODO: Check if yielded future is nil and throw a more meaningful exception
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
const unixPlatform = defined(macosx) or defined(freebsd) or
|
|
|
|
defined(netbsd) or defined(openbsd) or
|
|
|
|
defined(dragonfly) or defined(macos) or
|
|
|
|
defined(linux) or defined(android) or
|
|
|
|
defined(solaris)
|
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
when defined(windows):
|
|
|
|
import winlean, sets, hashes
|
2019-09-10 17:19:49 +00:00
|
|
|
elif unixPlatform:
|
2019-03-24 16:57:36 +00:00
|
|
|
import selectors
|
|
|
|
from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK,
|
2019-07-15 09:59:42 +00:00
|
|
|
MSG_NOSIGNAL, SIGPIPE
|
2019-03-24 16:57:36 +00:00
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
type
|
2019-09-10 17:19:49 +00:00
|
|
|
CallbackFunc* = proc (arg: pointer = nil) {.gcsafe.}
|
|
|
|
CallSoonProc* = proc (c: CallbackFunc, u: pointer = nil) {.gcsafe.}
|
|
|
|
|
2020-01-08 17:06:56 +00:00
|
|
|
AsyncCallback* = object
|
2019-09-10 17:19:49 +00:00
|
|
|
function*: CallbackFunc
|
|
|
|
udata*: pointer
|
|
|
|
|
2019-03-26 10:29:45 +00:00
|
|
|
AsyncError* = object of CatchableError
|
2018-07-20 08:58:01 +00:00
|
|
|
## Generic async exception
|
|
|
|
AsyncTimeoutError* = object of AsyncError
|
|
|
|
## Timeout exception
|
|
|
|
|
2020-01-07 05:26:18 +00:00
|
|
|
TimerCallback* = ref object
|
2019-03-24 16:57:36 +00:00
|
|
|
finishAt*: Moment
|
2018-05-16 08:22:34 +00:00
|
|
|
function*: AsyncCallback
|
|
|
|
|
2019-04-04 09:34:23 +00:00
|
|
|
TrackerBase* = ref object of RootRef
|
|
|
|
id*: string
|
|
|
|
dump*: proc(): string {.gcsafe.}
|
|
|
|
isLeaked*: proc(): bool {.gcsafe.}
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
PDispatcherBase = ref object of RootRef
|
|
|
|
timers*: HeapQueue[TimerCallback]
|
|
|
|
callbacks*: Deque[AsyncCallback]
|
2019-04-04 09:34:23 +00:00
|
|
|
trackers*: Table[string, TrackerBase]
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
proc `<`(a, b: TimerCallback): bool =
|
|
|
|
result = a.finishAt < b.finishAt
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
proc callSoon*(cbproc: CallbackFunc, data: pointer = nil) {.gcsafe.}
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
func getAsyncTimestamp*(a: Duration): auto {.inline.} =
|
|
|
|
## Return rounded up value of duration with milliseconds resolution.
|
|
|
|
##
|
|
|
|
## This function also take care on int32 overflow, because Linux and Windows
|
|
|
|
## accepts signed 32bit integer as timeout.
|
|
|
|
let milsec = Millisecond.nanoseconds()
|
|
|
|
let nansec = a.nanoseconds()
|
|
|
|
var res = nansec div milsec
|
|
|
|
let mid = nansec mod milsec
|
|
|
|
when defined(windows):
|
|
|
|
res = min(cast[int64](high(int32) - 1), res)
|
|
|
|
result = cast[DWORD](res)
|
|
|
|
result += DWORD(min(1'i32, cast[int32](mid)))
|
|
|
|
else:
|
|
|
|
res = min(cast[int64](high(int32) - 1), res)
|
|
|
|
result = cast[int32](res)
|
|
|
|
result += min(1, cast[int32](mid))
|
|
|
|
|
2018-06-02 23:03:48 +00:00
|
|
|
template processTimersGetTimeout(loop, timeout: untyped) =
|
2020-01-08 16:03:34 +00:00
|
|
|
var lastFinish = curTime
|
|
|
|
while loop.timers.len > 0:
|
2020-09-15 07:55:43 +00:00
|
|
|
if loop.timers[0].function.function.isNil:
|
2020-01-08 16:03:34 +00:00
|
|
|
discard loop.timers.pop()
|
|
|
|
continue
|
2020-01-07 05:26:18 +00:00
|
|
|
|
2020-01-08 16:03:34 +00:00
|
|
|
lastFinish = loop.timers[0].finishAt
|
|
|
|
if curTime < lastFinish:
|
|
|
|
break
|
2020-01-07 05:26:18 +00:00
|
|
|
|
2020-01-08 16:03:34 +00:00
|
|
|
loop.callbacks.addLast(loop.timers.pop().function)
|
|
|
|
if loop.timers.len > 0:
|
|
|
|
timeout = (lastFinish - curTime).getAsyncTimestamp()
|
2018-06-02 23:03:48 +00:00
|
|
|
|
|
|
|
if timeout == 0:
|
|
|
|
if len(loop.callbacks) == 0:
|
|
|
|
when defined(windows):
|
|
|
|
timeout = INFINITE
|
|
|
|
else:
|
|
|
|
timeout = -1
|
|
|
|
else:
|
|
|
|
if len(loop.callbacks) != 0:
|
|
|
|
timeout = 0
|
|
|
|
|
|
|
|
template processTimers(loop: untyped) =
|
2019-03-24 16:57:36 +00:00
|
|
|
var curTime = Moment.now()
|
2020-01-08 16:03:34 +00:00
|
|
|
while loop.timers.len > 0:
|
2020-09-15 07:55:43 +00:00
|
|
|
if loop.timers[0].function.function.isNil:
|
2020-01-08 16:03:34 +00:00
|
|
|
discard loop.timers.pop()
|
|
|
|
continue
|
2020-01-07 05:26:18 +00:00
|
|
|
|
2020-01-08 16:03:34 +00:00
|
|
|
if curTime < loop.timers[0].finishAt:
|
|
|
|
break
|
|
|
|
loop.callbacks.addLast(loop.timers.pop().function)
|
2018-06-02 23:03:48 +00:00
|
|
|
|
|
|
|
template processCallbacks(loop: untyped) =
|
|
|
|
var count = len(loop.callbacks)
|
|
|
|
for i in 0..<count:
|
2018-08-06 12:15:05 +00:00
|
|
|
# This is mostly workaround for people which are using `waitFor` where
|
|
|
|
# it must be used `await`. While using `waitFor` inside of callbacks
|
|
|
|
# dispatcher's callback list is got decreased and length of
|
|
|
|
# `loop.callbacks` become not equal to `count`, its why `IndexError`
|
|
|
|
# can be generated.
|
|
|
|
if len(loop.callbacks) == 0: break
|
2018-06-02 23:03:48 +00:00
|
|
|
let callable = loop.callbacks.popFirst()
|
2018-12-12 11:07:41 +00:00
|
|
|
if not isNil(callable.function):
|
|
|
|
callable.function(callable.udata)
|
2018-06-02 23:03:48 +00:00
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
when defined(windows) or defined(nimdoc):
|
|
|
|
type
|
|
|
|
WSAPROC_TRANSMITFILE = proc(hSocket: SocketHandle, hFile: Handle,
|
|
|
|
nNumberOfBytesToWrite: DWORD,
|
|
|
|
nNumberOfBytesPerSend: DWORD,
|
|
|
|
lpOverlapped: POVERLAPPED,
|
|
|
|
lpTransmitBuffers: pointer,
|
|
|
|
dwReserved: DWORD): cint {.
|
2019-03-26 12:29:35 +00:00
|
|
|
gcsafe, stdcall.}
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
CompletionKey = ULONG_PTR
|
|
|
|
|
|
|
|
CompletionData* = object
|
|
|
|
fd*: AsyncFD
|
|
|
|
cb*: CallbackFunc
|
|
|
|
errCode*: OSErrorCode
|
|
|
|
bytesCount*: int32
|
|
|
|
udata*: pointer
|
|
|
|
|
2018-05-25 01:05:13 +00:00
|
|
|
CustomOverlapped* = object of OVERLAPPED
|
|
|
|
data*: CompletionData
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
PDispatcher* = ref object of PDispatcherBase
|
|
|
|
ioPort: Handle
|
|
|
|
handles: HashSet[AsyncFD]
|
|
|
|
connectEx*: WSAPROC_CONNECTEX
|
|
|
|
acceptEx*: WSAPROC_ACCEPTEX
|
|
|
|
getAcceptExSockAddrs*: WSAPROC_GETACCEPTEXSOCKADDRS
|
|
|
|
transmitFile*: WSAPROC_TRANSMITFILE
|
|
|
|
|
2018-05-21 21:52:57 +00:00
|
|
|
PtrCustomOverlapped* = ptr CustomOverlapped
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
RefCustomOverlapped* = ref CustomOverlapped
|
|
|
|
|
|
|
|
AsyncFD* = distinct int
|
|
|
|
|
|
|
|
proc hash(x: AsyncFD): Hash {.borrow.}
|
|
|
|
proc `==`*(x: AsyncFD, y: AsyncFD): bool {.borrow.}
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
proc getFunc(s: SocketHandle, fun: var pointer, guid: var GUID): bool =
|
|
|
|
var bytesRet: DWORD
|
|
|
|
fun = nil
|
|
|
|
result = WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER, addr guid,
|
|
|
|
sizeof(GUID).DWORD, addr fun, sizeof(pointer).DWORD,
|
|
|
|
addr bytesRet, nil, nil) == 0
|
|
|
|
|
2019-09-23 17:24:26 +00:00
|
|
|
proc globalInit() =
|
|
|
|
var wsa: WSAData
|
|
|
|
if wsaStartup(0x0202'i16, addr wsa) != 0:
|
|
|
|
raiseOSError(osLastError())
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
proc initAPI(loop: PDispatcher) =
|
|
|
|
var
|
|
|
|
WSAID_TRANSMITFILE = GUID(
|
|
|
|
D1: 0xb5367df0'i32, D2: 0xcbac'i16, D3: 0x11cf'i16,
|
|
|
|
D4: [0x95'i8, 0xca'i8, 0x00'i8, 0x80'i8,
|
|
|
|
0x5f'i8, 0x48'i8, 0xa1'i8, 0x92'i8])
|
|
|
|
|
|
|
|
let sock = winlean.socket(winlean.AF_INET, 1, 6)
|
|
|
|
if sock == INVALID_SOCKET:
|
|
|
|
raiseOSError(osLastError())
|
|
|
|
|
|
|
|
var funcPointer: pointer = nil
|
|
|
|
if not getFunc(sock, funcPointer, WSAID_CONNECTEX):
|
|
|
|
let err = osLastError()
|
|
|
|
close(sock)
|
|
|
|
raiseOSError(err)
|
|
|
|
loop.connectEx = cast[WSAPROC_CONNECTEX](funcPointer)
|
|
|
|
if not getFunc(sock, funcPointer, WSAID_ACCEPTEX):
|
|
|
|
let err = osLastError()
|
|
|
|
close(sock)
|
|
|
|
raiseOSError(err)
|
|
|
|
loop.acceptEx = cast[WSAPROC_ACCEPTEX](funcPointer)
|
|
|
|
if not getFunc(sock, funcPointer, WSAID_GETACCEPTEXSOCKADDRS):
|
|
|
|
let err = osLastError()
|
|
|
|
close(sock)
|
|
|
|
raiseOSError(err)
|
|
|
|
loop.getAcceptExSockAddrs = cast[WSAPROC_GETACCEPTEXSOCKADDRS](funcPointer)
|
|
|
|
if not getFunc(sock, funcPointer, WSAID_TRANSMITFILE):
|
|
|
|
let err = osLastError()
|
|
|
|
close(sock)
|
|
|
|
raiseOSError(err)
|
|
|
|
loop.transmitFile = cast[WSAPROC_TRANSMITFILE](funcPointer)
|
|
|
|
close(sock)
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
proc newDispatcher*(): PDispatcher =
|
|
|
|
## Creates a new Dispatcher instance.
|
|
|
|
new result
|
|
|
|
result.ioPort = createIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 1)
|
2019-10-24 13:01:57 +00:00
|
|
|
when declared(initHashSet):
|
2019-06-04 16:51:35 +00:00
|
|
|
# After 0.20.0 Nim's stdlib version
|
|
|
|
result.handles = initHashSet[AsyncFD]()
|
|
|
|
else:
|
|
|
|
# Pre 0.20.0 Nim's stdlib version
|
|
|
|
result.handles = initSet[AsyncFD]()
|
2019-10-24 13:01:57 +00:00
|
|
|
when declared(initHeapQueue):
|
2019-06-04 16:51:35 +00:00
|
|
|
# After 0.20.0 Nim's stdlib version
|
|
|
|
result.timers = initHeapQueue[TimerCallback]()
|
|
|
|
else:
|
|
|
|
# Pre 0.20.0 Nim's stdlib version
|
|
|
|
result.timers = newHeapQueue[TimerCallback]()
|
2018-05-16 08:22:34 +00:00
|
|
|
result.callbacks = initDeque[AsyncCallback](64)
|
2019-04-04 09:34:23 +00:00
|
|
|
result.trackers = initTable[string, TrackerBase]()
|
2019-09-10 17:19:49 +00:00
|
|
|
initAPI(result)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
|
|
|
|
|
|
|
|
proc setGlobalDispatcher*(disp: PDispatcher) =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Set current thread's dispatcher instance to ``disp``.
|
2018-05-16 08:22:34 +00:00
|
|
|
if not gDisp.isNil:
|
2019-03-14 03:03:32 +00:00
|
|
|
doAssert gDisp.callbacks.len == 0
|
2018-05-16 08:22:34 +00:00
|
|
|
gDisp = disp
|
|
|
|
|
|
|
|
proc getGlobalDispatcher*(): PDispatcher =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Returns current thread's dispatcher instance.
|
2018-05-16 08:22:34 +00:00
|
|
|
if gDisp.isNil:
|
|
|
|
setGlobalDispatcher(newDispatcher())
|
|
|
|
result = gDisp
|
|
|
|
|
|
|
|
proc getIoHandler*(disp: PDispatcher): Handle =
|
|
|
|
## Returns the underlying IO Completion Port handle (Windows) or selector
|
|
|
|
## (Unix) for the specified dispatcher.
|
|
|
|
return disp.ioPort
|
|
|
|
|
|
|
|
proc register*(fd: AsyncFD) =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Register file descriptor ``fd`` in thread's dispatcher.
|
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
if createIoCompletionPort(fd.Handle, loop.ioPort,
|
2018-05-16 08:22:34 +00:00
|
|
|
cast[CompletionKey](fd), 1) == 0:
|
|
|
|
raiseOSError(osLastError())
|
2018-08-24 12:20:08 +00:00
|
|
|
loop.handles.incl(fd)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
proc poll*() =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Perform single asynchronous step.
|
2018-05-16 08:22:34 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2019-03-24 16:57:36 +00:00
|
|
|
var curTime = Moment.now()
|
2018-05-16 08:22:34 +00:00
|
|
|
var curTimeout = DWORD(0)
|
|
|
|
|
|
|
|
# Moving expired timers to `loop.callbacks` and calculate timeout
|
2018-06-02 23:03:48 +00:00
|
|
|
loop.processTimersGetTimeout(curTimeout)
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
# Processing handles
|
|
|
|
var lpNumberOfBytesTransferred: Dword
|
|
|
|
var lpCompletionKey: ULONG_PTR
|
2018-05-21 21:52:57 +00:00
|
|
|
var customOverlapped: PtrCustomOverlapped
|
2018-06-02 23:18:32 +00:00
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
let res = getQueuedCompletionStatus(
|
2019-03-24 16:57:36 +00:00
|
|
|
loop.ioPort, addr lpNumberOfBytesTransferred,
|
|
|
|
addr lpCompletionKey, cast[ptr POVERLAPPED](addr customOverlapped),
|
|
|
|
curTimeout).bool
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
if res:
|
|
|
|
customOverlapped.data.bytesCount = lpNumberOfBytesTransferred
|
|
|
|
customOverlapped.data.errCode = OSErrorCode(-1)
|
|
|
|
let acb = AsyncCallback(function: customOverlapped.data.cb,
|
|
|
|
udata: cast[pointer](customOverlapped))
|
|
|
|
loop.callbacks.addLast(acb)
|
|
|
|
else:
|
|
|
|
let errCode = osLastError()
|
|
|
|
if customOverlapped != nil:
|
|
|
|
customOverlapped.data.errCode = errCode
|
|
|
|
let acb = AsyncCallback(function: customOverlapped.data.cb,
|
|
|
|
udata: cast[pointer](customOverlapped))
|
|
|
|
loop.callbacks.addLast(acb)
|
|
|
|
else:
|
|
|
|
if int32(errCode) != WAIT_TIMEOUT:
|
|
|
|
raiseOSError(errCode)
|
|
|
|
|
|
|
|
# Moving expired timers to `loop.callbacks`.
|
2018-06-02 23:03:48 +00:00
|
|
|
loop.processTimers()
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
# All callbacks which will be added in process will be processed on next
|
|
|
|
# poll() call.
|
2018-06-02 23:03:48 +00:00
|
|
|
loop.processCallbacks()
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-10-02 10:50:14 +00:00
|
|
|
proc closeSocket*(fd: AsyncFD, aftercb: CallbackFunc = nil) =
|
2018-05-16 08:22:34 +00:00
|
|
|
## Closes a socket and ensures that it is unregistered.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2018-10-02 10:50:14 +00:00
|
|
|
loop.handles.excl(fd)
|
|
|
|
close(SocketHandle(fd))
|
2018-08-24 12:20:08 +00:00
|
|
|
if not isNil(aftercb):
|
|
|
|
var acb = AsyncCallback(function: aftercb)
|
|
|
|
loop.callbacks.addLast(acb)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-10-25 10:19:19 +00:00
|
|
|
proc closeHandle*(fd: AsyncFD, aftercb: CallbackFunc = nil) =
|
|
|
|
## Closes a (pipe/file) handle and ensures that it is unregistered.
|
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
loop.handles.excl(fd)
|
2020-06-24 08:21:52 +00:00
|
|
|
discard closeHandle(Handle(fd))
|
2018-10-25 10:19:19 +00:00
|
|
|
if not isNil(aftercb):
|
|
|
|
var acb = AsyncCallback(function: aftercb)
|
|
|
|
loop.callbacks.addLast(acb)
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
proc unregister*(fd: AsyncFD) =
|
|
|
|
## Unregisters ``fd``.
|
|
|
|
getGlobalDispatcher().handles.excl(fd)
|
|
|
|
|
|
|
|
proc contains*(disp: PDispatcher, fd: AsyncFD): bool =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Returns ``true`` if ``fd`` is registered in thread's dispatcher.
|
2018-05-16 08:22:34 +00:00
|
|
|
return fd in disp.handles
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
elif unixPlatform:
|
|
|
|
const
|
|
|
|
SIG_IGN = cast[proc(x: cint) {.noconv,gcsafe.}](1)
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
type
|
|
|
|
AsyncFD* = distinct cint
|
|
|
|
|
|
|
|
CompletionData* = object
|
|
|
|
fd*: AsyncFD
|
|
|
|
udata*: pointer
|
|
|
|
|
|
|
|
PCompletionData* = ptr CompletionData
|
|
|
|
|
|
|
|
SelectorData* = object
|
|
|
|
reader*: AsyncCallback
|
|
|
|
rdata*: CompletionData
|
|
|
|
writer*: AsyncCallback
|
|
|
|
wdata*: CompletionData
|
|
|
|
|
|
|
|
PDispatcher* = ref object of PDispatcherBase
|
|
|
|
selector: Selector[SelectorData]
|
|
|
|
keys: seq[ReadyKey]
|
|
|
|
|
|
|
|
proc `==`*(x, y: AsyncFD): bool {.borrow.}
|
|
|
|
|
2019-09-23 17:24:26 +00:00
|
|
|
proc globalInit() =
|
2019-09-10 17:19:49 +00:00
|
|
|
# We are ignoring SIGPIPE signal, because we are working with EPIPE.
|
|
|
|
posix.signal(cint(SIGPIPE), SIG_IGN)
|
|
|
|
|
2019-09-23 17:24:26 +00:00
|
|
|
proc initAPI(disp: PDispatcher) =
|
|
|
|
discard
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
proc newDispatcher*(): PDispatcher =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Create new dispatcher.
|
2018-05-16 08:22:34 +00:00
|
|
|
new result
|
|
|
|
result.selector = newSelector[SelectorData]()
|
2019-10-24 13:01:57 +00:00
|
|
|
when declared(initHeapQueue):
|
|
|
|
# After 0.20.0 Nim's stdlib version
|
2019-10-24 13:06:55 +00:00
|
|
|
result.timers = initHeapQueue[TimerCallback]()
|
2019-10-24 13:01:57 +00:00
|
|
|
else:
|
|
|
|
# Before 0.20.0 Nim's stdlib version
|
|
|
|
result.timers.newHeapQueue()
|
2018-05-16 08:22:34 +00:00
|
|
|
result.callbacks = initDeque[AsyncCallback](64)
|
|
|
|
result.keys = newSeq[ReadyKey](64)
|
2019-04-04 09:34:23 +00:00
|
|
|
result.trackers = initTable[string, TrackerBase]()
|
2019-09-10 17:19:49 +00:00
|
|
|
initAPI(result)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
|
|
|
|
|
|
|
|
proc setGlobalDispatcher*(disp: PDispatcher) =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Set current thread's dispatcher instance to ``disp``.
|
2018-05-16 08:22:34 +00:00
|
|
|
if not gDisp.isNil:
|
2019-03-14 03:03:32 +00:00
|
|
|
doAssert gDisp.callbacks.len == 0
|
2018-05-16 08:22:34 +00:00
|
|
|
gDisp = disp
|
|
|
|
|
|
|
|
proc getGlobalDispatcher*(): PDispatcher =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Returns current thread's dispatcher instance.
|
2018-05-16 08:22:34 +00:00
|
|
|
if gDisp.isNil:
|
|
|
|
setGlobalDispatcher(newDispatcher())
|
|
|
|
result = gDisp
|
|
|
|
|
|
|
|
proc getIoHandler*(disp: PDispatcher): Selector[SelectorData] =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Returns system specific OS queue.
|
2018-05-16 08:22:34 +00:00
|
|
|
return disp.selector
|
|
|
|
|
|
|
|
proc register*(fd: AsyncFD) =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Register file descriptor ``fd`` in thread's dispatcher.
|
|
|
|
let loop = getGlobalDispatcher()
|
2018-05-16 08:22:34 +00:00
|
|
|
var data: SelectorData
|
|
|
|
data.rdata.fd = fd
|
|
|
|
data.wdata.fd = fd
|
|
|
|
loop.selector.registerHandle(int(fd), {}, data)
|
|
|
|
|
|
|
|
proc unregister*(fd: AsyncFD) =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Unregister file descriptor ``fd`` from thread's dispatcher.
|
2018-05-16 08:22:34 +00:00
|
|
|
getGlobalDispatcher().selector.unregister(int(fd))
|
|
|
|
|
|
|
|
proc contains*(disp: PDispatcher, fd: AsyncFd): bool {.inline.} =
|
2018-08-24 12:20:08 +00:00
|
|
|
## Returns ``true`` if ``fd`` is registered in thread's dispatcher.
|
2018-05-16 08:22:34 +00:00
|
|
|
result = int(fd) in disp.selector
|
|
|
|
|
|
|
|
proc addReader*(fd: AsyncFD, cb: CallbackFunc, udata: pointer = nil) =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Start watching the file descriptor ``fd`` for read availability and then
|
|
|
|
## call the callback ``cb`` with specified argument ``udata``.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2018-05-16 08:22:34 +00:00
|
|
|
var newEvents = {Event.Read}
|
2018-08-24 12:20:08 +00:00
|
|
|
withData(loop.selector, int(fd), adata) do:
|
2018-05-16 08:22:34 +00:00
|
|
|
let acb = AsyncCallback(function: cb, udata: addr adata.rdata)
|
|
|
|
adata.reader = acb
|
2018-06-08 05:42:48 +00:00
|
|
|
adata.rdata = CompletionData(fd: fd, udata: udata)
|
2018-05-16 08:22:34 +00:00
|
|
|
newEvents.incl(Event.Read)
|
2020-01-08 17:06:56 +00:00
|
|
|
if not(isNil(adata.writer.function)):
|
2020-01-08 01:06:27 +00:00
|
|
|
newEvents.incl(Event.Write)
|
2018-05-16 08:22:34 +00:00
|
|
|
do:
|
|
|
|
raise newException(ValueError, "File descriptor not registered.")
|
2018-08-24 12:20:08 +00:00
|
|
|
loop.selector.updateHandle(int(fd), newEvents)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
proc removeReader*(fd: AsyncFD) =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Stop watching the file descriptor ``fd`` for read availability.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2018-05-16 08:22:34 +00:00
|
|
|
var newEvents: set[Event]
|
2018-08-24 12:20:08 +00:00
|
|
|
withData(loop.selector, int(fd), adata) do:
|
2018-06-08 05:42:48 +00:00
|
|
|
# We need to clear `reader` data, because `selectors` don't do it
|
2020-09-15 07:55:43 +00:00
|
|
|
adata.reader = default(AsyncCallback)
|
2018-08-24 12:20:08 +00:00
|
|
|
# adata.rdata = CompletionData()
|
2020-01-08 17:06:56 +00:00
|
|
|
if not(isNil(adata.writer.function)):
|
2020-01-08 01:06:27 +00:00
|
|
|
newEvents.incl(Event.Write)
|
2018-05-16 08:22:34 +00:00
|
|
|
do:
|
|
|
|
raise newException(ValueError, "File descriptor not registered.")
|
2018-08-24 12:20:08 +00:00
|
|
|
loop.selector.updateHandle(int(fd), newEvents)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
proc addWriter*(fd: AsyncFD, cb: CallbackFunc, udata: pointer = nil) =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Start watching the file descriptor ``fd`` for write availability and then
|
|
|
|
## call the callback ``cb`` with specified argument ``udata``.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2018-05-16 08:22:34 +00:00
|
|
|
var newEvents = {Event.Write}
|
2018-08-24 12:20:08 +00:00
|
|
|
withData(loop.selector, int(fd), adata) do:
|
2018-05-16 08:22:34 +00:00
|
|
|
let acb = AsyncCallback(function: cb, udata: addr adata.wdata)
|
|
|
|
adata.writer = acb
|
2018-06-08 05:42:48 +00:00
|
|
|
adata.wdata = CompletionData(fd: fd, udata: udata)
|
2018-05-16 08:22:34 +00:00
|
|
|
newEvents.incl(Event.Write)
|
2020-01-08 17:06:56 +00:00
|
|
|
if not(isNil(adata.reader.function)):
|
2020-01-08 01:06:27 +00:00
|
|
|
newEvents.incl(Event.Read)
|
2018-05-16 08:22:34 +00:00
|
|
|
do:
|
|
|
|
raise newException(ValueError, "File descriptor not registered.")
|
2018-08-24 12:20:08 +00:00
|
|
|
loop.selector.updateHandle(int(fd), newEvents)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
proc removeWriter*(fd: AsyncFD) =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Stop watching the file descriptor ``fd`` for write availability.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2018-05-16 08:22:34 +00:00
|
|
|
var newEvents: set[Event]
|
2018-08-24 12:20:08 +00:00
|
|
|
withData(loop.selector, int(fd), adata) do:
|
2018-06-08 05:42:48 +00:00
|
|
|
# We need to clear `writer` data, because `selectors` don't do it
|
2020-09-15 07:55:43 +00:00
|
|
|
adata.writer = default(AsyncCallback)
|
2018-08-24 12:20:08 +00:00
|
|
|
# adata.wdata = CompletionData()
|
2020-01-08 17:06:56 +00:00
|
|
|
if not(isNil(adata.reader.function)):
|
2020-01-08 01:06:27 +00:00
|
|
|
newEvents.incl(Event.Read)
|
2018-05-16 08:22:34 +00:00
|
|
|
do:
|
|
|
|
raise newException(ValueError, "File descriptor not registered.")
|
2018-08-24 12:20:08 +00:00
|
|
|
loop.selector.updateHandle(int(fd), newEvents)
|
|
|
|
|
|
|
|
proc closeSocket*(fd: AsyncFD, aftercb: CallbackFunc = nil) =
|
|
|
|
## Close asynchronous socket.
|
|
|
|
##
|
|
|
|
## Please note, that socket is not closed immediately. To avoid bugs with
|
|
|
|
## closing socket, while operation pending, socket will be closed as
|
|
|
|
## soon as all pending operations will be notified.
|
|
|
|
## You can execute ``aftercb`` before actual socket close operation.
|
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
|
|
|
|
proc continuation(udata: pointer) =
|
2020-06-05 16:11:51 +00:00
|
|
|
if SocketHandle(fd) in loop.selector:
|
|
|
|
unregister(fd)
|
|
|
|
close(SocketHandle(fd))
|
2018-10-02 10:50:14 +00:00
|
|
|
if not isNil(aftercb):
|
|
|
|
aftercb(nil)
|
2018-08-24 12:20:08 +00:00
|
|
|
|
|
|
|
withData(loop.selector, int(fd), adata) do:
|
2018-10-02 10:50:14 +00:00
|
|
|
# We are scheduling reader and writer callbacks to be called
|
|
|
|
# explicitly, so they can get an error and continue work.
|
2020-09-15 07:55:43 +00:00
|
|
|
# Callbacks marked as deleted so we don't need to get REAL notifications
|
|
|
|
# from system queue for this reader and writer.
|
|
|
|
|
2020-01-08 17:06:56 +00:00
|
|
|
if not(isNil(adata.reader.function)):
|
2020-09-15 07:55:43 +00:00
|
|
|
loop.callbacks.addLast(adata.reader)
|
|
|
|
adata.reader = default(AsyncCallback)
|
|
|
|
|
2020-01-08 17:06:56 +00:00
|
|
|
if not(isNil(adata.writer.function)):
|
2020-09-15 07:55:43 +00:00
|
|
|
loop.callbacks.addLast(adata.writer)
|
|
|
|
adata.writer = default(AsyncCallback)
|
2018-08-24 12:20:08 +00:00
|
|
|
|
2018-10-02 10:50:14 +00:00
|
|
|
# We can't unregister file descriptor from system queue here, because
|
|
|
|
# in such case processing queue will stuck on poll() call, because there
|
|
|
|
# can be no file descriptors registered in system queue.
|
|
|
|
var acb = AsyncCallback(function: continuation)
|
|
|
|
loop.callbacks.addLast(acb)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2019-07-15 09:59:42 +00:00
|
|
|
proc closeHandle*(fd: AsyncFD, aftercb: CallbackFunc = nil) {.inline.} =
|
|
|
|
## Close asynchronous file/pipe handle.
|
|
|
|
##
|
|
|
|
## Please note, that socket is not closed immediately. To avoid bugs with
|
|
|
|
## closing socket, while operation pending, socket will be closed as
|
|
|
|
## soon as all pending operations will be notified.
|
|
|
|
## You can execute ``aftercb`` before actual socket close operation.
|
|
|
|
closeSocket(fd, aftercb)
|
|
|
|
|
2018-05-25 01:05:13 +00:00
|
|
|
when ioselSupportedPlatform:
|
|
|
|
proc addSignal*(signal: int, cb: CallbackFunc,
|
|
|
|
udata: pointer = nil): int =
|
|
|
|
## Start watching signal ``signal``, and when signal appears, call the
|
2018-05-27 05:49:47 +00:00
|
|
|
## callback ``cb`` with specified argument ``udata``. Returns signal
|
|
|
|
## identifier code, which can be used to remove signal callback
|
|
|
|
## via ``removeSignal``.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2018-05-25 01:05:13 +00:00
|
|
|
var data: SelectorData
|
2018-08-24 12:20:08 +00:00
|
|
|
result = loop.selector.registerSignal(signal, data)
|
|
|
|
withData(loop.selector, result, adata) do:
|
2018-05-25 01:05:13 +00:00
|
|
|
adata.reader = AsyncCallback(function: cb, udata: addr adata.rdata)
|
|
|
|
adata.rdata.fd = AsyncFD(result)
|
|
|
|
adata.rdata.udata = udata
|
|
|
|
do:
|
|
|
|
raise newException(ValueError, "File descriptor not registered.")
|
|
|
|
|
|
|
|
proc removeSignal*(sigfd: int) =
|
|
|
|
## Remove watching signal ``signal``.
|
2018-08-24 12:20:08 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
loop.selector.unregister(sigfd)
|
2018-05-25 01:05:13 +00:00
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
proc poll*() =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Perform single asynchronous step.
|
2018-05-16 08:22:34 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
2019-03-24 16:57:36 +00:00
|
|
|
var curTime = Moment.now()
|
2018-05-16 08:22:34 +00:00
|
|
|
var curTimeout = 0
|
|
|
|
|
|
|
|
when ioselSupportedPlatform:
|
|
|
|
let customSet = {Event.Timer, Event.Signal, Event.Process,
|
|
|
|
Event.Vnode}
|
|
|
|
|
2018-05-25 20:00:32 +00:00
|
|
|
# Moving expired timers to `loop.callbacks` and calculate timeout.
|
2018-06-02 23:03:48 +00:00
|
|
|
loop.processTimersGetTimeout(curTimeout)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-05-25 20:00:32 +00:00
|
|
|
# Processing IO descriptors and all hardware events.
|
2018-06-02 23:06:58 +00:00
|
|
|
var count = loop.selector.selectInto(curTimeout, loop.keys)
|
2018-05-16 08:22:34 +00:00
|
|
|
for i in 0..<count:
|
|
|
|
let fd = loop.keys[i].fd
|
|
|
|
let events = loop.keys[i].events
|
|
|
|
|
2018-08-24 12:20:08 +00:00
|
|
|
withData(loop.selector, fd, adata) do:
|
|
|
|
if Event.Read in events or events == {Event.Error}:
|
2020-09-15 07:55:43 +00:00
|
|
|
if not isNil(adata.reader.function):
|
2018-10-02 10:50:14 +00:00
|
|
|
loop.callbacks.addLast(adata.reader)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-08-24 12:20:08 +00:00
|
|
|
if Event.Write in events or events == {Event.Error}:
|
2020-09-15 07:55:43 +00:00
|
|
|
if not isNil(adata.writer.function):
|
2018-10-02 10:50:14 +00:00
|
|
|
loop.callbacks.addLast(adata.writer)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-08-24 12:20:08 +00:00
|
|
|
if Event.User in events:
|
2020-09-15 07:55:43 +00:00
|
|
|
if not isNil(adata.reader.function):
|
2018-10-02 10:50:14 +00:00
|
|
|
loop.callbacks.addLast(adata.reader)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-08-24 12:20:08 +00:00
|
|
|
when ioselSupportedPlatform:
|
|
|
|
if customSet * events != {}:
|
2020-09-15 07:55:43 +00:00
|
|
|
if not isNil(adata.reader.function):
|
2018-10-02 10:50:14 +00:00
|
|
|
loop.callbacks.addLast(adata.reader)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
|
|
|
# Moving expired timers to `loop.callbacks`.
|
2018-06-02 23:03:48 +00:00
|
|
|
loop.processTimers()
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2018-06-02 23:03:48 +00:00
|
|
|
# All callbacks which will be added in process, will be processed on next
|
2018-05-16 08:22:34 +00:00
|
|
|
# poll() call.
|
2018-06-02 23:03:48 +00:00
|
|
|
loop.processCallbacks()
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
else:
|
|
|
|
proc initAPI() = discard
|
2019-09-23 17:24:26 +00:00
|
|
|
proc globalInit() = discard
|
2018-05-22 10:43:42 +00:00
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
proc setTimer*(at: Moment, cb: CallbackFunc,
|
|
|
|
udata: pointer = nil): TimerCallback =
|
2020-01-07 05:26:18 +00:00
|
|
|
## Arrange for the callback ``cb`` to be called at the given absolute
|
|
|
|
## timestamp ``at``. You can also pass ``udata`` to callback.
|
|
|
|
let loop = getGlobalDispatcher()
|
2020-01-08 01:06:27 +00:00
|
|
|
result = TimerCallback(finishAt: at,
|
2020-01-27 18:28:44 +00:00
|
|
|
function: AsyncCallback(function: cb, udata: udata))
|
2020-01-08 01:06:27 +00:00
|
|
|
loop.timers.push(result)
|
2020-01-07 05:26:18 +00:00
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
proc clearTimer*(timer: TimerCallback) {.inline.} =
|
2020-09-15 07:55:43 +00:00
|
|
|
timer.function = default(AsyncCallback)
|
2020-01-07 05:26:18 +00:00
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
proc addTimer*(at: Moment, cb: CallbackFunc, udata: pointer = nil) {.
|
|
|
|
inline, deprecated: "Use setTimer/clearTimer instead".} =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Arrange for the callback ``cb`` to be called at the given absolute
|
|
|
|
## timestamp ``at``. You can also pass ``udata`` to callback.
|
2020-01-27 18:28:44 +00:00
|
|
|
discard setTimer(at, cb, udata)
|
2018-05-16 08:22:34 +00:00
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
proc addTimer*(at: int64, cb: CallbackFunc, udata: pointer = nil) {.
|
|
|
|
inline, deprecated: "Use addTimer(Duration, cb, udata)".} =
|
2020-01-27 18:28:44 +00:00
|
|
|
discard setTimer(Moment.init(at, Millisecond), cb, udata)
|
2019-03-24 16:57:36 +00:00
|
|
|
|
2019-03-24 18:59:51 +00:00
|
|
|
proc addTimer*(at: uint64, cb: CallbackFunc, udata: pointer = nil) {.
|
|
|
|
inline, deprecated: "Use addTimer(Duration, cb, udata)".} =
|
2020-01-27 18:28:44 +00:00
|
|
|
discard setTimer(Moment.init(int64(at), Millisecond), cb, udata)
|
2019-03-24 18:59:51 +00:00
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
proc removeTimer*(at: Moment, cb: CallbackFunc, udata: pointer = nil) =
|
2018-05-27 05:49:47 +00:00
|
|
|
## Remove timer callback ``cb`` with absolute timestamp ``at`` from waiting
|
|
|
|
## queue.
|
2018-05-16 08:22:34 +00:00
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
var list = cast[seq[TimerCallback]](loop.timers)
|
|
|
|
var index = -1
|
|
|
|
for i in 0..<len(list):
|
|
|
|
if list[i].finishAt == at and list[i].function.function == cb and
|
|
|
|
list[i].function.udata == udata:
|
|
|
|
index = i
|
|
|
|
break
|
|
|
|
if index != -1:
|
|
|
|
loop.timers.del(index)
|
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
proc removeTimer*(at: int64, cb: CallbackFunc, udata: pointer = nil) {.
|
|
|
|
inline, deprecated: "Use removeTimer(Duration, cb, udata)".} =
|
|
|
|
removeTimer(Moment.init(at, Millisecond), cb, udata)
|
|
|
|
|
2019-03-24 18:59:51 +00:00
|
|
|
proc removeTimer*(at: uint64, cb: CallbackFunc, udata: pointer = nil) {.
|
|
|
|
inline, deprecated: "Use removeTimer(Duration, cb, udata)".} =
|
|
|
|
removeTimer(Moment.init(int64(at), Millisecond), cb, udata)
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
include asyncfutures2
|
|
|
|
|
2019-04-15 01:27:12 +00:00
|
|
|
proc sleepAsync*(duration: Duration): Future[void] =
|
2018-05-16 08:22:34 +00:00
|
|
|
## Suspends the execution of the current async procedure for the next
|
2019-04-15 01:27:12 +00:00
|
|
|
## ``duration`` time.
|
2020-08-15 22:45:41 +00:00
|
|
|
var retFuture = newFuture[void]("chronos.sleepAsync(Duration)")
|
2019-06-20 20:30:41 +00:00
|
|
|
let moment = Moment.fromNow(duration)
|
2020-01-08 01:06:27 +00:00
|
|
|
var timer: TimerCallback
|
2019-06-20 20:30:41 +00:00
|
|
|
|
|
|
|
proc completion(data: pointer) {.gcsafe.} =
|
2020-01-27 18:28:44 +00:00
|
|
|
retFuture.complete()
|
2019-06-20 20:30:41 +00:00
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
proc cancellation(udata: pointer) {.gcsafe.} =
|
|
|
|
clearTimer(timer)
|
2019-06-20 20:30:41 +00:00
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
retFuture.cancelCallback = cancellation
|
2020-01-07 05:26:18 +00:00
|
|
|
timer = setTimer(moment, completion, cast[pointer](retFuture))
|
2018-05-16 08:22:34 +00:00
|
|
|
return retFuture
|
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
proc sleepAsync*(ms: int): Future[void] {.
|
|
|
|
inline, deprecated: "Use sleepAsync(Duration)".} =
|
|
|
|
result = sleepAsync(ms.milliseconds())
|
|
|
|
|
|
|
|
proc withTimeout*[T](fut: Future[T], timeout: Duration): Future[bool] =
|
2018-05-16 08:22:34 +00:00
|
|
|
## Returns a future which will complete once ``fut`` completes or after
|
|
|
|
## ``timeout`` milliseconds has elapsed.
|
|
|
|
##
|
|
|
|
## If ``fut`` completes first the returned future will hold true,
|
|
|
|
## otherwise, if ``timeout`` milliseconds has elapsed first, the returned
|
|
|
|
## future will hold false.
|
2019-06-20 20:30:41 +00:00
|
|
|
var retFuture = newFuture[bool]("chronos.`withTimeout`")
|
|
|
|
var moment: Moment
|
2020-01-08 01:06:27 +00:00
|
|
|
var timer: TimerCallback
|
2019-06-20 20:30:41 +00:00
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
proc continuation(udata: pointer) {.gcsafe.} =
|
2019-06-20 20:30:41 +00:00
|
|
|
if not(retFuture.finished()):
|
2020-01-27 18:28:44 +00:00
|
|
|
if not(fut.finished()):
|
2019-06-20 20:30:41 +00:00
|
|
|
# Timer exceeded first.
|
2020-01-09 01:32:53 +00:00
|
|
|
fut.removeCallback(continuation)
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.cancel()
|
2018-05-16 08:22:34 +00:00
|
|
|
retFuture.complete(false)
|
|
|
|
else:
|
2019-06-20 20:30:41 +00:00
|
|
|
# Future `fut` completed/failed/cancelled first.
|
2020-01-08 01:06:27 +00:00
|
|
|
if not isNil(timer):
|
2020-01-07 05:26:18 +00:00
|
|
|
clearTimer(timer)
|
2019-06-20 20:30:41 +00:00
|
|
|
retFuture.complete(true)
|
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
proc cancellation(udata: pointer) {.gcsafe.} =
|
|
|
|
if not isNil(timer):
|
|
|
|
clearTimer(timer)
|
|
|
|
if not(fut.finished()):
|
|
|
|
fut.removeCallback(continuation)
|
|
|
|
fut.cancel()
|
2019-06-20 20:30:41 +00:00
|
|
|
|
|
|
|
if fut.finished():
|
|
|
|
retFuture.complete(true)
|
|
|
|
else:
|
|
|
|
if timeout.isZero():
|
|
|
|
retFuture.complete(false)
|
|
|
|
elif timeout.isInfinite():
|
2020-01-27 18:28:44 +00:00
|
|
|
retFuture.cancelCallback = cancellation
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.addCallback(continuation)
|
|
|
|
else:
|
|
|
|
moment = Moment.fromNow(timeout)
|
2020-01-27 18:28:44 +00:00
|
|
|
retFuture.cancelCallback = cancellation
|
2020-01-07 05:26:18 +00:00
|
|
|
timer = setTimer(moment, continuation, nil)
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.addCallback(continuation)
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
return retFuture
|
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
proc withTimeout*[T](fut: Future[T], timeout: int): Future[bool] {.
|
|
|
|
inline, deprecated: "Use withTimeout(Future[T], Duration)".} =
|
|
|
|
result = withTimeout(fut, timeout.milliseconds())
|
|
|
|
|
|
|
|
proc wait*[T](fut: Future[T], timeout = InfiniteDuration): Future[T] =
|
2018-07-20 08:58:01 +00:00
|
|
|
## Returns a future which will complete once future ``fut`` completes
|
|
|
|
## or if timeout of ``timeout`` milliseconds has been expired.
|
2018-08-24 12:20:08 +00:00
|
|
|
##
|
2018-07-20 09:11:38 +00:00
|
|
|
## If ``timeout`` is ``-1``, then statement ``await wait(fut)`` is
|
|
|
|
## equal to ``await fut``.
|
2019-06-20 20:30:41 +00:00
|
|
|
##
|
|
|
|
## TODO: In case when ``fut`` got cancelled, what result Future[T]
|
|
|
|
## should return, because it can't be cancelled too.
|
|
|
|
var retFuture = newFuture[T]("chronos.wait()")
|
|
|
|
var moment: Moment
|
2020-01-08 01:06:27 +00:00
|
|
|
var timer: TimerCallback
|
2019-06-20 20:30:41 +00:00
|
|
|
|
2018-07-20 08:58:01 +00:00
|
|
|
proc continuation(udata: pointer) {.gcsafe.} =
|
2019-06-20 20:30:41 +00:00
|
|
|
if not(retFuture.finished()):
|
2020-01-27 18:28:44 +00:00
|
|
|
if not(fut.finished()):
|
2019-06-20 20:30:41 +00:00
|
|
|
# Timer exceeded first.
|
2018-07-20 08:58:01 +00:00
|
|
|
fut.removeCallback(continuation)
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.cancel()
|
2019-05-07 20:11:40 +00:00
|
|
|
retFuture.fail(newException(AsyncTimeoutError, "Timeout exceeded!"))
|
2018-07-20 08:58:01 +00:00
|
|
|
else:
|
2019-06-20 20:30:41 +00:00
|
|
|
# Future `fut` completed/failed/cancelled first.
|
2020-01-08 01:06:27 +00:00
|
|
|
if not isNil(timer):
|
2020-01-07 05:26:18 +00:00
|
|
|
clearTimer(timer)
|
2019-06-20 20:30:41 +00:00
|
|
|
|
|
|
|
if fut.failed():
|
|
|
|
retFuture.fail(fut.error)
|
2019-03-27 22:56:17 +00:00
|
|
|
else:
|
2019-06-20 20:30:41 +00:00
|
|
|
when T is void:
|
|
|
|
retFuture.complete()
|
|
|
|
else:
|
|
|
|
retFuture.complete(fut.read())
|
|
|
|
|
2020-01-27 18:28:44 +00:00
|
|
|
proc cancellation(udata: pointer) {.gcsafe.} =
|
|
|
|
if not isNil(timer):
|
|
|
|
clearTimer(timer)
|
|
|
|
if not(fut.finished()):
|
|
|
|
fut.removeCallback(continuation)
|
|
|
|
fut.cancel()
|
2019-06-20 20:30:41 +00:00
|
|
|
|
|
|
|
if fut.finished():
|
|
|
|
if fut.failed():
|
|
|
|
retFuture.fail(fut.error)
|
2018-07-20 08:58:01 +00:00
|
|
|
else:
|
2019-06-20 20:30:41 +00:00
|
|
|
when T is void:
|
|
|
|
retFuture.complete()
|
|
|
|
else:
|
|
|
|
retFuture.complete(fut.read())
|
2018-07-20 08:58:01 +00:00
|
|
|
else:
|
2019-06-20 20:30:41 +00:00
|
|
|
if timeout.isZero():
|
|
|
|
retFuture.fail(newException(AsyncTimeoutError, "Timeout exceeded!"))
|
|
|
|
elif timeout.isInfinite():
|
2020-01-27 18:28:44 +00:00
|
|
|
retFuture.cancelCallback = cancellation
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.addCallback(continuation)
|
|
|
|
else:
|
|
|
|
moment = Moment.fromNow(timeout)
|
2020-01-27 18:28:44 +00:00
|
|
|
retFuture.cancelCallback = cancellation
|
2020-01-07 05:26:18 +00:00
|
|
|
timer = setTimer(moment, continuation, nil)
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.addCallback(continuation)
|
|
|
|
|
2018-07-20 08:58:01 +00:00
|
|
|
return retFuture
|
|
|
|
|
2019-03-24 16:57:36 +00:00
|
|
|
proc wait*[T](fut: Future[T], timeout = -1): Future[T] {.
|
|
|
|
inline, deprecated: "Use wait(Future[T], Duration)".} =
|
|
|
|
if timeout == -1:
|
|
|
|
wait(fut, InfiniteDuration)
|
|
|
|
elif timeout == 0:
|
|
|
|
wait(fut, ZeroDuration)
|
|
|
|
else:
|
|
|
|
wait(fut, timeout.milliseconds())
|
|
|
|
|
2018-05-16 08:22:34 +00:00
|
|
|
include asyncmacro2
|
|
|
|
|
2019-09-10 17:19:49 +00:00
|
|
|
proc callSoon*(cbproc: CallbackFunc, data: pointer = nil) =
|
2018-05-16 08:22:34 +00:00
|
|
|
## Schedule `cbproc` to be called as soon as possible.
|
|
|
|
## The callback is called when control returns to the event loop.
|
2019-04-04 09:34:23 +00:00
|
|
|
doAssert(not isNil(cbproc))
|
2018-05-16 08:22:34 +00:00
|
|
|
let acb = AsyncCallback(function: cbproc, udata: data)
|
|
|
|
getGlobalDispatcher().callbacks.addLast(acb)
|
|
|
|
|
|
|
|
proc runForever*() =
|
|
|
|
## Begins a never ending global dispatcher poll loop.
|
|
|
|
while true:
|
|
|
|
poll()
|
|
|
|
|
|
|
|
proc waitFor*[T](fut: Future[T]): T =
|
|
|
|
## **Blocks** the current thread until the specified future completes.
|
2019-06-20 20:30:41 +00:00
|
|
|
while not(fut.finished()):
|
2018-05-16 08:22:34 +00:00
|
|
|
poll()
|
|
|
|
|
2019-06-20 20:30:41 +00:00
|
|
|
fut.read()
|
2018-05-22 10:43:42 +00:00
|
|
|
|
2019-04-04 09:34:23 +00:00
|
|
|
proc addTracker*[T](id: string, tracker: T) =
|
|
|
|
## Add new ``tracker`` object to current thread dispatcher with identifier
|
|
|
|
## ``id``.
|
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
loop.trackers[id] = tracker
|
|
|
|
|
|
|
|
proc getTracker*(id: string): TrackerBase =
|
|
|
|
## Get ``tracker`` from current thread dispatcher using identifier ``id``.
|
|
|
|
let loop = getGlobalDispatcher()
|
|
|
|
result = loop.trackers.getOrDefault(id, nil)
|
2019-09-23 17:24:26 +00:00
|
|
|
|
2020-07-08 16:48:01 +00:00
|
|
|
when defined(chronosFutureTracking):
|
|
|
|
iterator pendingFutures*(): FutureBase =
|
|
|
|
## Iterates over the list of pending Futures (Future[T] objects which not
|
|
|
|
## yet completed, cancelled or failed).
|
|
|
|
var slider = futureList.head
|
|
|
|
while not(isNil(slider)):
|
|
|
|
yield slider
|
|
|
|
slider = slider.next
|
|
|
|
|
|
|
|
proc pendingFuturesCount*(): int =
|
|
|
|
## Returns number of pending Futures (Future[T] objects which not yet
|
|
|
|
## completed, cancelled or failed).
|
|
|
|
futureList.count
|
2020-07-06 06:33:13 +00:00
|
|
|
|
2019-09-23 17:24:26 +00:00
|
|
|
# Perform global per-module initialization.
|
|
|
|
globalInit()
|