diff options
Diffstat (limited to 'lib/pure/ioselects')
| -rw-r--r-- | lib/pure/ioselects/ioselectors_epoll.nim | 534 | ||||
| -rw-r--r-- | lib/pure/ioselects/ioselectors_kqueue.nim | 639 | ||||
| -rw-r--r-- | lib/pure/ioselects/ioselectors_poll.nim | 324 | ||||
| -rw-r--r-- | lib/pure/ioselects/ioselectors_select.nim | 454 |
4 files changed, 1951 insertions, 0 deletions
diff --git a/lib/pure/ioselects/ioselectors_epoll.nim b/lib/pure/ioselects/ioselectors_epoll.nim new file mode 100644 index 000000000..10658b78e --- /dev/null +++ b/lib/pure/ioselects/ioselectors_epoll.nim @@ -0,0 +1,534 @@ +# +# +# 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 implements Linux epoll(). + +import std/[posix, times, epoll] + +# Maximum number of events that can be returned +const MAX_EPOLL_EVENTS = 64 + +when not defined(android): + 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] + +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 eventfd(count: cuint, flags: cint): cint + {.cdecl, importc: "eventfd", header: "<sys/eventfd.h>".} + +when not defined(android): + proc signalfd(fd: cint, mask: var Sigset, flags: cint): cint + {.cdecl, importc: "signalfd", header: "<sys/signalfd.h>".} + +when hasThreadSupport: + type + SelectorImpl[T] = object + epollFD: cint + maxFD: int + numFD: int + fds: ptr SharedArray[SelectorKey[T]] + count*: int + Selector*[T] = ptr SelectorImpl[T] +else: + type + SelectorImpl[T] = object + epollFD: cint + maxFD: int + numFD: int + fds: seq[SelectorKey[T]] + count*: int + Selector*[T] = ref SelectorImpl[T] +type + SelectEventImpl = object + efd: cint + SelectEvent* = ptr SelectEventImpl + +proc newSelector*[T](): Selector[T] = + proc initialNumFD(): int {.inline.} = + when defined(nuttx): + result = NEPOLL_MAX + else: + result = 1024 + # Retrieve the maximum fd count (for current OS) via getrlimit() + var maxFD = maxDescriptors() + doAssert(maxFD > 0) + # Start with a reasonable size, checkFd() will grow this on demand + let numFD = initialNumFD() + + var epollFD = epoll_create1(O_CLOEXEC) + if epollFD < 0: + raiseOSError(osLastError()) + + when hasThreadSupport: + result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) + result.epollFD = epollFD + result.maxFD = maxFD + result.numFD = numFD + result.fds = allocSharedArray[SelectorKey[T]](numFD) + else: + result = Selector[T]() + result.epollFD = epollFD + result.maxFD = maxFD + result.numFD = numFD + result.fds = newSeq[SelectorKey[T]](numFD) + + for i in 0 ..< numFD: + result.fds[i].ident = InvalidIdent + +proc close*[T](s: Selector[T]) = + let res = posix.close(s.epollFD) + when hasThreadSupport: + deallocSharedArray(s.fds) + deallocShared(cast[pointer](s)) + if res != 0: + raiseIOSelectorsError(osLastError()) + +proc newSelectEvent*(): SelectEvent = + let fdci = eventfd(0, O_CLOEXEC or O_NONBLOCK) + if fdci == -1: + raiseIOSelectorsError(osLastError()) + result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) + result.efd = fdci + +proc trigger*(ev: SelectEvent) = + var data: uint64 = 1 + if posix.write(ev.efd, addr data, sizeof(uint64)) == -1: + raiseIOSelectorsError(osLastError()) + +proc close*(ev: SelectEvent) = + let res = posix.close(ev.efd) + deallocShared(cast[pointer](ev)) + if res != 0: + raiseIOSelectorsError(osLastError()) + +template checkFd(s, f) = + # TODO: I don't see how this can ever happen. You won't be able to create an + # FD if there is too many. -- DP + if f >= s.maxFD: + raiseIOSelectorsError("Maximum number of descriptors is exhausted!") + if f >= s.numFD: + var numFD = s.numFD + while numFD <= f: numFD *= 2 + when hasThreadSupport: + s.fds = reallocSharedArray(s.fds, s.numFD, numFD) + else: + s.fds.setLen(numFD) + for i in s.numFD ..< numFD: + s.fds[i].ident = InvalidIdent + s.numFD = numFD + +proc registerHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event], data: T) = + let fdi = int(fd) + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent, "Descriptor $# already registered" % $fdi) + s.setKey(fdi, events, 0, data) + if events != {}: + var epv = EpollEvent(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) != 0: + raiseIOSelectorsError(osLastError()) + inc(s.count) + +proc updateHandle*[T](s: Selector[T], fd: int | SocketHandle, events: set[Event]) = + let maskEvents = {Event.Timer, Event.Signal, Event.Process, Event.Vnode, + Event.User, Event.Oneshot, Event.Error} + let fdi = int(fd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, + "Descriptor $# is not registered in the selector!" % $fdi) + doAssert(pkey.events * maskEvents == {}) + if pkey.events != events: + var epv = EpollEvent(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 pkey.events == {}: + if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + inc(s.count) + else: + if events != {}: + if epoll_ctl(s.epollFD, EPOLL_CTL_MOD, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + else: + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + dec(s.count) + pkey.events = events + +proc unregister*[T](s: Selector[T], fd: int|SocketHandle) = + let fdi = int(fd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, + "Descriptor $# is not registered in the selector!" % $fdi) + if pkey.events != {}: + when not defined(android): + if Event.Read in pkey.events or Event.Write in pkey.events or Event.User in pkey.events: + var epv = EpollEvent() + # TODO: Refactor all these EPOLL_CTL_DEL + dec(s.count) into a proc. + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + dec(s.count) + elif Event.Timer in pkey.events: + if Event.Finished notin pkey.events: + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + dec(s.count) + if posix.close(cint(fdi)) != 0: + raiseIOSelectorsError(osLastError()) + elif Event.Signal in pkey.events: + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + var nmask, omask: Sigset + discard sigemptyset(nmask) + discard sigemptyset(omask) + discard sigaddset(nmask, cint(s.fds[fdi].param)) + unblockSignals(nmask, omask) + dec(s.count) + if posix.close(cint(fdi)) != 0: + raiseIOSelectorsError(osLastError()) + elif Event.Process in pkey.events: + if Event.Finished notin pkey.events: + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + var nmask, omask: Sigset + discard sigemptyset(nmask) + discard sigemptyset(omask) + discard sigaddset(nmask, SIGCHLD) + unblockSignals(nmask, omask) + dec(s.count) + if posix.close(cint(fdi)) != 0: + raiseIOSelectorsError(osLastError()) + else: + if Event.Read in pkey.events or Event.Write in pkey.events or Event.User in pkey.events: + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + dec(s.count) + elif Event.Timer in pkey.events: + if Event.Finished notin pkey.events: + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + dec(s.count) + if posix.close(cint(fdi)) != 0: + raiseIOSelectorsError(osLastError()) + clearKey(pkey) + +proc unregister*[T](s: Selector[T], ev: SelectEvent) = + let fdi = int(ev.efd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, "Event is not registered in the queue!") + doAssert(Event.User in pkey.events) + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + dec(s.count) + clearKey(pkey) + +proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, + data: T): int {.discardable.} = + var + newTs: Itimerspec + oldTs: Itimerspec + let fdi = timerfd_create(CLOCK_MONOTONIC, O_CLOEXEC or O_NONBLOCK).int + if fdi == -1: + raiseIOSelectorsError(osLastError()) + + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + + var events = {Event.Timer} + var epv = EpollEvent(events: EPOLLIN or EPOLLRDHUP) + epv.data.u64 = fdi.uint + + if oneshot: + newTs.it_interval.tv_sec = posix.Time(0) + newTs.it_interval.tv_nsec = 0 + newTs.it_value.tv_sec = posix.Time(timeout div 1_000) + newTs.it_value.tv_nsec = (timeout %% 1_000) * 1_000_000 + incl(events, Event.Oneshot) + epv.events = epv.events or EPOLLONESHOT + else: + newTs.it_interval.tv_sec = posix.Time(timeout div 1000) + newTs.it_interval.tv_nsec = (timeout %% 1_000) * 1_000_000 + newTs.it_value.tv_sec = newTs.it_interval.tv_sec + newTs.it_value.tv_nsec = newTs.it_interval.tv_nsec + + if timerfd_settime(fdi.cint, cint(0), newTs, oldTs) != 0: + raiseIOSelectorsError(osLastError()) + if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + s.setKey(fdi, events, 0, data) + inc(s.count) + result = fdi + +when not defined(android): + 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) + + let fdi = signalfd(-1, nmask, O_CLOEXEC or O_NONBLOCK).int + if fdi == -1: + raiseIOSelectorsError(osLastError()) + + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + + var epv = EpollEvent(events: EPOLLIN or EPOLLRDHUP) + epv.data.u64 = fdi.uint + if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + s.setKey(fdi, {Event.Signal}, 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) + + let fdi = signalfd(-1, nmask, O_CLOEXEC or O_NONBLOCK).int + if fdi == -1: + raiseIOSelectorsError(osLastError()) + + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + + var epv = EpollEvent(events: EPOLLIN or EPOLLRDHUP) + epv.data.u64 = fdi.uint + epv.events = EPOLLIN or EPOLLRDHUP + if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + s.setKey(fdi, {Event.Process, Event.Oneshot}, pid, data) + inc(s.count) + result = fdi + +proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = + let fdi = int(ev.efd) + doAssert(s.fds[fdi].ident == InvalidIdent, "Event is already registered in the queue!") + s.setKey(fdi, {Event.User}, 0, data) + var epv = EpollEvent(events: EPOLLIN or EPOLLRDHUP) + epv.data.u64 = ev.efd.uint + if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, ev.efd, addr epv) != 0: + raiseIOSelectorsError(osLastError()) + inc(s.count) + +proc selectInto*[T](s: Selector[T], timeout: int, + results: var openArray[ReadyKey]): int = + var + resTable: array[MAX_EPOLL_EVENTS, EpollEvent] + maxres = MAX_EPOLL_EVENTS + i, k: int + + if maxres > len(results): + maxres = len(results) + + verifySelectParams(timeout) + + let count = epoll_wait(s.epollFD, addr(resTable[0]), maxres.cint, + timeout.cint) + if count < 0: + result = 0 + let err = osLastError() + if cint(err) != EINTR: + raiseIOSelectorsError(err) + elif count == 0: + result = 0 + else: + i = 0 + k = 0 + while i < count: + let fdi = int(resTable[i].data.u64) + let pevents = resTable[i].events + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent) + var rkey = ReadyKey(fd: fdi, events: {}) + + if (pevents and EPOLLERR) != 0 or (pevents and EPOLLHUP) != 0: + if (pevents and EPOLLHUP) != 0: + rkey.errorCode = OSErrorCode ECONNRESET + else: + # Try reading SO_ERROR from fd. + var error: cint + var size = SockLen sizeof(error) + if getsockopt(SocketHandle fdi, SOL_SOCKET, SO_ERROR, addr(error), + addr(size)) == 0'i32: + rkey.errorCode = OSErrorCode error + + rkey.events.incl(Event.Error) + if (pevents and EPOLLOUT) != 0: + rkey.events.incl(Event.Write) + when not defined(android): + if (pevents and EPOLLIN) != 0: + if Event.Read in pkey.events: + rkey.events.incl(Event.Read) + elif Event.Timer in pkey.events: + var data: uint64 = 0 + if posix.read(cint(fdi), addr data, + sizeof(uint64)) != sizeof(uint64): + raiseIOSelectorsError(osLastError()) + rkey.events.incl(Event.Timer) + elif Event.Signal in pkey.events: + var data = SignalFdInfo() + if posix.read(cint(fdi), addr data, + sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): + raiseIOSelectorsError(osLastError()) + rkey.events.incl(Event.Signal) + elif Event.Process in pkey.events: + var data = SignalFdInfo() + if posix.read(cint(fdi), addr data, + sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): + raiseIOSelectorsError(osLastError()) + if cast[int](data.ssi_pid) == pkey.param: + rkey.events.incl(Event.Process) + else: + inc(i) + continue + elif Event.User in pkey.events: + var data: uint64 = 0 + if posix.read(cint(fdi), addr data, + sizeof(uint64)) != sizeof(uint64): + let err = osLastError() + if err == OSErrorCode(EAGAIN): + inc(i) + continue + else: + raiseIOSelectorsError(err) + rkey.events.incl(Event.User) + else: + if (pevents and EPOLLIN) != 0: + if Event.Read in pkey.events: + rkey.events.incl(Event.Read) + elif Event.Timer in pkey.events: + var data: uint64 = 0 + if posix.read(cint(fdi), addr data, + sizeof(uint64)) != sizeof(uint64): + raiseIOSelectorsError(osLastError()) + rkey.events.incl(Event.Timer) + elif Event.User in pkey.events: + var data: uint64 = 0 + if posix.read(cint(fdi), addr data, + sizeof(uint64)) != sizeof(uint64): + let err = osLastError() + if err == OSErrorCode(EAGAIN): + inc(i) + continue + else: + raiseIOSelectorsError(err) + rkey.events.incl(Event.User) + + if Event.Oneshot in pkey.events: + var epv = EpollEvent() + if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, cint(fdi), addr epv) != 0: + raiseIOSelectorsError(osLastError()) + # we will not clear key until it will be unregistered, so + # application can obtain data, but we will decrease counter, + # because epoll is empty. + dec(s.count) + # we are marking key with `Finished` event, to avoid double decrease. + pkey.events.incl(Event.Finished) + + results[k] = rkey + inc(k) + inc(i) + result = k + +proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] = + result = newSeq[ReadyKey](MAX_EPOLL_EVENTS) + let count = selectInto(s, timeout, result) + result.setLen(count) + +template isEmpty*[T](s: Selector[T]): bool = + (s.count == 0) + +proc contains*[T](s: Selector[T], fd: SocketHandle|int): bool {.inline.} = + return s.fds[fd.int].ident != InvalidIdent + +proc getData*[T](s: Selector[T], fd: SocketHandle|int): var T = + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + result = s.fds[fdi].data + +proc setData*[T](s: Selector[T], fd: SocketHandle|int, data: T): bool = + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + s.fds[fdi].data = data + result = true + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, + body: untyped) = + mixin checkFd + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + var value = addr(s.fds[fdi].data) + body + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1, + body2: untyped) = + mixin checkFd + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + var value = addr(s.fds[fdi].data) + body1 + else: + body2 + +proc getFd*[T](s: Selector[T]): int = + return s.epollFD.int diff --git a/lib/pure/ioselects/ioselectors_kqueue.nim b/lib/pure/ioselects/ioselectors_kqueue.nim new file mode 100644 index 000000000..513578eda --- /dev/null +++ b/lib/pure/ioselects/ioselectors_kqueue.nim @@ -0,0 +1,639 @@ +# +# +# 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 implements BSD kqueue(). + +import std/[posix, times, kqueue, nativesockets] + +const + # Maximum number of events that can be returned. + MAX_KQUEUE_EVENTS = 64 + # SIG_IGN and SIG_DFL declared in posix.nim as variables, but we need them + # to be constants and GC-safe. + SIG_DFL = cast[proc(x: cint) {.noconv,gcsafe.}](0) + SIG_IGN = cast[proc(x: cint) {.noconv,gcsafe.}](1) + +when defined(kqcache): + const CACHE_EVENTS = true + +when defined(macosx) or defined(freebsd) or defined(dragonfly): + when defined(macosx): + const MAX_DESCRIPTORS_ID = 29 # KERN_MAXFILESPERPROC (MacOS) + else: + const MAX_DESCRIPTORS_ID = 27 # KERN_MAXFILESPERPROC (FreeBSD) + proc sysctl(name: ptr cint, namelen: cuint, oldp: pointer, oldplen: ptr csize_t, + newp: pointer, newplen: csize_t): 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 MAX_DESCRIPTORS_ID = 7 # KERN_MAXFILES + proc sysctl(name: ptr cint, namelen: cuint, oldp: pointer, oldplen: ptr csize_t, + newp: pointer, newplen: csize_t): cint + {.importc: "sysctl",header: """#include <sys/param.h> + #include <sys/sysctl.h>""".} + +when hasThreadSupport: + type + SelectorImpl[T] = object + kqFD: cint + maxFD: int + changes: ptr SharedArray[KEvent] + fds: ptr SharedArray[SelectorKey[T]] + count*: int + changesLock: Lock + changesSize: int + changesLength: int + sock: cint + Selector*[T] = ptr SelectorImpl[T] +else: + type + SelectorImpl[T] = object + kqFD: cint + maxFD: int + changes: seq[KEvent] + fds: seq[SelectorKey[T]] + count*: int + sock: cint + Selector*[T] = ref SelectorImpl[T] + +type + SelectEventImpl = object + rfd: cint + wfd: cint + + SelectEvent* = ptr SelectEventImpl + # SelectEvent is declared as `ptr` to be placed in `shared memory`, + # so you can share one SelectEvent handle between threads. + +proc getUnique[T](s: Selector[T]): int {.inline.} = + # we create duplicated handles to get unique indexes for our `fds` array. + result = posix.fcntl(s.sock, F_DUPFD_CLOEXEC, s.sock) + if result == -1: + raiseIOSelectorsError(osLastError()) + +proc newSelector*[T](): owned(Selector[T]) = + var maxFD = 0.cint + var size = csize_t(sizeof(cint)) + var namearr = [1.cint, MAX_DESCRIPTORS_ID.cint] + # Obtain maximum number of opened file descriptors for process + if sysctl(addr(namearr[0]), 2, cast[pointer](addr maxFD), addr size, + nil, 0) != 0: + raiseIOSelectorsError(osLastError()) + + var kqFD = kqueue() + if kqFD < 0: + raiseIOSelectorsError(osLastError()) + + # we allocating empty socket to duplicate it handle in future, to get unique + # indexes for `fds` array. This is needed to properly identify + # {Event.Timer, Event.Signal, Event.Process} events. + let usock = createNativeSocket(posix.AF_INET, posix.SOCK_STREAM, + posix.IPPROTO_TCP).cint + if usock == -1: + let err = osLastError() + discard posix.close(kqFD) + raiseIOSelectorsError(err) + + when hasThreadSupport: + result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) + result.fds = allocSharedArray[SelectorKey[T]](maxFD) + result.changes = allocSharedArray[KEvent](MAX_KQUEUE_EVENTS) + result.changesSize = MAX_KQUEUE_EVENTS + initLock(result.changesLock) + else: + result = Selector[T]() + result.fds = newSeq[SelectorKey[T]](maxFD) + result.changes = newSeqOfCap[KEvent](MAX_KQUEUE_EVENTS) + + for i in 0 ..< maxFD: + result.fds[i].ident = InvalidIdent + + result.sock = usock + result.kqFD = kqFD + result.maxFD = maxFD.int + +proc close*[T](s: Selector[T]) = + let res1 = posix.close(s.kqFD) + let res2 = posix.close(s.sock) + when hasThreadSupport: + deinitLock(s.changesLock) + deallocSharedArray(s.fds) + deallocShared(cast[pointer](s)) + if res1 != 0 or res2 != 0: + raiseIOSelectorsError(osLastError()) + +proc newSelectEvent*(): SelectEvent = + var fds: array[2, cint] + if posix.pipe(fds) != 0: + raiseIOSelectorsError(osLastError()) + setNonBlocking(fds[0]) + setNonBlocking(fds[1]) + result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) + result.rfd = fds[0] + result.wfd = fds[1] + +proc trigger*(ev: SelectEvent) = + var data: uint64 = 1 + if posix.write(ev.wfd, addr data, sizeof(uint64)) != sizeof(uint64): + raiseIOSelectorsError(osLastError()) + +proc close*(ev: SelectEvent) = + let res1 = posix.close(ev.rfd) + let res2 = posix.close(ev.wfd) + deallocShared(cast[pointer](ev)) + if res1 != 0 or res2 != 0: + raiseIOSelectorsError(osLastError()) + +template checkFd(s, f) = + if f >= s.maxFD: + raiseIOSelectorsError("Maximum number of descriptors is exhausted!") + +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 + +when hasThreadSupport: + template modifyKQueue[T](s: Selector[T], nident: uint, nfilter: cshort, + nflags: cushort, nfflags: cuint, ndata: int, + nudata: pointer) = + mixin withChangeLock + s.withChangeLock(): + if s.changesLength == s.changesSize: + # if cache array is full, we allocating new with size * 2 + let newSize = s.changesSize shl 1 + let rdata = allocSharedArray[KEvent](newSize) + copyMem(rdata, s.changes, s.changesSize * sizeof(KEvent)) + s.changesSize = newSize + s.changes[s.changesLength] = KEvent(ident: nident, + filter: nfilter, flags: nflags, + fflags: nfflags, data: ndata, + udata: nudata) + inc(s.changesLength) + + when not declared(CACHE_EVENTS): + template flushKQueue[T](s: Selector[T]) = + mixin withChangeLock + s.withChangeLock(): + if s.changesLength > 0: + if kevent(s.kqFD, addr(s.changes[0]), cint(s.changesLength), + nil, 0, nil) == -1: + let res = osLastError() + if cint(res) != ENOENT: # ignore pipes whose read end is closed + raiseIOSelectorsError(res) + s.changesLength = 0 +else: + template modifyKQueue[T](s: Selector[T], nident: uint, nfilter: cshort, + nflags: cushort, nfflags: cuint, ndata: int, + nudata: pointer) = + s.changes.add(KEvent(ident: nident, + filter: nfilter, flags: nflags, + fflags: nfflags, data: ndata, + udata: nudata)) + + when not declared(CACHE_EVENTS): + template flushKQueue[T](s: Selector[T]) = + let length = cint(len(s.changes)) + if length > 0: + if kevent(s.kqFD, addr(s.changes[0]), length, + nil, 0, nil) == -1: + let res = osLastError() + if cint(res) != ENOENT: # ignore pipes whose read end is closed + raiseIOSelectorsError(res) + s.changes.setLen(0) + +proc registerHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event], data: T) = + let fdi = int(fd) + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + s.setKey(fdi, events, 0, data) + + if events != {}: + if Event.Read in events: + modifyKQueue(s, uint(fdi), EVFILT_READ, EV_ADD, 0, 0, nil) + inc(s.count) + if Event.Write in events: + modifyKQueue(s, uint(fdi), EVFILT_WRITE, EV_ADD, 0, 0, nil) + inc(s.count) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + +proc updateHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event]) = + let maskEvents = {Event.Timer, Event.Signal, Event.Process, Event.Vnode, + Event.User, Event.Oneshot, Event.Error} + let fdi = int(fd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, + "Descriptor $# is not registered in the queue!" % $fdi) + doAssert(pkey.events * maskEvents == {}) + + if pkey.events != events: + if (Event.Read in pkey.events) and (Event.Read notin events): + modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) + dec(s.count) + if (Event.Write in pkey.events) and (Event.Write notin events): + modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil) + dec(s.count) + if (Event.Read notin pkey.events) and (Event.Read in events): + modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) + inc(s.count) + if (Event.Write notin pkey.events) and (Event.Write in events): + modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil) + inc(s.count) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + + pkey.events = events + +proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, + data: T): int {.discardable.} = + let fdi = getUnique(s) + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + + let events = if oneshot: {Event.Timer, Event.Oneshot} else: {Event.Timer} + let flags: cushort = if oneshot: EV_ONESHOT or EV_ADD else: EV_ADD + + s.setKey(fdi, events, 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, flags, 0, cint(timeout), nil) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + + inc(s.count) + result = fdi + +proc registerSignal*[T](s: Selector[T], signal: int, + data: T): int {.discardable.} = + let fdi = getUnique(s) + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + + s.setKey(fdi, {Event.Signal}, signal, data) + var nmask, 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)) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + + inc(s.count) + result = fdi + +proc registerProcess*[T](s: Selector[T], pid: int, + data: T): int {.discardable.} = + let fdi = getUnique(s) + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + + var kflags: cushort = EV_ONESHOT or EV_ADD + setKey(s, fdi, {Event.Process, Event.Oneshot}, pid, data) + + modifyKQueue(s, pid.uint, EVFILT_PROC, kflags, NOTE_EXIT, 0, + cast[pointer](fdi)) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + + inc(s.count) + result = fdi + +proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = + let fdi = ev.rfd.int + doAssert(s.fds[fdi].ident == InvalidIdent, "Event is already registered in the queue!") + setKey(s, fdi, {Event.User}, 0, data) + + modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + + inc(s.count) + +template processVnodeEvents(events: set[Event]): cuint = + var rfflags = 0.cuint + if events == {Event.VnodeWrite, Event.VnodeDelete, Event.VnodeExtend, + Event.VnodeAttrib, Event.VnodeLink, Event.VnodeRename, + Event.VnodeRevoke}: + rfflags = NOTE_DELETE or NOTE_WRITE or NOTE_EXTEND or NOTE_ATTRIB or + NOTE_LINK or NOTE_RENAME or NOTE_REVOKE + else: + if Event.VnodeDelete in events: rfflags = rfflags or NOTE_DELETE + if Event.VnodeWrite in events: rfflags = rfflags or NOTE_WRITE + if Event.VnodeExtend in events: rfflags = rfflags or NOTE_EXTEND + if Event.VnodeAttrib in events: rfflags = rfflags or NOTE_ATTRIB + if Event.VnodeLink in events: rfflags = rfflags or NOTE_LINK + if Event.VnodeRename in events: rfflags = rfflags or NOTE_RENAME + if Event.VnodeRevoke in events: rfflags = rfflags or NOTE_REVOKE + rfflags + +proc registerVnode*[T](s: Selector[T], fd: cint, events: set[Event], data: T) = + let fdi = fd.int + setKey(s, fdi, {Event.Vnode} + events, 0, data) + var fflags = processVnodeEvents(events) + + modifyKQueue(s, fdi.uint, EVFILT_VNODE, EV_ADD or EV_CLEAR, fflags, 0, nil) + + when not declared(CACHE_EVENTS): + flushKQueue(s) + + inc(s.count) + +proc unregister*[T](s: Selector[T], fd: int|SocketHandle) = + let fdi = int(fd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, + "Descriptor [" & $fdi & "] is not registered in the queue!") + + if pkey.events != {}: + if pkey.events * {Event.Read, Event.Write} != {}: + if Event.Read in pkey.events: + modifyKQueue(s, uint(fdi), EVFILT_READ, EV_DELETE, 0, 0, nil) + dec(s.count) + if Event.Write in pkey.events: + modifyKQueue(s, uint(fdi), EVFILT_WRITE, EV_DELETE, 0, 0, nil) + dec(s.count) + when not declared(CACHE_EVENTS): + flushKQueue(s) + elif Event.Timer in pkey.events: + if Event.Finished notin pkey.events: + modifyKQueue(s, uint(fdi), EVFILT_TIMER, EV_DELETE, 0, 0, nil) + when not declared(CACHE_EVENTS): + flushKQueue(s) + dec(s.count) + if posix.close(cint(pkey.ident)) != 0: + raiseIOSelectorsError(osLastError()) + elif Event.Signal in pkey.events: + var nmask, omask: Sigset + let signal = cint(pkey.param) + discard sigemptyset(nmask) + discard sigemptyset(omask) + discard sigaddset(nmask, signal) + unblockSignals(nmask, omask) + posix.signal(signal, SIG_DFL) + modifyKQueue(s, uint(pkey.param), EVFILT_SIGNAL, EV_DELETE, 0, 0, nil) + when not declared(CACHE_EVENTS): + flushKQueue(s) + dec(s.count) + if posix.close(cint(pkey.ident)) != 0: + raiseIOSelectorsError(osLastError()) + elif Event.Process in pkey.events: + if Event.Finished notin pkey.events: + modifyKQueue(s, uint(pkey.param), EVFILT_PROC, EV_DELETE, 0, 0, nil) + when not declared(CACHE_EVENTS): + flushKQueue(s) + dec(s.count) + if posix.close(cint(pkey.ident)) != 0: + raiseIOSelectorsError(osLastError()) + elif Event.Vnode in pkey.events: + modifyKQueue(s, uint(fdi), EVFILT_VNODE, EV_DELETE, 0, 0, nil) + when not declared(CACHE_EVENTS): + flushKQueue(s) + dec(s.count) + elif Event.User in pkey.events: + modifyKQueue(s, uint(fdi), EVFILT_READ, EV_DELETE, 0, 0, nil) + when not declared(CACHE_EVENTS): + flushKQueue(s) + dec(s.count) + + clearKey(pkey) + +proc unregister*[T](s: Selector[T], ev: SelectEvent) = + let fdi = int(ev.rfd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, "Event is not registered in the queue!") + doAssert(Event.User in pkey.events) + modifyKQueue(s, uint(fdi), EVFILT_READ, EV_DELETE, 0, 0, nil) + when not declared(CACHE_EVENTS): + flushKQueue(s) + clearKey(pkey) + dec(s.count) + +proc selectInto*[T](s: Selector[T], timeout: int, + results: var openArray[ReadyKey]): int = + var + tv: Timespec + resTable: array[MAX_KQUEUE_EVENTS, KEvent] + ptv = addr tv + maxres = MAX_KQUEUE_EVENTS + + verifySelectParams(timeout) + + if timeout != -1: + if timeout >= 1000: + tv.tv_sec = posix.Time(timeout div 1_000) + tv.tv_nsec = (timeout %% 1_000) * 1_000_000 + else: + tv.tv_sec = posix.Time(0) + tv.tv_nsec = timeout * 1_000_000 + else: + ptv = nil + + if maxres > len(results): + maxres = len(results) + + var count = 0 + when not declared(CACHE_EVENTS): + count = kevent(s.kqFD, nil, cint(0), addr(resTable[0]), cint(maxres), ptv) + else: + when hasThreadSupport: + s.withChangeLock(): + if s.changesLength > 0: + count = kevent(s.kqFD, addr(s.changes[0]), cint(s.changesLength), + addr(resTable[0]), cint(maxres), ptv) + s.changesLength = 0 + else: + count = kevent(s.kqFD, nil, cint(0), addr(resTable[0]), cint(maxres), + ptv) + else: + let length = cint(len(s.changes)) + if length > 0: + count = kevent(s.kqFD, addr(s.changes[0]), length, + addr(resTable[0]), cint(maxres), ptv) + s.changes.setLen(0) + else: + count = kevent(s.kqFD, nil, cint(0), addr(resTable[0]), cint(maxres), + ptv) + + if count < 0: + result = 0 + let err = osLastError() + if cint(err) != EINTR: + raiseIOSelectorsError(err) + elif count == 0: + result = 0 + else: + var i = 0 + var k = 0 # do not delete this, because `continue` used in cycle. + var pkey: ptr SelectorKey[T] + while i < count: + let kevent = addr(resTable[i]) + var rkey = ReadyKey(fd: int(kevent.ident), events: {}) + + if (kevent.flags and EV_ERROR) != 0: + rkey.events = {Event.Error} + rkey.errorCode = OSErrorCode(kevent.data) + + case kevent.filter: + of EVFILT_READ: + pkey = addr(s.fds[int(kevent.ident)]) + rkey.events.incl(Event.Read) + if Event.User in pkey.events: + var data: uint64 = 0 + if posix.read(cint(kevent.ident), addr data, + sizeof(uint64)) != sizeof(uint64): + let err = osLastError() + if err == OSErrorCode(EAGAIN): + # someone already consumed event data + inc(i) + continue + else: + raiseIOSelectorsError(err) + rkey.events = {Event.User} + of EVFILT_WRITE: + pkey = addr(s.fds[int(kevent.ident)]) + rkey.events.incl(Event.Write) + rkey.events = {Event.Write} + of EVFILT_TIMER: + pkey = addr(s.fds[int(kevent.ident)]) + if Event.Oneshot in pkey.events: + # we will not clear key until it will be unregistered, so + # application can obtain data, but we will decrease counter, + # because kqueue is empty. + dec(s.count) + # we are marking key with `Finished` event, to avoid double decrease. + pkey.events.incl(Event.Finished) + rkey.events.incl(Event.Timer) + of EVFILT_VNODE: + pkey = addr(s.fds[int(kevent.ident)]) + rkey.events.incl(Event.Vnode) + if (kevent.fflags and NOTE_DELETE) != 0: + rkey.events.incl(Event.VnodeDelete) + if (kevent.fflags and NOTE_WRITE) != 0: + rkey.events.incl(Event.VnodeWrite) + if (kevent.fflags and NOTE_EXTEND) != 0: + rkey.events.incl(Event.VnodeExtend) + if (kevent.fflags and NOTE_ATTRIB) != 0: + rkey.events.incl(Event.VnodeAttrib) + if (kevent.fflags and NOTE_LINK) != 0: + rkey.events.incl(Event.VnodeLink) + if (kevent.fflags and NOTE_RENAME) != 0: + rkey.events.incl(Event.VnodeRename) + if (kevent.fflags and NOTE_REVOKE) != 0: + rkey.events.incl(Event.VnodeRevoke) + of EVFILT_SIGNAL: + pkey = addr(s.fds[cast[int](kevent.udata)]) + rkey.fd = cast[int](kevent.udata) + rkey.events.incl(Event.Signal) + of EVFILT_PROC: + rkey.fd = cast[int](kevent.udata) + pkey = addr(s.fds[cast[int](kevent.udata)]) + # we will not clear key, until it will be unregistered, so + # application can obtain data, but we will decrease counter, + # because kqueue is empty. + dec(s.count) + # we are marking key with `Finished` event, to avoid double decrease. + pkey.events.incl(Event.Finished) + rkey.events.incl(Event.Process) + else: + doAssert(true, "Unsupported kqueue filter in the queue!") + + if (kevent.flags and EV_EOF) != 0: + # TODO this error handling needs to be rethought. + # `fflags` can sometimes be `0x80000000` and thus we use 'cast' + # here: + if kevent.fflags != 0: + rkey.errorCode = cast[OSErrorCode](kevent.fflags) + else: + # This assumes we are dealing with sockets. + # TODO: For future-proofing it might be a good idea to give the + # user access to the raw `kevent`. + rkey.errorCode = OSErrorCode(ECONNRESET) + rkey.events.incl(Event.Error) + + results[k] = rkey + inc(k) + inc(i) + result = k + +proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] = + result = newSeq[ReadyKey](MAX_KQUEUE_EVENTS) + let count = selectInto(s, timeout, result) + result.setLen(count) + +template isEmpty*[T](s: Selector[T]): bool = + (s.count == 0) + +proc contains*[T](s: Selector[T], fd: SocketHandle|int): bool {.inline.} = + return s.fds[fd.int].ident != InvalidIdent + +proc getData*[T](s: Selector[T], fd: SocketHandle|int): var T = + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + result = s.fds[fdi].data + +proc setData*[T](s: Selector[T], fd: SocketHandle|int, data: T): bool = + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + s.fds[fdi].data = data + result = true + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, + body: untyped) = + mixin checkFd + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + var value = addr(s.fds[fdi].data) + body + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1, + body2: untyped) = + mixin checkFd + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + var value = addr(s.fds[fdi].data) + body1 + else: + body2 + + +proc getFd*[T](s: Selector[T]): int = + return s.kqFD.int diff --git a/lib/pure/ioselects/ioselectors_poll.nim b/lib/pure/ioselects/ioselectors_poll.nim new file mode 100644 index 000000000..7c5347156 --- /dev/null +++ b/lib/pure/ioselects/ioselectors_poll.nim @@ -0,0 +1,324 @@ +# +# +# 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 implements Posix poll(). + +import std/[posix, times] + +# Maximum number of events that can be returned +const MAX_POLL_EVENTS = 64 + +const hasEventFds = defined(zephyr) or defined(nimPollHasEventFds) + +when hasEventFds: + proc eventfd(count: cuint, flags: cint): cint + {.cdecl, importc: "eventfd", header: "<sys/eventfd.h>".} + +when hasThreadSupport: + type + SelectorImpl[T] = object + maxFD : int + pollcnt: int + fds: ptr SharedArray[SelectorKey[T]] + pollfds: ptr SharedArray[TPollFd] + count*: int + lock: Lock + Selector*[T] = ptr SelectorImpl[T] +else: + type + SelectorImpl[T] = object + maxFD : int + pollcnt: int + fds: seq[SelectorKey[T]] + pollfds: seq[TPollFd] + count*: int + Selector*[T] = ref SelectorImpl[T] + +type + 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 = maxDescriptors() + + when hasThreadSupport: + result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) + result.maxFD = maxFD + result.fds = allocSharedArray[SelectorKey[T]](maxFD) + result.pollfds = allocSharedArray[TPollFd](maxFD) + initLock(result.lock) + else: + result = Selector[T]() + result.maxFD = maxFD + result.fds = newSeq[SelectorKey[T]](maxFD) + result.pollfds = newSeq[TPollFd](maxFD) + + for i in 0 ..< maxFD: + result.fds[i].ident = InvalidIdent + +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) + doAssert(i < s.pollcnt, + "Descriptor [" & $sock & "] is not registered in the 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) + +template checkFd(s, f) = + if f >= s.maxFD: + raiseIOSelectorsError("Maximum number of descriptors is exhausted!") + +proc registerHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event], data: T) = + var fdi = int(fd) + s.checkFd(fdi) + doAssert(s.fds[fdi].ident == InvalidIdent) + setKey(s, fdi, events, 0, data) + if events != {}: s.pollAdd(fdi.cint, events) + +proc updateHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event]) = + let maskEvents = {Event.Timer, Event.Signal, Event.Process, Event.Vnode, + Event.User, Event.Oneshot, Event.Error} + let fdi = int(fd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, + "Descriptor [" & $fdi & "] is not registered in the queue!") + doAssert(pkey.events * maskEvents == {}) + + if pkey.events != events: + if pkey.events == {}: + s.pollAdd(fd.cint, events) + else: + if events != {}: + s.pollUpdate(fd.cint, events) + else: + s.pollRemove(fd.cint) + pkey.events = events + +proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = + var fdi = int(ev.rfd) + doAssert(s.fds[fdi].ident == InvalidIdent, "Event is already registered in the queue!") + var events = {Event.User} + setKey(s, fdi, events, 0, data) + events.incl(Event.Read) + s.pollAdd(fdi.cint, events) + +proc unregister*[T](s: Selector[T], fd: int|SocketHandle) = + let fdi = int(fd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, + "Descriptor [" & $fdi & "] is not registered in the queue!") + pkey.ident = InvalidIdent + if pkey.events != {}: + pkey.events = {} + s.pollRemove(fdi.cint) + +proc unregister*[T](s: Selector[T], ev: SelectEvent) = + let fdi = int(ev.rfd) + s.checkFd(fdi) + var pkey = addr(s.fds[fdi]) + doAssert(pkey.ident != InvalidIdent, "Event is not registered in the queue!") + doAssert(Event.User in pkey.events) + pkey.ident = InvalidIdent + pkey.events = {} + s.pollRemove(fdi.cint) + +proc newSelectEvent*(): SelectEvent = + when not hasEventFds: + var fds: array[2, cint] + if posix.pipe(fds) != 0: + raiseIOSelectorsError(osLastError()) + setNonBlocking(fds[0]) + setNonBlocking(fds[1]) + result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) + result.rfd = fds[0] + result.wfd = fds[1] + else: + let fdci = eventfd(0, posix.O_NONBLOCK) + if fdci == -1: + raiseIOSelectorsError(osLastError()) + result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) + result.rfd = fdci + result.wfd = fdci + +proc trigger*(ev: SelectEvent) = + var data: uint64 = 1 + if posix.write(ev.wfd, addr data, sizeof(uint64)) != sizeof(uint64): + raiseIOSelectorsError(osLastError()) + +proc close*(ev: SelectEvent) = + let res1 = posix.close(ev.rfd) + let res2 = + when hasEventFds: 0 + else: posix.close(ev.wfd) + + deallocShared(cast[pointer](ev)) + if res1 != 0 or res2 != 0: + raiseIOSelectorsError(osLastError()) + +proc selectInto*[T](s: Selector[T], timeout: int, + results: var openArray[ReadyKey]): int = + var maxres = MAX_POLL_EVENTS + if maxres > len(results): + maxres = len(results) + + verifySelectParams(timeout) + + s.withPollLock(): + let count = posix.poll(addr(s.pollfds[0]), Tnfds(s.pollcnt), timeout) + if count < 0: + result = 0 + let err = osLastError() + if cint(err) != EINTR: + raiseIOSelectorsError(err) + elif count == 0: + result = 0 + else: + var i = 0 + var k = 0 + var rindex = 0 + while (i < s.pollcnt) and (k < count) and (rindex < maxres): + let revents = s.pollfds[i].revents + if revents != 0: + let fd = s.pollfds[i].fd + var pkey = addr(s.fds[fd]) + var rkey = ReadyKey(fd: int(fd), events: {}) + + if (revents and POLLIN) != 0: + rkey.events.incl(Event.Read) + if Event.User in pkey.events: + var data: uint64 = 0 + if posix.read(fd, addr data, sizeof(uint64)) != sizeof(uint64): + let err = osLastError() + if err != OSErrorCode(EAGAIN): + raiseIOSelectorsError(err) + else: + # someone already consumed event data + inc(i) + continue + rkey.events = {Event.User} + if (revents and POLLOUT) != 0: + rkey.events.incl(Event.Write) + if (revents and POLLERR) != 0 or (revents and POLLHUP) != 0 or + (revents and POLLNVAL) != 0: + rkey.events.incl(Event.Error) + results[rindex] = rkey + s.pollfds[i].revents = 0 + inc(rindex) + inc(k) + inc(i) + result = k + +proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] = + result = newSeq[ReadyKey](MAX_POLL_EVENTS) + let count = selectInto(s, timeout, result) + result.setLen(count) + +template isEmpty*[T](s: Selector[T]): bool = + (s.count == 0) + +proc contains*[T](s: Selector[T], fd: SocketHandle|int): bool {.inline.} = + return s.fds[fd.int].ident != InvalidIdent + +proc getData*[T](s: Selector[T], fd: SocketHandle|int): var T = + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + result = s.fds[fdi].data + +proc setData*[T](s: Selector[T], fd: SocketHandle|int, data: T): bool = + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + s.fds[fdi].data = data + result = true + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, + body: untyped) = + mixin checkFd + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + var value = addr(s.getData(fdi)) + body + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1, + body2: untyped) = + mixin checkFd + let fdi = int(fd) + s.checkFd(fdi) + if fdi in s: + var value = addr(s.getData(fdi)) + body1 + else: + body2 + + +proc getFd*[T](s: Selector[T]): int = + return -1 diff --git a/lib/pure/ioselects/ioselectors_select.nim b/lib/pure/ioselects/ioselectors_select.nim new file mode 100644 index 000000000..6c516395b --- /dev/null +++ b/lib/pure/ioselects/ioselectors_select.nim @@ -0,0 +1,454 @@ +# +# +# 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 implements Posix and Windows select(). + +import std/[times, nativesockets] + +when defined(windows): + import std/winlean + when defined(gcc): + {.passl: "-lws2_32".} + elif defined(vcc): + {.passl: "ws2_32.lib".} + const platformHeaders = """#include <winsock2.h> + #include <windows.h>""" + const EAGAIN = WSAEWOULDBLOCK +else: + const platformHeaders = """#include <sys/select.h> + #include <sys/time.h> + #include <sys/types.h> + #include <unistd.h>""" +type + FdSet {.importc: "fd_set", header: platformHeaders, pure, final.} = object +var + FD_SETSIZE {.importc: "FD_SETSIZE", header: platformHeaders.}: cint + +proc IOFD_SET(fd: SocketHandle, fdset: ptr FdSet) + {.cdecl, importc: "FD_SET", header: platformHeaders, inline.} +proc IOFD_CLR(fd: SocketHandle, fdset: ptr FdSet) + {.cdecl, importc: "FD_CLR", header: platformHeaders, inline.} +proc IOFD_ZERO(fdset: ptr FdSet) + {.cdecl, importc: "FD_ZERO", header: platformHeaders, inline.} + +when defined(windows): + proc IOFD_ISSET(fd: SocketHandle, fdset: ptr FdSet): cint + {.stdcall, importc: "FD_ISSET", header: platformHeaders, inline.} + proc ioselect(nfds: cint, readFds, writeFds, exceptFds: ptr FdSet, + timeout: ptr Timeval): cint + {.stdcall, importc: "select", header: platformHeaders.} +else: + proc IOFD_ISSET(fd: SocketHandle, fdset: ptr FdSet): cint + {.cdecl, importc: "FD_ISSET", header: platformHeaders, inline.} + proc ioselect(nfds: cint, readFds, writeFds, exceptFds: ptr FdSet, + timeout: ptr Timeval): cint + {.cdecl, importc: "select", header: platformHeaders.} + +when hasThreadSupport: + type + SelectorImpl[T] = object + rSet: FdSet + wSet: FdSet + eSet: FdSet + maxFD: int + fds: ptr SharedArray[SelectorKey[T]] + count*: int + lock: Lock + Selector*[T] = ptr SelectorImpl[T] +else: + type + SelectorImpl[T] = object + rSet: FdSet + wSet: FdSet + eSet: FdSet + maxFD: int + fds: seq[SelectorKey[T]] + count*: int + Selector*[T] = ref SelectorImpl[T] + +type + SelectEventImpl = object + rsock: SocketHandle + wsock: SocketHandle + 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] = + when hasThreadSupport: + result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) + result.fds = allocSharedArray[SelectorKey[T]](FD_SETSIZE) + initLock result.lock + else: + result = Selector[T]() + result.fds = newSeq[SelectorKey[T]](FD_SETSIZE) + + for i in 0 ..< FD_SETSIZE: + result.fds[i].ident = InvalidIdent + + IOFD_ZERO(addr result.rSet) + IOFD_ZERO(addr result.wSet) + IOFD_ZERO(addr result.eSet) + +proc close*[T](s: Selector[T]) = + when hasThreadSupport: + deallocSharedArray(s.fds) + deallocShared(cast[pointer](s)) + deinitLock(s.lock) + +when defined(windows): + proc newSelectEvent*(): SelectEvent = + var ssock = createNativeSocket() + var wsock = createNativeSocket() + var rsock: SocketHandle = INVALID_SOCKET + var saddr = Sockaddr_in() + + 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: + raiseIOSelectorsError(osLastError()) + + if winlean.listen(ssock, 1) != 0: + raiseIOSelectorsError(osLastError()) + + var namelen = sizeof(saddr).SockLen + if getsockname(ssock, cast[ptr SockAddr](addr(saddr)), + addr(namelen)) != 0'i32: + raiseIOSelectorsError(osLastError()) + + saddr.sin_addr.s_addr = 0x0100007F + if winlean.connect(wsock, cast[ptr SockAddr](addr(saddr)), + sizeof(saddr).SockLen) != 0: + raiseIOSelectorsError(osLastError()) + namelen = sizeof(saddr).SockLen + rsock = winlean.accept(ssock, cast[ptr SockAddr](addr(saddr)), + cast[ptr SockLen](addr(namelen))) + if rsock == SocketHandle(-1): + raiseIOSelectorsError(osLastError()) + + if winlean.closesocket(ssock) != 0: + raiseIOSelectorsError(osLastError()) + + var mode = clong(1) + if ioctlsocket(rsock, FIONBIO, addr(mode)) != 0: + raiseIOSelectorsError(osLastError()) + mode = clong(1) + if ioctlsocket(wsock, FIONBIO, addr(mode)) != 0: + raiseIOSelectorsError(osLastError()) + + result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) + result.rsock = rsock + result.wsock = wsock + + proc trigger*(ev: SelectEvent) = + var data: uint64 = 1 + if winlean.send(ev.wsock, cast[pointer](addr data), + cint(sizeof(uint64)), 0) != sizeof(uint64): + raiseIOSelectorsError(osLastError()) + + proc close*(ev: SelectEvent) = + let res1 = winlean.closesocket(ev.rsock) + let res2 = winlean.closesocket(ev.wsock) + deallocShared(cast[pointer](ev)) + if res1 != 0 or res2 != 0: + raiseIOSelectorsError(osLastError()) + +else: + proc newSelectEvent*(): SelectEvent = + var fds: array[2, cint] + if posix.pipe(fds) != 0: + raiseIOSelectorsError(osLastError()) + setNonBlocking(fds[0]) + setNonBlocking(fds[1]) + result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) + result.rsock = SocketHandle(fds[0]) + result.wsock = SocketHandle(fds[1]) + + proc trigger*(ev: SelectEvent) = + var data: uint64 = 1 + if posix.write(cint(ev.wsock), addr data, sizeof(uint64)) != sizeof(uint64): + raiseIOSelectorsError(osLastError()) + + proc close*(ev: SelectEvent) = + let res1 = posix.close(cint(ev.rsock)) + let res2 = posix.close(cint(ev.wsock)) + deallocShared(cast[pointer](ev)) + if res1 != 0 or res2 != 0: + raiseIOSelectorsError(osLastError()) + +proc setSelectKey[T](s: Selector[T], fd: SocketHandle, events: set[Event], + data: T) = + var i = 0 + let fdi = int(fd) + while i < FD_SETSIZE: + if s.fds[i].ident == InvalidIdent: + var pkey = addr(s.fds[i]) + pkey.ident = fdi + pkey.events = events + pkey.data = data + break + inc(i) + if i >= FD_SETSIZE: + raiseIOSelectorsError("Maximum number of descriptors is exhausted!") + +proc getKey[T](s: Selector[T], fd: SocketHandle): ptr SelectorKey[T] = + var i = 0 + let fdi = int(fd) + while i < FD_SETSIZE: + if s.fds[i].ident == fdi: + result = addr(s.fds[i]) + break + inc(i) + doAssert(i < FD_SETSIZE, + "Descriptor [" & $int(fd) & "] is not registered in the queue!") + +proc delKey[T](s: Selector[T], fd: SocketHandle) = + var empty: T + var i = 0 + while i < FD_SETSIZE: + if s.fds[i].ident == fd.int: + s.fds[i].ident = InvalidIdent + s.fds[i].events = {} + s.fds[i].data = empty + break + inc(i) + doAssert(i < FD_SETSIZE, + "Descriptor [" & $int(fd) & "] is not registered in the queue!") + +proc registerHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event], data: T) = + when not defined(windows): + let fdi = int(fd) + s.withSelectLock(): + s.setSelectKey(fd, events, data) + when not defined(windows): + if fdi > s.maxFD: s.maxFD = fdi + if Event.Read in events: + IOFD_SET(fd, addr s.rSet) + inc(s.count) + if Event.Write in events: + IOFD_SET(fd, addr s.wSet) + IOFD_SET(fd, addr s.eSet) + inc(s.count) + +proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = + when not defined(windows): + let fdi = int(ev.rsock) + s.withSelectLock(): + s.setSelectKey(ev.rsock, {Event.User}, data) + when not defined(windows): + if fdi > s.maxFD: s.maxFD = fdi + IOFD_SET(ev.rsock, addr s.rSet) + inc(s.count) + +proc updateHandle*[T](s: Selector[T], fd: int | SocketHandle, + events: set[Event]) = + let maskEvents = {Event.Timer, Event.Signal, Event.Process, Event.Vnode, + Event.User, Event.Oneshot, Event.Error} + s.withSelectLock(): + var pkey = s.getKey(fd) + doAssert(pkey.events * maskEvents == {}) + if pkey.events != events: + if (Event.Read in pkey.events) and (Event.Read notin events): + IOFD_CLR(fd, addr s.rSet) + dec(s.count) + if (Event.Write in pkey.events) and (Event.Write notin events): + IOFD_CLR(fd, addr s.wSet) + IOFD_CLR(fd, addr s.eSet) + dec(s.count) + if (Event.Read notin pkey.events) and (Event.Read in events): + IOFD_SET(fd, addr s.rSet) + inc(s.count) + if (Event.Write notin pkey.events) and (Event.Write in events): + IOFD_SET(fd, addr s.wSet) + IOFD_SET(fd, addr s.eSet) + inc(s.count) + pkey.events = events + +proc unregister*[T](s: Selector[T], fd: SocketHandle|int) = + s.withSelectLock(): + let fd = fd.SocketHandle + var pkey = s.getKey(fd) + if Event.Read in pkey.events or Event.User in pkey.events: + IOFD_CLR(fd, addr s.rSet) + dec(s.count) + if Event.Write in pkey.events: + IOFD_CLR(fd, addr s.wSet) + IOFD_CLR(fd, addr s.eSet) + dec(s.count) + s.delKey(fd) + +proc unregister*[T](s: Selector[T], ev: SelectEvent) = + let fd = ev.rsock + s.withSelectLock(): + var pkey = s.getKey(fd) + IOFD_CLR(fd, addr s.rSet) + dec(s.count) + s.delKey(fd) + +proc selectInto*[T](s: Selector[T], timeout: int, + results: var openArray[ReadyKey]): int = + var tv = Timeval() + var ptv = addr tv + var rset, wset, eset: FdSet + + verifySelectParams(timeout) + + if timeout != -1: + when defined(genode) or defined(freertos) or defined(zephyr) or defined(nuttx): + tv.tv_sec = posix.Time(timeout div 1_000) + else: + 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 = ioselect(cint(s.maxFD) + 1, addr(rset), addr(wset), + addr(eset), ptv) + if count < 0: + result = 0 + when defined(windows): + raiseIOSelectorsError(osLastError()) + else: + let err = osLastError() + if cint(err) != EINTR: + raiseIOSelectorsError(err) + elif count == 0: + result = 0 + else: + var rindex = 0 + var i = 0 + var k = 0 + + while (i < FD_SETSIZE) and (k < count): + if s.fds[i].ident != InvalidIdent: + var flag = false + var pkey = addr(s.fds[i]) + var rkey = ReadyKey(fd: int(pkey.ident), events: {}) + let fd = SocketHandle(pkey.ident) + if IOFD_ISSET(fd, addr rset) != 0: + if Event.User in pkey.events: + var data: uint64 = 0 + if recv(fd, cast[pointer](addr(data)), + sizeof(uint64).cint, 0) != sizeof(uint64): + let err = osLastError() + if cint(err) != EAGAIN: + raiseIOSelectorsError(err) + else: + inc(i) + inc(k) + continue + else: + flag = true + rkey.events = {Event.User} + else: + flag = true + rkey.events = {Event.Read} + if IOFD_ISSET(fd, addr wset) != 0: + rkey.events.incl(Event.Write) + if IOFD_ISSET(fd, addr eset) != 0: + rkey.events.incl(Event.Error) + flag = true + if flag: + results[rindex] = rkey + inc(rindex) + inc(k) + inc(i) + result = rindex + +proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] = + result = newSeq[ReadyKey](FD_SETSIZE) + var count = selectInto(s, timeout, result) + result.setLen(count) + +proc flush*[T](s: Selector[T]) = discard + +template isEmpty*[T](s: Selector[T]): bool = + (s.count == 0) + +proc contains*[T](s: Selector[T], fd: SocketHandle|int): bool {.inline.} = + s.withSelectLock(): + result = false + + let fdi = int(fd) + for i in 0..<FD_SETSIZE: + if s.fds[i].ident == fdi: + return true + +proc getData*[T](s: Selector[T], fd: SocketHandle|int): var T = + s.withSelectLock(): + let fdi = int(fd) + for i in 0..<FD_SETSIZE: + if s.fds[i].ident == fdi: + return s.fds[i].data + +proc setData*[T](s: Selector[T], fd: SocketHandle|int, data: T): bool = + s.withSelectLock(): + let fdi = int(fd) + var i = 0 + while i < FD_SETSIZE: + if s.fds[i].ident == fdi: + var pkey = addr(s.fds[i]) + pkey.data = data + result = true + break + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, + body: untyped) = + mixin withSelectLock + s.withSelectLock(): + var value: ptr T + let fdi = int(fd) + var i = 0 + while i < FD_SETSIZE: + if s.fds[i].ident == fdi: + value = addr(s.fds[i].data) + break + inc(i) + if i != FD_SETSIZE: + body + +template withData*[T](s: Selector[T], fd: SocketHandle|int, value, + body1, body2: untyped) = + mixin withSelectLock + s.withSelectLock(): + block: + var value: ptr T + let fdi = int(fd) + var i = 0 + while i < FD_SETSIZE: + if s.fds[i].ident == fdi: + value = addr(s.fds[i].data) + break + inc(i) + if i != FD_SETSIZE: + body1 + else: + body2 + + +proc getFd*[T](s: Selector[T]): int = + return -1 |