diff options
Diffstat (limited to 'lib/pure/ioselectors.nim')
| -rw-r--r-- | lib/pure/ioselectors.nim | 1764 |
1 files changed, 0 insertions, 1764 deletions
diff --git a/lib/pure/ioselectors.nim b/lib/pure/ioselectors.nim deleted file mode 100644 index 034b182ab..000000000 --- a/lib/pure/ioselectors.nim +++ /dev/null @@ -1,1764 +0,0 @@ -# -# -# Nim's Runtime Library -# (c) Copyright 2016 Eugene Kabanov -# -# See the file "copying.txt", included in this -# distribution, for details about the copyright. -# - -## This module allows high-level and efficient I/O multiplexing. -## -## Supported OS primitives: ``epoll``, ``kqueue``, ``poll`` and -## Windows ``select``. -## -## To use threadsafe version of this module, it needs to be compiled -## with both ``-d:threadsafe`` and ``--threads:on`` options. -## -## Supported features: files, sockets, pipes, timers, processes, signals -## and user events. -## -## Fully supported OS: MacOSX, FreeBSD, OpenBSD, NetBSD, Linux. -## -## Partially supported OS: Windows (only sockets and user events), -## Solaris (files, sockets, handles and user events). -## -## TODO: ``/dev/poll``, ``event ports`` and filesystem events. - -import os - -const hasThreadSupport = compileOption("threads") and defined(threadsafe) - -const supportedPlatform = defined(macosx) or defined(freebsd) or - defined(netbsd) or defined(openbsd) or - defined(linux) - -const bsdPlatform = defined(macosx) or defined(freebsd) or - defined(netbsd) or defined(openbsd) - -when defined(linux): - import posix, times -elif bsdPlatform: - import posix, kqueue, times -elif defined(windows): - import winlean -else: - import posix - -when defined(nimdoc): - type - Selector*[T] = ref object - ## An object which holds descriptors to be checked for read/write status - - Event* {.pure.} = enum - ## An enum which hold event types - Read, ## Descriptor is available for read - Write, ## Descriptor is available for write - Timer, ## Timer descriptor is completed - Signal, ## Signal is raised - Process, ## Process is finished - Vnode, ## Currently not supported - User, ## User event is raised - Error ## Error happens while waiting, for descriptor - - ReadyKey*[T] = object - ## An object which holds result for descriptor - fd* : int ## file/socket descriptor - events*: set[Event] ## set of events - data*: T ## application-defined data - - SelectEvent* = object - ## An object which holds user defined event - - proc newSelector*[T](): Selector[T] = - ## Creates a new selector - - proc close*[T](s: Selector[T]) = - ## Closes selector - - proc registerHandle*[T](s: Selector[T], fd: SocketHandle, events: set[Event], - data: T) = - ## Registers file/socket descriptor ``fd`` to selector ``s`` - ## with events set in ``events``. The ``data`` is application-defined - ## data, which to be passed when event happens. - - proc updateHandle*[T](s: Selector[T], fd: SocketHandle, events: set[Event]) = - ## Update file/socket descriptor ``fd``, registered in selector - ## ``s`` with new events set ``event``. - - proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, - data: T): int {.discardable.} = - ## Registers timer notification with ``timeout`` in milliseconds - ## to selector ``s``. - ## If ``oneshot`` is ``true`` timer will be notified only once. - ## Set ``oneshot`` to ``false`` if your want periodic notifications. - ## The ``data`` is application-defined data, which to be passed, when - ## time limit expired. - - proc registerSignal*[T](s: Selector[T], signal: int, - data: T): int {.discardable.} = - ## Registers Unix signal notification with ``signal`` to selector - ## ``s``. The ``data`` is application-defined data, which to be - ## passed, when signal raises. - ## - ## This function is not supported for ``Windows``. - - proc registerProcess*[T](s: Selector[T], pid: int, - data: T): int {.discardable.} = - ## Registers process id (pid) notification when process has - ## exited to selector ``s``. - ## The ``data`` is application-defined data, which to be passed, when - ## process with ``pid`` has exited. - - proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = - ## Registers selector event ``ev`` to selector ``s``. - ## ``data`` application-defined data, which to be passed, when - ## ``ev`` happens. - - proc newEvent*(): SelectEvent = - ## Creates new event ``SelectEvent``. - - proc setEvent*(ev: SelectEvent) = - ## Trigger event ``ev``. - - proc close*(ev: SelectEvent) = - ## Closes selector event ``ev``. - - proc unregister*[T](s: Selector[T], ev: SelectEvent) = - ## Unregisters event ``ev`` from selector ``s``. - - proc unregister*[T](s: Selector[T], fd: int|SocketHandle|cint) = - ## Unregisters file/socket descriptor ``fd`` from selector ``s``. - - proc flush*[T](s: Selector[T]) = - ## Flushes all changes was made to kernel pool/queue. - ## This function is usefull only for BSD and MacOS, because - ## kqueue supports bulk changes to be made. - ## On Linux/Windows and other Posix compatible operation systems, - ## ``flush`` is alias for `discard`. - - proc selectInto*[T](s: Selector[T], timeout: int, - results: var openarray[ReadyKey[T]]): int = - ## Process call waiting for events registered in selector ``s``. - ## The ``timeout`` argument specifies the minimum number of milliseconds - ## the function will be blocked, if no events are not ready. Specifying a - ## timeout of ``-1`` causes function to block indefinitely. - ## All available events will be stored in ``results`` array. - ## - ## Function returns number of triggered events. - - proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = - ## Process call waiting for events registered in selector ``s``. - ## The ``timeout`` argument specifies the minimum number of milliseconds - ## the function will be blocked, if no events are not ready. Specifying a - ## timeout of -1 causes function to block indefinitely. - ## - ## Function returns sequence of triggered events. - - template isEmpty*[T](s: Selector[T]): bool = - ## Returns ``true``, if there no registered events or descriptors - ## in selector. - - template withData*[T](s: Selector[T], fd: SocketHandle, value, - body: untyped) = - ## retrieves the application-data assigned with descriptor ``fd`` - ## to ``value``. This ``value`` can be modified in the scope of - ## the ``withData`` call. - ## - ## .. code-block:: nim - ## - ## s.withData(fd, value) do: - ## # block is executed only if ``fd`` registered in selector ``s`` - ## value.uid = 1000 - ## - - template withData*[T](s: Selector[T], fd: SocketHandle, value, - body1, body2: untyped) = - ## retrieves the application-data assigned with descriptor ``fd`` - ## to ``value``. This ``value`` can be modified in the scope of - ## the ``withData`` call. - ## - ## .. code-block:: nim - ## - ## s.withData(fd, value) do: - ## # block is executed only if ``fd`` registered in selector ``s``. - ## value.uid = 1000 - ## do: - ## # block is executed if ``fd`` not registered in selector ``s``. - ## raise - ## - -else: - when defined(macosx) or defined(freebsd): - when defined(macosx): - const maxDescriptors = 29 # KERN_MAXFILESPERPROC (MacOS) - else: - const maxDescriptors = 27 # KERN_MAXFILESPERPROC (FreeBSD) - proc sysctl(name: ptr cint, namelen: cuint, oldp: pointer, oldplen: ptr int, - newp: pointer, newplen: int): cint - {.importc: "sysctl",header: """#include <sys/types.h> - #include <sys/sysctl.h>"""} - elif defined(netbsd) or defined(openbsd): - # OpenBSD and NetBSD don't have KERN_MAXFILESPERPROC, so we are using - # KERN_MAXFILES, because KERN_MAXFILES is always bigger, - # than KERN_MAXFILESPERPROC - const maxDescriptors = 7 # KERN_MAXFILES - proc sysctl(name: ptr cint, namelen: cuint, oldp: pointer, oldplen: ptr int, - newp: pointer, newplen: int): cint - {.importc: "sysctl",header: """#include <sys/param.h> - #include <sys/sysctl.h>"""} - elif defined(linux) or defined(solaris): - proc ulimit(cmd: cint): clong - {.importc: "ulimit", header: "<ulimit.h>", varargs.} - elif defined(windows): - discard - else: - var - RLIMIT_NOFILE {.importc: "RLIMIT_NOFILE", - header: "<sys/resource.h>".}: cint - type - rlimit {.importc: "struct rlimit", - header: "<sys/resource.h>", pure, final.} = object - rlim_cur: int - rlim_max: int - proc getrlimit(resource: cint, rlp: var rlimit): cint - {.importc: "getrlimit",header: "<sys/resource.h>".} - - proc getMaxFds*(): int = - when defined(macosx) or defined(freebsd) or defined(netbsd) or - defined(openbsd): - var count = cint(0) - var size = sizeof(count) - var namearr = [cint(1), cint(maxDescriptors)] - - if sysctl(addr namearr[0], 2, cast[pointer](addr count), addr size, - nil, 0) != 0: - raiseOsError(osLastError()) - result = count - elif defined(linux) or defined(solaris): - result = int(ulimit(4, 0)) - elif defined(windows): - result = FD_SETSIZE - else: - var a = rlimit() - if getrlimit(RLIMIT_NOFILE, a) != 0: - raiseOsError(osLastError()) - result = a.rlim_max - - when hasThreadSupport: - import locks - - type - Event* {.pure.} = enum - Read, Write, Timer, Signal, Process, Vnode, User, Error, - flagHandle, flagTimer, flagSignal, flagProcess, flagVnode, flagUser, - flagOneshot - - ReadyKey*[T] = object - fd* : int - events*: set[Event] - data*: T - - SelectorKey[T] = object - ident : int - flags : set[Event] - param : int - key : ReadyKey[T] - - when not defined(windows): - type - SharedArrayHolder[T] = object - part: array[16, T] - SharedArray {.unchecked.}[T] = array[0..100_000_000, T] - - proc allocSharedArray[T](nsize: int): ptr SharedArray[T] = - let holder = cast[ptr SharedArrayHolder[T]]( - allocShared0(sizeof(T) * nsize) - ) - result = cast[ptr SharedArray[T]](addr(holder.part[0])) - - proc deallocSharedArray[T](sa: ptr SharedArray[T]) = - deallocShared(cast[pointer](sa)) - - template setNonBlocking(fd) = - var x: int = fcntl(fd, F_GETFL, 0) - if x == -1: raiseOSError(osLastError()) - else: - var mode = x or O_NONBLOCK - if fcntl(fd, F_SETFL, mode) == -1: - raiseOSError(osLastError()) - - template setKey(s, f1, f2, e, p, d) = - s.fds[f1].ident = f1 - s.fds[f1].flags = e - s.fds[f1].param = p - s.fds[f1].key.fd = f2 - s.fds[f1].key.data = d - - template clearKey(s, f) = - s.fds[f].ident = 0 - s.fds[f].flags = {} - - template checkMaxFd(s, fd) = - if fd.uint >= s.maxFD: - raise newException(ValueError, "Maximum file descriptors exceeded") - - when supportedPlatform: - template blockSignals(newmask: var Sigset, oldmask: var Sigset) = - when hasThreadSupport: - if posix.pthread_sigmask(SIG_BLOCK, newmask, oldmask) == -1: - raiseOSError(osLastError()) - else: - if posix.sigprocmask(SIG_BLOCK, newmask, oldmask) == -1: - raiseOSError(osLastError()) - - template unblockSignals(newmask: var Sigset, oldmask: var Sigset) = - when hasThreadSupport: - if posix.pthread_sigmask(SIG_UNBLOCK, newmask, oldmask) == -1: - raiseOSError(osLastError()) - else: - if posix.sigprocmask(SIG_UNBLOCK, newmask, oldmask) == -1: - raiseOSError(osLastError()) - # - # BSD kqueue - # - # I have tried to adopt kqueue's EVFILT_USER filter for user-events, but it - # looks not very usable, because of 2 cases: - # 1) EVFILT_USER does not supported by OpenBSD and NetBSD - # 2) You can't have one event, which you can use with many kqueue handles. - # So decision was made in favor of the pipes - # - when bsdPlatform: - const - # Maximum number of cached changes - MAX_KQUEUE_CHANGE_EVENTS = 64 - # Maximum number of events that can be returned - MAX_KQUEUE_RESULT_EVENTS = 64 - - type - SelectorImpl[T] = object - kqFD : cint - maxFD : uint - changesTable: array[MAX_KQUEUE_CHANGE_EVENTS, KEvent] - changesCount: int - fds: ptr SharedArray[SelectorKey[T]] - count: int - when hasThreadSupport: - changesLock: Lock - Selector*[T] = ptr SelectorImpl[T] - - type - SelectEventImpl = object - rfd: cint - wfd: cint - # SelectEvent is declared as `ptr` to be placed in `shared memory`, - # so you can share one SelectEvent handle between threads. - type SelectEvent* = ptr SelectEventImpl - - proc newSelector*[T](): Selector[T] = - var maxFD = getMaxFds() - var kqFD = kqueue() - if kqFD < 0: - raiseOsError(osLastError()) - - result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) - result.kqFD = kqFD - result.maxFD = maxFD.uint - result.fds = allocSharedArray[SelectorKey[T]](maxFD) - when hasThreadSupport: - initLock(result.changesLock) - - proc close*[T](s: Selector[T]) = - if posix.close(s.kqFD) != 0: - raiseOSError(osLastError()) - when hasThreadSupport: - deinitLock(s.changesLock) - deallocSharedArray(s.fds) - deallocShared(cast[pointer](s)) - - when hasThreadSupport: - template withChangeLock[T](s: Selector[T], body: untyped) = - acquire(s.changesLock) - {.locks: [s.changesLock].}: - try: - body - finally: - release(s.changesLock) - else: - template withChangeLock(s, body: untyped) = - body - - template modifyKQueue[T](s: Selector[T], nident: uint, nfilter: cshort, - nflags: cushort, nfflags: cuint, ndata: int, - nudata: pointer) = - mixin withChangeLock - s.withChangeLock(): - s.changesTable[s.changesCount] = KEvent(ident: nident, - filter: nfilter, flags: nflags, - fflags: nfflags, data: ndata, - udata: nudata) - inc(s.changesCount) - if s.changesCount == MAX_KQUEUE_CHANGE_EVENTS: - if kevent(s.kqFD, addr(s.changesTable[0]), cint(s.changesCount), - nil, 0, nil) == -1: - raiseOSError(osLastError()) - s.changesCount = 0 - - proc registerHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event], data: T) = - var fdi = int(fd) - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - setKey(s, fdi, fdi, {Event.flagHandle} + events, 0, data) - if events != {}: - if Event.Read in events: - modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) - inc(s.count) - if Event.Write in events: - modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil) - inc(s.count) - - proc updateHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event]) = - var fdi = int(fd) - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident != 0) - doAssert(Event.flagHandle in s.fds[fdi].flags) - var ne = events + {Event.flagHandle} - var oe = s.fds[fdi].flags - if oe != ne: - if (Event.Read in oe) and (Event.Read notin ne): - modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) - dec(s.count) - if (Event.Write in oe) and (Event.Write notin ne): - modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil) - dec(s.count) - if (Event.Read notin oe) and (Event.Read in ne): - modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) - inc(s.count) - if (Event.Write notin oe) and (Event.Write in ne): - modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil) - inc(s.count) - s.fds[fdi].flags = ne - - proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, - data: T): int {.discardable.} = - var fdi = posix.socket(posix.AF_INET, posix.SOCK_STREAM, - posix.IPPROTO_TCP).int - if fdi == -1: - raiseOsError(osLastError()) - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - var mflags = if oneshot: {Event.flagTimer, Event.flagOneshot} - else: {Event.flagTimer} - var kflags: cushort = if oneshot: EV_ONESHOT or EV_ADD - else: EV_ADD - setKey(s, fdi, fdi, mflags, 0, data) - # EVFILT_TIMER on Open/Net(BSD) has granularity of only milliseconds, - # but MacOS and FreeBSD allow use `0` as `fflags` to use milliseconds - # too - modifyKQueue(s, fdi.uint, EVFILT_TIMER, kflags, 0, cint(timeout), nil) - inc(s.count) - result = fdi - - proc registerSignal*[T](s: Selector[T], signal: int, - data: T): int {.discardable.} = - var fdi = posix.socket(posix.AF_INET, posix.SOCK_STREAM, - posix.IPPROTO_TCP).int - if fdi == -1: - raiseOsError(osLastError()) - - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - setKey(s, fdi, signal, {Event.flagSignal}, signal, data) - # block signal `signal` - var nmask: Sigset - var omask: Sigset - discard sigemptyset(nmask) - discard sigemptyset(omask) - discard sigaddset(nmask, cint(signal)) - blockSignals(nmask, omask) - # to be compatible with linux semantic we need to "eat" signals - posix.signal(cint(signal), SIG_IGN) - modifyKQueue(s, signal.uint, EVFILT_SIGNAL, EV_ADD, 0, 0, - cast[pointer](fdi)) - inc(s.count) - result = fdi - - proc registerProcess*[T](s: Selector[T], pid: int, - data: T): int {.discardable.} = - var fdi = posix.socket(posix.AF_INET, posix.SOCK_STREAM, - posix.IPPROTO_TCP).int - if fdi == -1: - raiseOsError(osLastError()) - - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - var kflags: cushort = EV_ONESHOT or EV_ADD - setKey(s, fdi, pid, {Event.flagProcess, Event.flagOneshot}, pid, data) - modifyKQueue(s, pid.uint, EVFILT_PROC, kflags, NOTE_EXIT, 0, - cast[pointer](fdi)) - inc(s.count) - result = fdi - - proc unregister*[T](s: Selector[T], fd: int|SocketHandle|cint) = - var fdi = int(fd) - if fdi.uint < s.maxFD: - var flags = s.fds[fdi].flags - var filter: cshort = 0 - if s.fds[fdi].ident != 0 and flags != {}: - if Event.flagHandle in flags: - # if events == 0, than descriptor was modified with - # updateHandle(fd, 0), so it was already deleted from kqueue. - if flags != {Event.flagHandle}: - if Event.Read in flags: - modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) - dec(s.count) - if Event.Write in flags: - modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil) - dec(s.count) - elif Event.flagTimer in flags: - filter = EVFILT_TIMER - discard posix.close(cint(s.fds[fdi].key.fd)) - modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) - dec(s.count) - elif Event.flagSignal in flags: - filter = EVFILT_SIGNAL - # unblocking signal - var nmask = Sigset() - var omask = Sigset() - var signal = cint(s.fds[fdi].param) - discard sigaddset(nmask, signal) - unblockSignals(nmask, omask) - posix.signal(signal, SIG_DFL) - discard posix.close(cint(s.fds[fdi].key.fd)) - modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) - dec(s.count) - elif Event.flagProcess in flags: - filter = EVFILT_PROC - discard posix.close(cint(s.fds[fdi].key.fd)) - modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) - dec(s.count) - elif Event.flagUser in flags: - filter = EVFILT_READ - modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) - dec(s.count) - clearKey(s, fdi) - - proc flush*[T](s: Selector[T]) = - s.withChangeLock(): - var tv = Timespec() - if kevent(s.kqFD, addr(s.changesTable[0]), cint(s.changesCount), - nil, 0, addr tv) == -1: - raiseOSError(osLastError()) - s.changesCount = 0 - - template isEmpty*[T](s: Selector[T]): bool = - (s.count == 0) - - proc newEvent*(): SelectEvent = - var fds: array[2, cint] - - if posix.pipe(fds) == -1: - raiseOSError(osLastError()) - - setNonBlocking(fds[0]) - setNonBlocking(fds[1]) - - result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) - result.rfd = fds[0] - result.wfd = fds[1] - - proc setEvent*(ev: SelectEvent) = - var data: int = 1 - if posix.write(ev.wfd, addr data, sizeof(int)) != sizeof(int): - raiseOSError(osLastError()) - - proc close*(ev: SelectEvent) = - discard posix.close(cint(ev.rfd)) - discard posix.close(cint(ev.wfd)) - deallocShared(cast[pointer](ev)) - - proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = - let fdi = ev.rfd.int - doAssert(s.fds[fdi].ident == 0) - setKey(s, fdi, fdi, {Event.flagUser}, 0, data) - modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) - inc(s.count) - - proc unregister*[T](s: Selector[T], ev: SelectEvent) = - let fdi = ev.rfd.int - var flags = s.fds[fdi].flags - if s.fds[fdi].ident != 0 and flags != {}: - modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) - dec(s.count) - clearKey(s, fdi) - - proc selectInto*[T](s: Selector[T], timeout: int, - results: var openarray[ReadyKey[T]]): int = - var - tv: Timespec - resultsTable: array[MAX_KQUEUE_RESULT_EVENTS, KEvent] - ptv: ptr Timespec = addr tv - - if timeout != -1: - if timeout >= 1000: - tv.tv_sec = (timeout div 1_000).Time - tv.tv_nsec = (timeout %% 1_000) * 1_000_000 - else: - tv.tv_sec = 0.Time - tv.tv_nsec = timeout * 1_000_000 - else: - ptv = nil - - var maxResults = MAX_KQUEUE_RESULT_EVENTS - if maxResults > len(results): - maxResults = len(results) - - var count = 0 - s.withChangeLock(): - count = kevent(s.kqFD, - addr(s.changesTable[0]), cint(s.changesCount), - addr(resultsTable[0]), cint(maxResults), ptv) - s.changesCount = 0 - if count >= 0: - var skey: ptr SelectorKey[T] - var i = 0 - var k = 0 - while i < count: - var kevent = addr(resultsTable[i]) - if (kevent.flags and EV_ERROR) == 0: - var events: set[Event] = {} - case kevent.filter - of EVFILT_READ: - skey = addr(s.fds[kevent.ident.int]) - if Event.flagHandle in skey.flags: - events = {Event.Read} - elif Event.flagUser in skey.flags: - var data: int = 0 - if posix.read(kevent.ident.cint, addr data, - sizeof(int)) != sizeof(int): - let err = osLastError() - if err == OSErrorCode(EAGAIN): - # someone already consumed event data - inc(i) - continue - else: - raiseOSError(osLastError()) - events = {Event.User} - else: - events = {Event.Read} - of EVFILT_WRITE: - skey = addr(s.fds[kevent.ident.int]) - events = {Event.Write} - of EVFILT_TIMER: - skey = addr(s.fds[kevent.ident.int]) - if Event.flagOneshot in skey.flags: - if posix.close(skey.ident.cint) == -1: - raiseOSError(osLastError()) - clearKey(s, skey.ident) - # no need to modify kqueue, because EV_ONESHOT is already made - # this for us - dec(s.count) - events = {Event.Timer} - of EVFILT_VNODE: - skey = addr(s.fds[kevent.ident.int]) - events = {Event.Vnode} - of EVFILT_SIGNAL: - skey = addr(s.fds[cast[int](kevent.udata)]) - events = {Event.Signal} - of EVFILT_PROC: - skey = addr(s.fds[cast[int](kevent.udata)]) - if posix.close(skey.ident.cint) == -1: - raiseOSError(osLastError()) - clearKey(s, skey.ident) - # no need to modify kqueue, because EV_ONESHOT is already made - # this for us - dec(s.count) - events = {Event.Process} - else: - raise newException(ValueError, - "Unsupported kqueue filter in queue") - - if (kevent.flags and EV_EOF) != 0: - events = events + {Event.Error} - results[k].fd = skey.key.fd - results[k].events = events - results[k].data = skey.key.data - inc(k) - inc(i) - result = k - else: - result = 0 - let err = osLastError() - if cint(err) != EINTR: - raiseOSError(err) - - proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = - result = newSeq[ReadyKey[T]](MAX_KQUEUE_RESULT_EVENTS) - var count = selectInto(s, timeout, result) - result.setLen(count) - - # - # Linux epoll - # - - elif defined(linux): - const - # Maximum number of events that can be returned - MAX_EPOLL_RESULT_EVENTS = 64 - type - SignalFdInfo* {.importc: "struct signalfd_siginfo", - header: "<sys/signalfd.h>", pure, final.} = object - ssi_signo*: uint32 - ssi_errno*: int32 - ssi_code*: int32 - ssi_pid*: uint32 - ssi_uid*: uint32 - ssi_fd*: int32 - ssi_tid*: uint32 - ssi_band*: uint32 - ssi_overrun*: uint32 - ssi_trapno*: uint32 - ssi_status*: int32 - ssi_int*: int32 - ssi_ptr*: uint64 - ssi_utime*: uint64 - ssi_stime*: uint64 - ssi_addr*: uint64 - pad* {.importc: "__pad".}: array[0..47, uint8] - type - eventFdData {.importc: "eventfd_t", - header: "<sys/eventfd.h>", pure, final.} = uint64 - epoll_data {.importc: "union epoll_data", - header: "<sys/epoll.h>", - pure, final.} = object - u64 {.importc: "u64".}: uint64 - - epoll_event {.importc: "struct epoll_event", - header: "<sys/epoll.h>", pure, final.} = object - events: uint32 # Epoll events - data: epoll_data # User data variable - const - EPOLL_CTL_ADD = 1 # Add a file descriptor to the interface. - EPOLL_CTL_DEL = 2 # Remove a file descriptor from the interface. - EPOLL_CTL_MOD = 3 # Change file descriptor epoll_event structure. - const - EPOLLIN = 0x00000001 - EPOLLOUT = 0x00000004 - EPOLLERR = 0x00000008 - EPOLLHUP = 0x00000010 - EPOLLRDHUP = 0x00002000 - EPOLLONESHOT = 1 shl 30 - - proc epoll_create(size: cint): cint - {.importc: "epoll_create", header: "<sys/epoll.h>".} - proc epoll_ctl(epfd: cint; op: cint; fd: cint; event: ptr epoll_event): cint - {.importc: "epoll_ctl", header: "<sys/epoll.h>".} - proc epoll_wait(epfd: cint; events: ptr epoll_event; maxevents: cint; - timeout: cint): cint - {.importc: "epoll_wait", header: "<sys/epoll.h>".} - proc timerfd_create(clock_id: ClockId, flags: cint): cint - {.cdecl, importc: "timerfd_create", header: "<sys/timerfd.h>".} - proc timerfd_settime(ufd: cint, flags: cint, - utmr: var Itimerspec, otmr: var Itimerspec): cint - {.cdecl, importc: "timerfd_settime", header: "<sys/timerfd.h>".} - proc signalfd(fd: cint, mask: var Sigset, flags: cint): cint - {.cdecl, importc: "signalfd", header: "<sys/signalfd.h>".} - proc eventfd(count: cuint, flags: cint): cint - {.cdecl, importc: "eventfd", header: "<sys/eventfd.h>".} - - type - SelectorImpl[T] = object - epollFD : cint - maxFD : uint - fds: ptr SharedArray[SelectorKey[T]] - count: int - - Selector*[T] = ptr SelectorImpl[T] - - SelectEventImpl = object - efd: cint - - SelectEvent* = ptr SelectEventImpl - - proc newSelector*[T](): Selector[T] = - var maxFD = getMaxFds() - var epollFD = epoll_create(MAX_EPOLL_RESULT_EVENTS) - if epollFD < 0: - raiseOsError(osLastError()) - - result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) - result.epollFD = epollFD - result.maxFD = maxFD.uint - result.fds = allocSharedArray[SelectorKey[T]](maxFD) - - proc close*[T](s: Selector[T]) = - if posix.close(s.epollFD) != 0: - raiseOSError(osLastError()) - deallocSharedArray(s.fds) - deallocShared(cast[pointer](s)) - - proc registerHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event], data: T) = - var fdi = int(fd) - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - setKey(s, fdi, fdi, events + {Event.flagHandle}, 0, data) - if events != {}: - var epv: epoll_event - epv.events = EPOLLRDHUP - epv.data.u64 = fdi.uint - if Event.Read in events: - epv.events = epv.events or EPOLLIN - if Event.Write in events: - epv.events = epv.events or EPOLLOUT - if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - inc(s.count) - - proc updateHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event]) = - var fdi = int(fd) - s.checkMaxFd(fdi) - var oe = s.fds[fdi].flags - doAssert(s.fds[fdi].ident != 0) - doAssert(Event.flagHandle in oe) - var ne = events + {Event.flagHandle} - if oe != ne: - var epv: epoll_event - epv.data.u64 = fdi.uint - epv.events = EPOLLRDHUP - - if Event.Read in events: - epv.events = epv.events or EPOLLIN - if Event.Write in events: - epv.events = epv.events or EPOLLOUT - - if oe == {Event.flagHandle}: - if ne != {Event.flagHandle}: - if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, - addr epv) == -1: - raiseOSError(osLastError()) - inc(s.count) - else: - if ne != {Event.flagHandle}: - if epoll_ctl(s.epollFD, EPOLL_CTL_MOD, fdi.cint, - addr epv) == -1: - raiseOSError(osLastError()) - else: - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, - addr epv) == -1: - raiseOSError(osLastError()) - dec(s.count) - s.fds[fdi].flags = ne - - proc unregister*[T](s: Selector[T], fd: int|SocketHandle|cint) = - var epv: epoll_event - var fdi = int(fd) - if fdi.uint < s.maxFD: - var flags = s.fds[fdi].flags - if s.fds[fdi].ident != 0 and flags != {}: - if Event.flagHandle in flags: - # if events == {flagHandle}, then descriptor was already - # unregistered from epoll with updateHandle() call. - # This check is done to omit EBADF error. - if flags != {Event.flagHandle}: - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, - addr epv) == -1: - raiseOSError(osLastError()) - dec(s.count) - elif Event.flagTimer in flags: - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - discard posix.close(fdi.cint) - dec(s.count) - elif Event.flagSignal in flags: - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - var nmask: Sigset - var omask: Sigset - discard sigemptyset(nmask) - discard sigemptyset(omask) - discard sigaddset(nmask, cint(s.fds[fdi].param)) - unblockSignals(nmask, omask) - discard posix.close(fdi.cint) - dec(s.count) - elif Event.flagProcess in flags: - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - var nmask: Sigset - var omask: Sigset - discard sigemptyset(nmask) - discard sigemptyset(omask) - discard sigaddset(nmask, SIGCHLD) - unblockSignals(nmask, omask) - discard posix.close(fdi.cint) - dec(s.count) - clearKey(s, fdi) - - proc unregister*[T](s: Selector[T], ev: SelectEvent) = - let fdi = int(ev.efd) - if fdi.uint < s.maxFD: - if s.fds[fdi].ident != 0 and (Event.flagUser in s.fds[fdi].flags): - clearKey(s, fdi) - var epv: epoll_event - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - dec(s.count) - - proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, - data: T): int {.discardable.} = - var - new_ts: Itimerspec - old_ts: Itimerspec - var fdi = timerfd_create(CLOCK_MONOTONIC, 0) - if fdi == -1: - raiseOSError(osLastError()) - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - var flags = {Event.flagTimer} - var epv: epoll_event - epv.data.u64 = fdi.uint - epv.events = EPOLLIN or EPOLLRDHUP - setNonBlocking(fdi.cint) - if oneshot: - new_ts.it_interval.tv_sec = 0.Time - new_ts.it_interval.tv_nsec = 0 - new_ts.it_value.tv_sec = (timeout div 1_000).Time - new_ts.it_value.tv_nsec = (timeout %% 1_000) * 1_000_000 - flags = flags + {Event.flagOneshot} - epv.events = epv.events or EPOLLONESHOT - else: - new_ts.it_interval.tv_sec = (timeout div 1000).Time - new_ts.it_interval.tv_nsec = (timeout %% 1_000) * 1_000_000 - new_ts.it_value.tv_sec = new_ts.it_interval.tv_sec - new_ts.it_value.tv_nsec = new_ts.it_interval.tv_nsec - if timerfd_settime(fdi.cint, cint(0), new_ts, old_ts) == -1: - raiseOSError(osLastError()) - if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - setKey(s, fdi, fdi, flags, 0, data) - inc(s.count) - result = fdi - - proc registerSignal*[T](s: Selector[T], signal: int, - data: T): int {.discardable.} = - var - nmask: Sigset - omask: Sigset - - discard sigemptyset(nmask) - discard sigemptyset(omask) - discard sigaddset(nmask, cint(signal)) - blockSignals(nmask, omask) - - var fdi = signalfd(-1, nmask, 0).int - if fdi == -1: - raiseOSError(osLastError()) - - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - setNonBlocking(fdi.cint) - - var epv: epoll_event - epv.data.u64 = fdi.uint - epv.events = EPOLLIN or EPOLLRDHUP - if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - setKey(s, fdi, signal, {Event.flagSignal}, signal, data) - inc(s.count) - result = fdi - - proc registerProcess*[T](s: Selector, pid: int, - data: T): int {.discardable.} = - var - nmask: Sigset - omask: Sigset - - discard sigemptyset(nmask) - discard sigemptyset(omask) - discard sigaddset(nmask, posix.SIGCHLD) - blockSignals(nmask, omask) - - var fdi = signalfd(-1, nmask, 0).int - if fdi == -1: - raiseOSError(osLastError()) - - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - setNonBlocking(fdi.cint) - - var epv: epoll_event - epv.data.u64 = fdi.uint - epv.events = EPOLLIN or EPOLLRDHUP - if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: - raiseOSError(osLastError()) - setKey(s, fdi, pid, {Event.flagProcess}, pid, data) - inc(s.count) - result = fdi - - proc flush*[T](s: Selector[T]) = - discard - - template isEmpty*[T](s: Selector[T]): bool = - (s.count == 0) - - proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = - let fdi = int(ev.efd) - doAssert(s.fds[fdi].ident == 0) - setKey(s, fdi, fdi, {Event.flagUser}, 0, data) - var epv = epoll_event(events: EPOLLIN or EPOLLRDHUP) - epv.data.u64 = ev.efd.uint - if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, ev.efd, addr epv) == -1: - raiseOSError(osLastError()) - inc(s.count) - - proc setEvent*(ev: SelectEvent) = - var data : uint64 = 1 - if posix.write(ev.efd, addr data, sizeof(uint64)) == -1: - raiseOSError(osLastError()) - - proc close*(ev: SelectEvent) = - discard posix.close(ev.efd) - deallocShared(cast[pointer](ev)) - - proc newEvent*(): SelectEvent = - var fdi = eventfd(0, 0) - if fdi == -1: - raiseOSError(osLastError()) - setNonBlocking(fdi) - result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) - result.efd = cint(fdi) - - proc selectInto*[T](s: Selector[T], timeout: int, - results: var openarray[ReadyKey[T]]): int = - var - resultsTable: array[MAX_EPOLL_RESULT_EVENTS, epoll_event] - - var maxResults = MAX_EPOLL_RESULT_EVENTS - if maxResults > len(results): - maxResults = len(results) - - var count = epoll_wait(s.epollFD, addr(resultsTable[0]), maxResults.cint, - timeout.cint) - if count > 0: - var i = 0 - var k = 0 - while i < count: - var events: set[Event] = {} - let fdi = int(resultsTable[i].data.u64) - var skey = addr(s.fds[fdi]) - let pevents = resultsTable[i].events - var flags = s.fds[fdi].flags - - if skey.ident != 0 and flags != {}: - block processItem: - if (pevents and EPOLLERR) != 0 or (pevents and EPOLLHUP) != 0: - events = events + {Event.Error} - if (pevents and EPOLLOUT) != 0: - events = events + {Event.Write} - if (pevents and EPOLLIN) != 0: - if Event.flagHandle in flags: - events = events + {Event.Read} - elif Event.flagTimer in flags: - var data: uint64 = 0 - if posix.read(fdi.cint, addr data, - sizeof(uint64)) != sizeof(uint64): - raiseOSError(osLastError()) - events = events + {Event.Timer} - elif Event.flagSignal in flags: - var data: SignalFdInfo - if posix.read(fdi.cint, addr data, - sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): - raiseOsError(osLastError()) - events = events + {Event.Signal} - elif Event.flagProcess in flags: - var data: SignalFdInfo - if posix.read(fdi.cint, addr data, - sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): - raiseOsError(osLastError()) - if cast[int](data.ssi_pid) == skey.param: - events = events + {Event.Process} - # we want to free resources for this event - flags = flags + {Event.flagOneshot} - else: - break processItem - elif Event.flagUser in flags: - var data: uint = 0 - if posix.read(fdi.cint, addr data, - sizeof(uint)) != sizeof(uint): - let err = osLastError() - if err == OSErrorCode(EAGAIN): - # someone already consumed event data - inc(i) - continue - else: - raiseOSError(err) - events = events + {Event.User} - else: - raise newException(ValueError, - "Unsupported epoll event in queue") - results[k].fd = skey.key.fd - results[k].events = events - results[k].data = skey.key.data - - if Event.flagOneshot in flags: - var epv: epoll_event - try: - if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, - addr epv) == -1: - raiseOSError(osLastError()) - finally: - discard posix.close(fdi.cint) - s.fds[fdi].ident = 0 - s.fds[fdi].flags = {} - dec(s.count) - inc(k) - inc(i) - result = k - elif count == 0: - discard - else: - result = 0 - let err = osLastError() - if cint(err) != EINTR: - raiseOSError(err) - - proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = - result = newSeq[ReadyKey[T]](MAX_EPOLL_RESULT_EVENTS) - var count = selectInto(s, timeout, result) - result.setLen(count) - - # - # Windows select - # - - elif defined(windows): - const FD_SETSIZE = 64 - - import hashes, nativesockets - - when hasThreadSupport: - import sharedtables - else: - import tables - - proc hash*(x: SocketHandle): Hash {.borrow.} - proc `$`*(x: SocketHandle): string {.borrow.} - - proc WSAFDIsSet(s: SocketHandle, fdSet: var TFdSet): bool {. - stdcall, importc: "__WSAFDIsSet", dynlib: "ws2_32.dll", noSideEffect.} - - template iFD_ISSET(s: SocketHandle, fdSet: var TFdSet): bool = - if WSAFDIsSet(s, fdSet): true else: false - - template iFD_SET(s: SocketHandle, fdSet: var TFdSet) = - block: - var i = 0 - while i < fdSet.fd_count: - if fdSet.fd_array[i] == s: - break - inc(i) - if i == fdSet.fd_count: - if fdSet.fd_count < ioselectors.FD_SETSIZE: - fdSet.fd_array[i] = s - inc(fdSet.fd_count) - - template iFD_CLR(s: SocketHandle, fdSet: var TFdSet) = - block: - var i = 0 - while i < fdSet.fd_count: - if fdSet.fd_array[i] == s: - if i == fdSet.fd_count - 1: - fdSet.fd_array[i] = 0.SocketHandle - else: - while i < (fdSet.fd_count - 1): - fdSet.fd_array[i] = fdSet.fd_array[i + 1] - inc(i) - dec(fdSet.fd_count) - break - inc(i) - - template iFD_ZERO(fdSet: var TFdSet) = - fdSet.fd_count = 0 - - when hasThreadSupport: - type - SelectorImpl[T] = object - rSet: TFdSet - wSet: TFdSet - eSet: TFdSet - maxFD: uint - fds: SharedTable[SocketHandle, SelectorKey[T]] - count: int - lock: Lock - else: - type - SelectorImpl[T] = object - rSet: TFdSet - wSet: TFdSet - eSet: TFdSet - maxFD: uint - fds: Table[SocketHandle, SelectorKey[T]] - count: int - - when hasThreadSupport: - type Selector*[T] = ptr SelectorImpl[T] - else: - type Selector*[T] = ref SelectorImpl[T] - - type - SelectEventImpl = object - rsock: SocketHandle - wsock: SocketHandle - - type SelectEvent* = ptr SelectEventImpl - - when hasThreadSupport: - template withSelectLock[T](s: Selector[T], body: untyped) = - acquire(s.lock) - {.locks: [s.lock].}: - try: - body - finally: - release(s.lock) - else: - template withSelectLock[T](s: Selector[T], body: untyped) = - body - - proc newSelector*[T](): Selector[T] = - var maxFD = FD_SETSIZE - when hasThreadSupport: - result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) - result.maxFD = maxFD.uint - result.fds = initSharedTable[SocketHandle, SelectorKey[T]]() - initLock result.lock - else: - result = Selector[T](maxFD: FD_SETSIZE) - result.maxFD = maxFD.uint - result.fds = initTable[SocketHandle, SelectorKey[T]]() - - iFD_ZERO(result.rSet) - iFD_ZERO(result.wSet) - iFD_ZERO(result.eSet) - - proc close*(s: Selector) = - when hasThreadSupport: - deinitSharedTable(s.fds) - deallocShared(cast[pointer](s)) - - template isEmpty*[T](s: Selector[T]): bool = - (s.count == 0) - - template selectAdd[T](s: Selector[T], fd: SocketHandle, - events: set[Event]) = - mixin withSelectLock - s.withSelectLock(): - if Event.Read in events: - if s.rSet.fd_count == FD_SETSIZE: - raise newException(ValueError, "Maximum numbers of fds exceeded") - iFD_SET(fd, s.rSet) - inc(s.count) - if Event.Write in events: - if s.wSet.fd_count == FD_SETSIZE: - raise newException(ValueError, "Maximum numbers of fds exceeded") - iFD_SET(fd, s.wSet) - iFD_SET(fd, s.eSet) - inc(s.count) - - proc registerHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event], data: T) = - var fdi = int(fd) - var flags = {Event.flagHandle} + events - var nkey = SelectorKey[T](ident: fdi, flags: flags) - nkey.key.fd = fdi - nkey.key.data = data - - if s.fds.hasKeyOrPut(fd, nkey): - raise newException(ValueError, "Re-use of non closed descriptor") - selectAdd(s, fd, flags) - - proc updateHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event]) = - s.withSelectLock(): - withValue(s.fds, fd, skey) do: - if Event.flagHandle in skey.flags: - var oe = skey.flags - var ne = events + {Event.flagHandle} - if oe != ne: - if (Event.Read in oe) and (Event.Read notin ne): - iFD_CLR(fd, s.rSet) - dec(s.count) - if (Event.Write in oe) and (Event.Write notin ne): - iFD_CLR(fd, s.wSet) - iFD_CLR(fd, s.eSet) - dec(s.count) - if (Event.Read notin oe) and (Event.Read in ne): - iFD_SET(fd, s.rSet) - inc(s.count) - if (Event.Write notin oe) and (Event.Write in ne): - iFD_SET(fd, s.wSet) - iFD_SET(fd, s.eSet) - inc(s.count) - skey.flags = ne - else: - raise newException(ValueError, - "Could not update non-handle descriptor") - do: - raise newException(ValueError, - "Descriptor is not registered in queue") - - proc registerTimer*[T](s: Selector, timeout: int, oneshot: bool, - data: T): int {.discardable.} = - raise newException(ValueError, "Not implemented") - - proc registerSignal*[T](s: Selector, signal: int, - data: T): int {.discardable.} = - raise newException(ValueError, "Not implemented") - - proc registerProcess*[T](s: Selector, pid: int, - data: T): int {.discardable.} = - raise newException(ValueError, "Not implemented") - - proc flush*[T](s: Selector[T]) = discard - - proc unregister*[T](s: Selector[T], ev: SelectEvent) = - let fd = ev.rsock - s.withSelectLock(): - iFD_CLR(fd, s.rSet) - dec(s.count) - s.fds.del(fd) - - - proc unregister*[T](s: Selector[T], fd: SocketHandle) = - s.withSelectLock(): - s.fds.withValue(fd, skey) do: - if Event.Read in skey.flags: - iFD_CLR(fd, s.rSet) - dec(s.count) - if Event.Write in skey.flags: - iFD_CLR(fd, s.wSet) - iFD_CLR(fd, s.eSet) - dec(s.count) - s.fds.del(fd) - - proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = - var flags = {Event.flagUser, Event.Read} - var nkey = SelectorKey[T](ident: ev.rsock.int, flags: flags) - nkey.key.fd = ev.rsock.int - nkey.key.data = data - if s.fds.hasKeyOrPut(ev.rsock, nkey): - raise newException(ValueError, "Re-use of non closed descriptor") - selectAdd(s, ev.rsock, flags) - - proc newEvent*(): SelectEvent = - var ssock = newNativeSocket() - var wsock = newNativeSocket() - var rsock: SocketHandle = INVALID_SOCKET - var saddr = Sockaddr_in() - try: - saddr.sin_family = winlean.AF_INET - saddr.sin_port = 0 - saddr.sin_addr.s_addr = INADDR_ANY - if bindAddr(ssock, cast[ptr SockAddr](addr(saddr)), - sizeof(saddr).SockLen) < 0'i32: - raiseOSError(osLastError()) - - if winlean.listen(ssock, 1) == -1: - raiseOSError(osLastError()) - - var namelen = sizeof(saddr).SockLen - if getsockname(ssock, cast[ptr SockAddr](addr(saddr)), - addr(namelen)) == -1'i32: - raiseOSError(osLastError()) - - saddr.sin_addr.s_addr = 0x0100007F - if winlean.connect(wsock, cast[ptr SockAddr](addr(saddr)), - sizeof(saddr).SockLen) == -1: - raiseOSError(osLastError()) - namelen = sizeof(saddr).SockLen - rsock = winlean.accept(ssock, cast[ptr SockAddr](addr(saddr)), - cast[ptr SockLen](addr(namelen))) - if rsock == SocketHandle(-1): - raiseOSError(osLastError()) - - if winlean.closesocket(ssock) == -1: - raiseOSError(osLastError()) - - var mode = clong(1) - if ioctlsocket(rsock, FIONBIO, addr(mode)) == -1: - raiseOSError(osLastError()) - mode = clong(1) - if ioctlsocket(wsock, FIONBIO, addr(mode)) == -1: - raiseOSError(osLastError()) - - result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) - result.rsock = rsock - result.wsock = wsock - except: - discard winlean.closesocket(ssock) - discard winlean.closesocket(wsock) - if rsock != INVALID_SOCKET: - discard winlean.closesocket(rsock) - - proc setEvent*(ev: SelectEvent) = - var data: int = 1 - if winlean.send(ev.wsock, cast[pointer](addr data), - cint(sizeof(int)), 0) != sizeof(int): - raiseOSError(osLastError()) - - proc close*(ev: SelectEvent) = - discard winlean.closesocket(ev.rsock) - discard winlean.closesocket(ev.wsock) - deallocShared(cast[pointer](ev)) - - proc selectInto*[T](s: Selector[T], timeout: int, - results: var openarray[ReadyKey[T]]): int = - var tv = Timeval() - var ptv = addr tv - var rset, wset, eset: TFdSet - - if timeout != -1: - tv.tv_sec = timeout.int32 div 1_000 - tv.tv_usec = (timeout.int32 %% 1_000) * 1_000 - else: - ptv = nil - - s.withSelectLock(): - rset = s.rSet - wset = s.wSet - eset = s.eSet - - var count = select(cint(0), addr(rset), addr(wset), - addr(eset), ptv).int - if count > 0: - var rindex = 0 - var i = 0 - while i < rset.fd_count: - let fd = rset.fd_array[i] - if iFD_ISSET(fd, rset): - var events = {Event.Read} - if iFD_ISSET(fd, eset): events = events + {Event.Error} - if iFD_ISSET(fd, wset): events = events + {Event.Write} - s.fds.withValue(fd, skey) do: - if Event.flagHandle in skey.flags: - skey.key.events = events - elif Event.flagUser in skey.flags: - var data: int = 0 - if winlean.recv(fd, cast[pointer](addr(data)), - sizeof(int).cint, 0) != sizeof(int): - let err = osLastError() - if err != OSErrorCode(WSAEWOULDBLOCK): - raiseOSError(err) - else: - # someone already consumed event data - inc(i) - continue - skey.key.events = {Event.User} - results[rindex].fd = skey.key.fd - results[rindex].data = skey.key.data - results[rindex].events = skey.key.events - inc(rindex) - inc(i) - - i = 0 - while i < wset.fd_count: - let fd = wset.fd_array[i] - if iFD_ISSET(fd, wset): - var events = {Event.Write} - if not iFD_ISSET(fd, rset): - if iFD_ISSET(fd, eset): events = events + {Event.Error} - s.fds.withValue(fd, skey) do: - skey.key.events = events - results[rindex].fd = skey.key.fd - results[rindex].data = skey.key.data - results[rindex].events = skey.key.events - inc(rindex) - inc(i) - count = rindex - elif count == 0: - discard - else: - raiseOSError(osLastError()) - result = count - - proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = - result = newSeq[ReadyKey[T]](FD_SETSIZE) - var count = selectInto(s, timeout, result) - result.setLen(count) - - # - # Posix poll - # - - else: - # Maximum number of events that can be returned - const MAX_POLL_RESULT_EVENTS = 64 - - type - SelectorImpl[T] = object - maxFD : uint - pollcnt: int - fds: ptr SharedArray[SelectorKey[T]] - pollfds: ptr SharedArray[TPollFd] - count: int - when hasThreadSupport: - lock: Lock - - Selector*[T] = ptr SelectorImpl[T] - - SelectEventImpl = object - rfd: cint - wfd: cint - - SelectEvent* = ptr SelectEventImpl - - when hasThreadSupport: - template withPollLock[T](s: Selector[T], body: untyped) = - acquire(s.lock) - {.locks: [s.lock].}: - try: - body - finally: - release(s.lock) - else: - template withPollLock(s, body: untyped) = - body - - proc newSelector*[T](): Selector[T] = - var maxFD = getMaxFds() - - result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) - result.maxFD = maxFD.uint - result.fds = allocSharedArray[SelectorKey[T]](maxFD) - result.pollfds = allocSharedArray[TPollFd](maxFD) - when hasThreadSupport: - initLock(result.lock) - - proc close*[T](s: Selector[T]) = - when hasThreadSupport: - deinitLock(s.lock) - deallocSharedArray(s.fds) - deallocSharedArray(s.pollfds) - deallocShared(cast[pointer](s)) - - template pollAdd[T](s: Selector[T], sock: cint, events: set[Event]) = - withPollLock(s): - var pollev: cshort = 0 - if Event.Read in events: pollev = pollev or POLLIN - if Event.Write in events: pollev = pollev or POLLOUT - s.pollfds[s.pollcnt].fd = cint(sock) - s.pollfds[s.pollcnt].events = pollev - inc(s.count) - inc(s.pollcnt) - - template pollUpdate[T](s: Selector[T], sock: cint, events: set[Event]) = - withPollLock(s): - var i = 0 - var pollev: cshort = 0 - if Event.Read in events: pollev = pollev or POLLIN - if Event.Write in events: pollev = pollev or POLLOUT - - while i < s.pollcnt: - if s.pollfds[i].fd == sock: - s.pollfds[i].events = pollev - break - inc(i) - - if i == s.pollcnt: - raise newException(ValueError, - "Descriptor is not registered in queue") - - template pollRemove[T](s: Selector[T], sock: cint) = - withPollLock(s): - var i = 0 - while i < s.pollcnt: - if s.pollfds[i].fd == sock: - if i == s.pollcnt - 1: - s.pollfds[i].fd = 0 - s.pollfds[i].events = 0 - s.pollfds[i].revents = 0 - else: - while i < (s.pollcnt - 1): - s.pollfds[i].fd = s.pollfds[i + 1].fd - s.pollfds[i].events = s.pollfds[i + 1].events - inc(i) - break - inc(i) - dec(s.pollcnt) - dec(s.count) - - proc registerHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event], data: T) = - var fdi = int(fd) - s.checkMaxFd(fdi) - doAssert(s.fds[fdi].ident == 0) - setKey(s, fdi, fdi, {Event.flagHandle} + events, 0, data) - s.pollAdd(fdi.cint, events) - - proc updateHandle*[T](s: Selector[T], fd: SocketHandle, - events: set[Event]) = - var fdi = int(fd) - s.checkMaxFd(fdi) - var oe = s.fds[fdi].flags - doAssert(s.fds[fdi].ident != 0) - doAssert(Event.flagHandle in oe) - var ne = events + {Event.flagHandle} - if ne != oe: - if events != {}: - s.pollUpdate(fd.cint, events) - else: - s.pollRemove(fd.cint) - s.fds[fdi].flags = ne - - proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, - data: T): int {.discardable.} = - raise newException(ValueError, "Not implemented") - - proc registerSignal*[T](s: Selector[T], signal: int, - data: T): int {.discardable.} = - raise newException(ValueError, "Not implemented") - - proc registerProcess*[T](s: Selector[T], pid: int, - data: T): int {.discardable.} = - raise newException(ValueError, "Not implemented") - - proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = - var fdi = int(ev.rfd) - doAssert(s.fds[fdi].ident == 0) - var events = {Event.flagUser, Event.Read} - setKey(s, fdi, fdi, events, 0, data) - s.pollAdd(fdi.cint, events) - - proc flush*[T](s: Selector[T]) = discard - - template isEmpty*[T](s: Selector[T]): bool = - (s.count == 0) - - proc unregister*[T](s: Selector[T], fd: int|SocketHandle|cint) = - var fdi = int(fd) - if fdi.uint < s.maxFD: - if s.fds[fdi].ident != 0 and s.fds[fdi].flags != {}: - clearKey(s, fdi) - s.pollRemove(fdi.cint) - - proc unregister*[T](s: Selector[T], ev: SelectEvent) = - var fdi = int(ev.rfd) - if fdi.uint < s.maxFD: - if s.fds[fdi].ident != 0 and (Event.flagUser in s.fds[fdi].flags): - clearKey(s, fdi) - s.pollRemove(fdi.cint) - - proc newEvent*(): SelectEvent = - var fds: array[2, cint] - if posix.pipe(fds) == -1: - raiseOSError(osLastError()) - setNonBlocking(fds[0]) - setNonBlocking(fds[1]) - result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) - result.rfd = fds[0] - result.wfd = fds[1] - - proc setEvent*(ev: SelectEvent) = - var data: int = 1 - if posix.write(ev.wfd, addr data, sizeof(int)) != sizeof(int): - raiseOSError(osLastError()) - - proc close*(ev: SelectEvent) = - discard posix.close(cint(ev.rfd)) - discard posix.close(cint(ev.wfd)) - deallocShared(cast[pointer](ev)) - - proc selectInto*[T](s: Selector[T], timeout: int, - results: var openarray[ReadyKey[T]]): int = - var maxResults = MAX_POLL_RESULT_EVENTS - if maxResults > len(results): - maxResults = len(results) - - s.withPollLock(): - var count = posix.poll(addr(s.pollfds[0]), Tnfds(s.pollcnt), timeout) - if count > 0: - var i = 0 - var k = 0 - var rindex = 0 - while (i < s.pollcnt) and (k < count) and (rindex < maxResults): - let revents = s.pollfds[i].revents - let fd = s.pollfds[i].fd - if revents != 0: - var events: set[Event] = {} - if (revents and POLLIN) != 0: - events = events + {Event.Read} - if (revents and POLLOUT) != 0: - events = events + {Event.Write} - if (revents and POLLERR) != 0 or (revents and POLLHUP) != 0 or - (revents and POLLNVAL) != 0: - events = events + {Event.Error} - - var skey = addr(s.fds[fd]) - if Event.flagUser in skey.flags: - if Event.Read in events: - var data: int = 0 - if posix.read(fd, addr data, sizeof(int)) != sizeof(int): - let err = osLastError() - if err != OSErrorCode(EAGAIN): - raiseOSError(osLastError()) - else: - # someone already consumed event data - inc(i) - continue - events = {Event.User} - - results[rindex].fd = fd - results[rindex].events = events - results[rindex].data = skey.key.data - s.pollfds[i].revents = 0 - inc(rindex) - inc(k) - inc(i) - result = k - elif count == 0: - discard - else: - let err = osLastError() - if err.cint == EINTR: - discard - else: - raiseOSError(osLastError()) - - proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = - result = newSeq[ReadyKey[T]](MAX_POLL_RESULT_EVENTS) - var count = selectInto(s, timeout, result) - result.setLen(count) - - when not defined(windows): - template withData*[T](s: Selector[T], fd: SocketHandle, value, - body: untyped) = - var fdi = int(fd) - s.checkMaxFd(fdi) - if s.fds[fdi].ident != 0: - var value = addr(s.fds[fdi].key.data) - body - - template withData*[T](s: Selector[T], fd: SocketHandle, value, body1, - body2: untyped) = - var fdi = int(fd) - s.checkMaxFd(fdi) - if s.fds[fdi].ident != 0: - var value = addr(s.fds[fdi].key.data) - body1 - else: - body2 - else: - template withData*(s: Selector, fd: SocketHandle, value, body: untyped) = - s.fds.withValue(fd, skey) do: - var value {.inject.} = addr(skey.key.data) - body - - template withData*(s: Selector, fd: SocketHandle, value, - body1, body2: untyped) = - s.fds.withValue(fd, skey) do: - var value {.inject.} = addr(skey.key.data) - body1 - do: - body2 |