jyapayne/Nim
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Event enum is now .pure.

Browse source 
Modified getMaxFds() to work more properly.
Removed seqs from code
Some exceptions replaced with doAssert
This commit is contained in:
cheatfate 2016-06-26 09:51:01 +03:00
commit 2eb34a8129
2 changed files with 377 additions and 446 deletions
Download patch file Download diff file Expand all files Collapse all files

435
lib/pure/ioselectors.nim
View file

@ -50,16 +50,16 @@ when defined(nimdoc):
Selector*[T] = ref object Selector*[T] = ref object
## An object which holds descriptors to be checked for read/write status ## An object which holds descriptors to be checked for read/write status
Event* = enum Event* {.pure.} = enum
## An enum which hold event types ## An enum which hold event types
eventRead, ## Descriptor is available for read Read, ## Descriptor is available for read
eventWrite, ## Descriptor is available for write Write, ## Descriptor is available for write
eventTimer, ## Timer descriptor is completed Timer, ## Timer descriptor is completed
eventSignal, ## Signal is raised Signal, ## Signal is raised
eventProcess, ## Process is finished Process, ## Process is finished
eventVnode, ## Currently not supported Vnode, ## Currently not supported
eventUser, ## User event is raised User, ## User event is raised
eventError ## Error happens while waiting, for descriptor Error ## Error happens while waiting, for descriptor
ReadyKey*[T] = object ReadyKey*[T] = object
## An object which holds result for descriptor ## An object which holds result for descriptor
@ -189,10 +189,30 @@ when defined(nimdoc):
## ##
else: else:
when not defined(windows): when defined(macosx) or defined(freebsd):
when defined(macosx): when defined(macosx):
var const maxDescriptors = 29 # KERN_MAXFILESPERPROC (MacOS)
OPEN_MAX {.importc: "OPEN_MAX", header: "<sys/resource.h>".}: cint 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 var
RLIMIT_NOFILE {.importc: "RLIMIT_NOFILE", RLIMIT_NOFILE {.importc: "RLIMIT_NOFILE",
header: "<sys/resource.h>".}: cint header: "<sys/resource.h>".}: cint
@ -201,24 +221,36 @@ else:
header: "<sys/resource.h>", pure, final.} = object header: "<sys/resource.h>", pure, final.} = object
rlim_cur: int rlim_cur: int
rlim_max: int rlim_max: int
proc getrlimit(resource: cint, rlp: var rlimit): cint {. proc getrlimit(resource: cint, rlp: var rlimit): cint
importc: "getrlimit",header: "<sys/resource.h>"} {.importc: "getrlimit",header: "<sys/resource.h>".}
proc getMaxFds*(): int = 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() var a = rlimit()
if getrlimit(RLIMIT_NOFILE, a) != 0: if getrlimit(RLIMIT_NOFILE, a) != 0:
raiseOsError(osLastError()) raiseOsError(osLastError())
result = a.rlim_max result = a.rlim_max
when defined(macosx):
if a.rlim_max > OPEN_MAX:
result = OPEN_MAX
when hasThreadSupport: when hasThreadSupport:
import locks import locks
type type
Event* = enum Event* {.pure.} = enum
eventRead, eventWrite, eventTimer, eventSignal, eventProcess, Read, Write, Timer, Signal, Process, Vnode, User, Error,
eventVnode, eventUser, eventError,
flagHandle, flagTimer, flagSignal, flagProcess, flagVnode, flagUser, flagHandle, flagTimer, flagSignal, flagProcess, flagVnode, flagUser,
flagOneshot flagOneshot
@ -234,7 +266,6 @@ else:
key : ReadyKey[T] key : ReadyKey[T]
when not defined(windows): when not defined(windows):
when hasThreadSupport:
type type
SharedArrayHolder[T] = object SharedArrayHolder[T] = object
part: array[16, T] part: array[16, T]
@ -298,7 +329,6 @@ else:
MAX_KQUEUE_CHANGE_EVENTS = 64 MAX_KQUEUE_CHANGE_EVENTS = 64
MAX_KQUEUE_RESULT_EVENTS = 64 MAX_KQUEUE_RESULT_EVENTS = 64
when hasThreadSupport:
type type
SelectorImpl[T] = object SelectorImpl[T] = object
kqFD : cint kqFD : cint
@ -307,21 +337,9 @@ else:
changesCount: int changesCount: int
fds: ptr SharedArray[SelectorKey[T]] fds: ptr SharedArray[SelectorKey[T]]
count: int count: int
changesLock: Lock
else:
type
SelectorImpl[T] = object
kqFD : cint
maxFD : uint
changesTable: array[MAX_KQUEUE_CHANGE_EVENTS, KEvent]
changesCount: int
fds: seq[SelectorKey[T]]
count: int
when hasThreadSupport: when hasThreadSupport:
type Selector*[T] = ptr SelectorImpl[T] changesLock: Lock
else: Selector*[T] = ptr SelectorImpl[T]
type Selector*[T] = ref SelectorImpl[T]
type type
SelectEventImpl = object SelectEventImpl = object
@ -336,15 +354,13 @@ else:
var kqFD = kqueue() var kqFD = kqueue()
if kqFD < 0: if kqFD < 0:
raiseOsError(osLastError()) raiseOsError(osLastError())
when hasThreadSupport:
result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T])))
result.kqFD = kqFD result.kqFD = kqFD
result.maxFD = maxFD.uint result.maxFD = maxFD.uint
result.fds = allocSharedArray[SelectorKey[T]](maxFD) result.fds = allocSharedArray[SelectorKey[T]](maxFD)
when hasThreadSupport:
initLock(result.changesLock) initLock(result.changesLock)
else:
result = Selector[T](kqFD: kqFD, maxFD: maxFD.uint)
result.fds = newSeq[SelectorKey[T]](maxFD)
proc close*[T](s: Selector[T]) = proc close*[T](s: Selector[T]) =
if posix.close(s.kqFD) != 0: if posix.close(s.kqFD) != 0:
@ -386,17 +402,15 @@ else:
events: set[Event], data: T) = events: set[Event], data: T) =
var fdi = int(fd) var fdi = int(fd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
setKey(s, fdi, fdi, {flagHandle} + events, 0, data) setKey(s, fdi, fdi, {Event.flagHandle} + events, 0, data)
if events != {}: if events != {}:
if eventRead in events: if Event.Read in events:
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
if eventWrite in events: if Event.Write in events:
modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -404,30 +418,24 @@ else:
events: set[Event]) = events: set[Event]) =
var fdi = int(fd) var fdi = int(fd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident != 0: doAssert(s.fds[fdi].ident != 0)
if flagHandle in s.fds[fdi].flags: doAssert(Event.flagHandle in s.fds[fdi].flags)
var ne = events + {flagHandle} var ne = events + {Event.flagHandle}
var oe = s.fds[fdi].flags var oe = s.fds[fdi].flags
if oe != ne: if oe != ne:
if (eventRead in oe) and (eventRead notin ne): if (Event.Read in oe) and (Event.Read notin ne):
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
if (eventWrite in oe) and (eventWrite notin ne): if (Event.Write in oe) and (Event.Write notin ne):
modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
if (eventRead notin oe) and (eventRead in ne): if (Event.Read notin oe) and (Event.Read in ne):
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
if (eventWrite notin oe) and (eventWrite in ne): if (Event.Write notin oe) and (Event.Write in ne):
modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
s.fds[fdi].flags = ne s.fds[fdi].flags = ne
else:
raise newException(ValueError,
"Could not update non-handle descriptor")
else:
raise newException(ValueError,
"Descriptor is not registered in queue")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -438,9 +446,9 @@ else:
if fdi == -1: if fdi == -1:
raiseOsError(osLastError()) raiseOsError(osLastError())
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
var mflags = if oneshot: {flagTimer, flagOneshot} var mflags = if oneshot: {Event.flagTimer, Event.flagOneshot}
else: {flagTimer} else: {Event.flagTimer}
var kflags: cushort = if oneshot: EV_ONESHOT or EV_ADD var kflags: cushort = if oneshot: EV_ONESHOT or EV_ADD
else: EV_ADD else: EV_ADD
setKey(s, fdi, fdi, mflags, 0, data) setKey(s, fdi, fdi, mflags, 0, data)
@ -450,8 +458,6 @@ else:
modifyKQueue(s, fdi.uint, EVFILT_TIMER, kflags, 0, cint(timeout), nil) modifyKQueue(s, fdi.uint, EVFILT_TIMER, kflags, 0, cint(timeout), nil)
inc(s.count) inc(s.count)
result = fdi result = fdi
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -463,8 +469,8 @@ else:
raiseOsError(osLastError()) raiseOsError(osLastError())
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
setKey(s, fdi, signal, {flagSignal}, signal, data) setKey(s, fdi, signal, {Event.flagSignal}, signal, data)
# block signal `signal` # block signal `signal`
var nmask: Sigset var nmask: Sigset
var omask: Sigset var omask: Sigset
@ -478,8 +484,6 @@ else:
cast[pointer](fdi)) cast[pointer](fdi))
inc(s.count) inc(s.count)
result = fdi result = fdi
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -491,15 +495,13 @@ else:
raiseOsError(osLastError()) raiseOsError(osLastError())
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
var kflags: cushort = EV_ONESHOT or EV_ADD var kflags: cushort = EV_ONESHOT or EV_ADD
setKey(s, fdi, pid, {flagProcess, flagOneshot}, pid, data) setKey(s, fdi, pid, {Event.flagProcess, Event.flagOneshot}, pid, data)
modifyKQueue(s, pid.uint, EVFILT_PROC, kflags, NOTE_EXIT, 0, modifyKQueue(s, pid.uint, EVFILT_PROC, kflags, NOTE_EXIT, 0,
cast[pointer](fdi)) cast[pointer](fdi))
inc(s.count) inc(s.count)
result = fdi result = fdi
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -509,22 +511,22 @@ else:
var flags = s.fds[fdi].flags var flags = s.fds[fdi].flags
var filter: cshort = 0 var filter: cshort = 0
if s.fds[fdi].ident != 0 and flags != {}: if s.fds[fdi].ident != 0 and flags != {}:
if flagHandle in flags: if Event.flagHandle in flags:
# if events == 0, than descriptor was modified with # if events == 0, than descriptor was modified with
# updateHandle(fd, 0), so it was already deleted from kqueue. # updateHandle(fd, 0), so it was already deleted from kqueue.
if flags != {flagHandle}: if flags != {Event.flagHandle}:
if eventRead in flags: if Event.Read in flags:
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
if eventWrite in flags: if Event.Write in flags:
modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
elif flagTimer in flags: elif Event.flagTimer in flags:
filter = EVFILT_TIMER filter = EVFILT_TIMER
discard posix.close(cint(s.fds[fdi].key.fd)) discard posix.close(cint(s.fds[fdi].key.fd))
modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
elif flagSignal in flags: elif Event.flagSignal in flags:
filter = EVFILT_SIGNAL filter = EVFILT_SIGNAL
# unblocking signal # unblocking signal
var nmask = Sigset() var nmask = Sigset()
@ -536,12 +538,12 @@ else:
discard posix.close(cint(s.fds[fdi].key.fd)) discard posix.close(cint(s.fds[fdi].key.fd))
modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
elif flagProcess in flags: elif Event.flagProcess in flags:
filter = EVFILT_PROC filter = EVFILT_PROC
discard posix.close(cint(s.fds[fdi].key.fd)) discard posix.close(cint(s.fds[fdi].key.fd))
modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
elif flagUser in flags: elif Event.flagUser in flags:
filter = EVFILT_READ filter = EVFILT_READ
modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil) modifyKQueue(s, fdi.uint, filter, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
@ -583,15 +585,10 @@ else:
proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) =
let fdi = ev.rfd.int let fdi = ev.rfd.int
if fdi.uint < s.maxFD: doAssert(s.fds[fdi].ident == 0)
if s.fds[fdi].ident == 0 and s.fds[fdi].flags == {}: setKey(s, fdi, fdi, {Event.flagUser}, 0, data)
setKey(s, fdi, fdi, {flagUser}, 0, data)
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else:
raise newException(ValueError, "Event wait still pending!")
proc unregister*[T](s: Selector[T], ev: SelectEvent) = proc unregister*[T](s: Selector[T], ev: SelectEvent) =
let fdi = ev.rfd.int let fdi = ev.rfd.int
@ -640,9 +637,9 @@ else:
case kevent.filter case kevent.filter
of EVFILT_READ: of EVFILT_READ:
skey = addr(s.fds[kevent.ident.int]) skey = addr(s.fds[kevent.ident.int])
if flagHandle in skey.flags: if Event.flagHandle in skey.flags:
events = {eventRead} events = {Event.Read}
elif flagUser in skey.flags: elif Event.flagUser in skey.flags:
var data: int = 0 var data: int = 0
if posix.read(kevent.ident.cint, addr data, if posix.read(kevent.ident.cint, addr data,
sizeof(int)) != sizeof(int): sizeof(int)) != sizeof(int):
@ -653,28 +650,28 @@ else:
continue continue
else: else:
raiseOSError(osLastError()) raiseOSError(osLastError())
events = {eventUser} events = {Event.User}
else: else:
events = {eventRead} events = {Event.Read}
of EVFILT_WRITE: of EVFILT_WRITE:
skey = addr(s.fds[kevent.ident.int]) skey = addr(s.fds[kevent.ident.int])
events = {eventWrite} events = {Event.Write}
of EVFILT_TIMER: of EVFILT_TIMER:
skey = addr(s.fds[kevent.ident.int]) skey = addr(s.fds[kevent.ident.int])
if flagOneshot in skey.flags: if Event.flagOneshot in skey.flags:
if posix.close(skey.ident.cint) == -1: if posix.close(skey.ident.cint) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
clearKey(s, skey.ident) clearKey(s, skey.ident)
# no need to modify kqueue, because EV_ONESHOT is already made # no need to modify kqueue, because EV_ONESHOT is already made
# this for us # this for us
dec(s.count) dec(s.count)
events = {eventTimer} events = {Event.Timer}
of EVFILT_VNODE: of EVFILT_VNODE:
skey = addr(s.fds[kevent.ident.int]) skey = addr(s.fds[kevent.ident.int])
events = {eventVnode} events = {Event.Vnode}
of EVFILT_SIGNAL: of EVFILT_SIGNAL:
skey = addr(s.fds[cast[int](kevent.udata)]) skey = addr(s.fds[cast[int](kevent.udata)])
events = {eventSignal} events = {Event.Signal}
of EVFILT_PROC: of EVFILT_PROC:
skey = addr(s.fds[cast[int](kevent.udata)]) skey = addr(s.fds[cast[int](kevent.udata)])
if posix.close(skey.ident.cint) == -1: if posix.close(skey.ident.cint) == -1:
@ -683,13 +680,13 @@ else:
# no need to modify kqueue, because EV_ONESHOT is already made # no need to modify kqueue, because EV_ONESHOT is already made
# this for us # this for us
dec(s.count) dec(s.count)
events = {eventProcess} events = {Event.Process}
else: else:
raise newException(ValueError, raise newException(ValueError,
"Unsupported kqueue filter in queue") "Unsupported kqueue filter in queue")
if (kevent.flags and EV_EOF) != 0: if (kevent.flags and EV_EOF) != 0:
events = events + {eventError} events = events + {Event.Error}
results[k].fd = skey.key.fd results[k].fd = skey.key.fd
results[k].events = events results[k].events = events
results[k].data = skey.key.data results[k].data = skey.key.data
@ -775,50 +772,35 @@ else:
proc eventfd(count: cuint, flags: cint): cint proc eventfd(count: cuint, flags: cint): cint
{.cdecl, importc: "eventfd", header: "<sys/eventfd.h>".} {.cdecl, importc: "eventfd", header: "<sys/eventfd.h>".}
when hasThreadSupport:
type type
SelectorImpl[T] = object SelectorImpl[T] = object
epollFD : cint epollFD : cint
maxFD : uint maxFD : uint
fds: ptr SharedArray[SelectorKey[T]] fds: ptr SharedArray[SelectorKey[T]]
count: int count: int
else:
type
SelectorImpl[T] = object
epollFD : cint
maxFD : uint
fds: seq[SelectorKey[T]]
count: int
when hasThreadSupport: Selector*[T] = ptr SelectorImpl[T]
type Selector*[T] = ptr SelectorImpl[T]
else:
type Selector*[T] = ref SelectorImpl[T]
type
SelectEventImpl = object SelectEventImpl = object
efd: cint efd: cint
type SelectEvent* = ptr SelectEventImpl SelectEvent* = ptr SelectEventImpl
proc newSelector*[T](): Selector[T] = proc newSelector*[T](): Selector[T] =
var maxFD = getMaxFds() var maxFD = getMaxFds()
var epollFD = epoll_create(MAX_EPOLL_RESULT_EVENTS) var epollFD = epoll_create(MAX_EPOLL_RESULT_EVENTS)
if epollFD < 0: if epollFD < 0:
raiseOsError(osLastError()) raiseOsError(osLastError())
when hasThreadSupport:
result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T])))
result.epollFD = epollFD result.epollFD = epollFD
result.maxFD = maxFD.uint result.maxFD = maxFD.uint
result.fds = allocSharedArray[SelectorKey[T]](maxFD) result.fds = allocSharedArray[SelectorKey[T]](maxFD)
else:
result = Selector[T](epollFD: epollFD, maxFD: maxFD.uint)
result.fds = newSeq[SelectorKey[T]](maxFD)
proc close*[T](s: Selector[T]) = proc close*[T](s: Selector[T]) =
if posix.close(s.epollFD) != 0: if posix.close(s.epollFD) != 0:
raiseOSError(osLastError()) raiseOSError(osLastError())
when hasThreadSupport:
deallocSharedArray(s.fds) deallocSharedArray(s.fds)
deallocShared(cast[pointer](s)) deallocShared(cast[pointer](s))
@ -826,21 +808,19 @@ else:
events: set[Event], data: T) = events: set[Event], data: T) =
var fdi = int(fd) var fdi = int(fd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
setKey(s, fdi, fdi, events + {flagHandle}, 0, data) setKey(s, fdi, fdi, events + {Event.flagHandle}, 0, data)
if events != {}: if events != {}:
var epv: epoll_event var epv: epoll_event
epv.events = EPOLLRDHUP epv.events = EPOLLRDHUP
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
if eventRead in events: if Event.Read in events:
epv.events = epv.events or EPOLLIN epv.events = epv.events or EPOLLIN
if eventWrite in events: if Event.Write in events:
epv.events = epv.events or EPOLLOUT epv.events = epv.events or EPOLLOUT
if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
inc(s.count) inc(s.count)
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -848,22 +828,22 @@ else:
events: set[Event]) = events: set[Event]) =
var fdi = int(fd) var fdi = int(fd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident != 0:
var oe = s.fds[fdi].flags var oe = s.fds[fdi].flags
if flagHandle in oe: doAssert(s.fds[fdi].ident != 0)
var ne = events + {flagHandle} doAssert(Event.flagHandle in oe)
var ne = events + {Event.flagHandle}
if oe != ne: if oe != ne:
var epv: epoll_event var epv: epoll_event
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
epv.events = EPOLLRDHUP epv.events = EPOLLRDHUP
if eventRead in events: if Event.Read in events:
epv.events = epv.events or EPOLLIN epv.events = epv.events or EPOLLIN
if eventWrite in events: if Event.Write in events:
epv.events = epv.events or EPOLLOUT epv.events = epv.events or EPOLLOUT
if oe == {flagHandle}: if oe == {Event.flagHandle}:
if ne != {flagHandle}: if ne != {Event.flagHandle}:
if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint,
addr epv) == -1: addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
@ -879,12 +859,6 @@ else:
raiseOSError(osLastError()) raiseOSError(osLastError())
dec(s.count) dec(s.count)
s.fds[fdi].flags = ne s.fds[fdi].flags = ne
else:
raise newException(ValueError,
"Could not update non-handle descriptor")
else:
raise newException(ValueError,
"Descriptor is not registered in queue")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -894,21 +868,21 @@ else:
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
var flags = s.fds[fdi].flags var flags = s.fds[fdi].flags
if s.fds[fdi].ident != 0 and flags != {}: if s.fds[fdi].ident != 0 and flags != {}:
if flagHandle in flags: if Event.flagHandle in flags:
# if events == {flagHandle}, then descriptor was already # if events == {flagHandle}, then descriptor was already
# unregistered from epoll with updateHandle() call. # unregistered from epoll with updateHandle() call.
# This check is done to omit EBADF error. # This check is done to omit EBADF error.
if flags != {flagHandle}: if flags != {Event.flagHandle}:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint,
addr epv) == -1: addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
dec(s.count) dec(s.count)
elif flagTimer in flags: elif Event.flagTimer in flags:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
discard posix.close(fdi.cint) discard posix.close(fdi.cint)
dec(s.count) dec(s.count)
elif flagSignal in flags: elif Event.flagSignal in flags:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
var nmask: Sigset var nmask: Sigset
@ -919,7 +893,7 @@ else:
unblockSignals(nmask, omask) unblockSignals(nmask, omask)
discard posix.close(fdi.cint) discard posix.close(fdi.cint)
dec(s.count) dec(s.count)
elif flagProcess in flags: elif Event.flagProcess in flags:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
var nmask: Sigset var nmask: Sigset
@ -935,7 +909,7 @@ else:
proc unregister*[T](s: Selector[T], ev: SelectEvent) = proc unregister*[T](s: Selector[T], ev: SelectEvent) =
let fdi = int(ev.efd) let fdi = int(ev.efd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident != 0 and flagUser in s.fds[fdi].flags: if s.fds[fdi].ident != 0 and (Event.flagUser in s.fds[fdi].flags):
clearKey(s, fdi) clearKey(s, fdi)
var epv: epoll_event var epv: epoll_event
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
@ -951,8 +925,8 @@ else:
if fdi == -1: if fdi == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
var flags = {flagTimer} var flags = {Event.flagTimer}
var epv: epoll_event var epv: epoll_event
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
epv.events = EPOLLIN or EPOLLRDHUP epv.events = EPOLLIN or EPOLLRDHUP
@ -962,7 +936,7 @@ else:
new_ts.it_interval.tv_nsec = 0 new_ts.it_interval.tv_nsec = 0
new_ts.it_value.tv_sec = (timeout div 1_000).Time new_ts.it_value.tv_sec = (timeout div 1_000).Time
new_ts.it_value.tv_nsec = (timeout %% 1_000) * 1_000_000 new_ts.it_value.tv_nsec = (timeout %% 1_000) * 1_000_000
flags = flags + {flagOneshot} flags = flags + {Event.flagOneshot}
epv.events = epv.events or EPOLLONESHOT epv.events = epv.events or EPOLLONESHOT
else: else:
new_ts.it_interval.tv_sec = (timeout div 1000).Time new_ts.it_interval.tv_sec = (timeout div 1000).Time
@ -976,8 +950,6 @@ else:
setKey(s, fdi, fdi, flags, 0, data) setKey(s, fdi, fdi, flags, 0, data)
inc(s.count) inc(s.count)
result = fdi result = fdi
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -995,18 +967,16 @@ else:
if fdi == -1: if fdi == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
setNonBlocking(fdi.cint) setNonBlocking(fdi.cint)
var epv: epoll_event var epv: epoll_event
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
epv.events = EPOLLIN or EPOLLRDHUP epv.events = EPOLLIN or EPOLLRDHUP
if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
setKey(s, fdi, signal, {flagSignal}, signal, data) setKey(s, fdi, signal, {Event.flagSignal}, signal, data)
inc(s.count) inc(s.count)
result = fdi result = fdi
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -1020,28 +990,22 @@ else:
discard sigemptyset(omask) discard sigemptyset(omask)
discard sigaddset(nmask, posix.SIGCHLD) discard sigaddset(nmask, posix.SIGCHLD)
blockSignals(nmask, omask) blockSignals(nmask, omask)
try:
var fdi = signalfd(-1, nmask, 0) var fdi = signalfd(-1, nmask, 0)
if fd == -1: if fd == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
setNonBlocking(fdi.cint) setNonBlocking(fdi.cint)
var epv: epoll_event var epv: epoll_event
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
epv.events = EPOLLIN or EPOLLRDHUP epv.events = EPOLLIN or EPOLLRDHUP
if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
setKey(s, fdi, pid, {flagProcess}, pid, data) setKey(s, fdi, pid, {Event.flagProcess}, pid, data)
inc(s.count) inc(s.count)
result = fdi result = fdi
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
except:
if fd != -1: discard posix.close(fd.cint)
unblockSignals(omask, nmask)
proc flush*[T](s: Selector[T]) = proc flush*[T](s: Selector[T]) =
discard discard
@ -1051,18 +1015,13 @@ else:
proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) =
let fdi = int(ev.efd) let fdi = int(ev.efd)
if fdi.uint < s.maxFD: doAssert(s.fds[fdi].ident == 0)
if s.fds[fdi].ident == 0: setKey(s, fdi, fdi, {Event.flagUser}, 0, data)
setKey(s, fdi, fdi, {flagUser}, 0, data)
var epv = epoll_event(events: EPOLLIN or EPOLLRDHUP) var epv = epoll_event(events: EPOLLIN or EPOLLRDHUP)
epv.data.u64 = ev.efd.uint epv.data.u64 = ev.efd.uint
if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, ev.efd, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, ev.efd, addr epv) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
inc(s.count) inc(s.count)
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else:
raise newException(ValueError, "Maximum file descriptors exceeded")
proc setEvent*(ev: SelectEvent) = proc setEvent*(ev: SelectEvent) =
var data : uint64 = 1 var data : uint64 = 1
@ -1105,36 +1064,36 @@ else:
if skey.ident != 0 and flags != {}: if skey.ident != 0 and flags != {}:
block processItem: block processItem:
if (pevents and EPOLLERR) != 0 or (pevents and EPOLLHUP) != 0: if (pevents and EPOLLERR) != 0 or (pevents and EPOLLHUP) != 0:
events = events + {eventError} events = events + {Event.Error}
if (pevents and EPOLLOUT) != 0: if (pevents and EPOLLOUT) != 0:
events = events + {eventWrite} events = events + {Event.Write}
if (pevents and EPOLLIN) != 0: if (pevents and EPOLLIN) != 0:
if flagHandle in flags: if Event.flagHandle in flags:
events = events + {eventRead} events = events + {Event.Read}
elif flagTimer in flags: elif Event.flagTimer in flags:
var data: uint64 = 0 var data: uint64 = 0
if posix.read(fdi.cint, addr data, if posix.read(fdi.cint, addr data,
sizeof(uint64)) != sizeof(uint64): sizeof(uint64)) != sizeof(uint64):
raiseOSError(osLastError()) raiseOSError(osLastError())
events = events + {eventTimer} events = events + {Event.Timer}
elif flagSignal in flags: elif Event.flagSignal in flags:
var data: SignalFdInfo var data: SignalFdInfo
if posix.read(fdi.cint, addr data, if posix.read(fdi.cint, addr data,
sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): sizeof(SignalFdInfo)) != sizeof(SignalFdInfo):
raiseOsError(osLastError()) raiseOsError(osLastError())
events = events + {eventSignal} events = events + {Event.Signal}
elif flagProcess in flags: elif Event.flagProcess in flags:
var data: SignalFdInfo var data: SignalFdInfo
if posix.read(fdi.cint, addr data, if posix.read(fdi.cint, addr data,
sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): sizeof(SignalFdInfo)) != sizeof(SignalFdInfo):
raiseOsError(osLastError()) raiseOsError(osLastError())
if cast[int](data.ssi_pid) == skey.param: if cast[int](data.ssi_pid) == skey.param:
events = events + {eventProcess} events = events + {Event.Process}
# we want to free resources for this event # we want to free resources for this event
flags = flags + {flagOneshot} flags = flags + {Event.flagOneshot}
else: else:
break processItem break processItem
elif flagUser in flags: elif Event.flagUser in flags:
var data: uint = 0 var data: uint = 0
if posix.read(fdi.cint, addr data, if posix.read(fdi.cint, addr data,
sizeof(uint)) != sizeof(uint): sizeof(uint)) != sizeof(uint):
@ -1145,7 +1104,7 @@ else:
continue continue
else: else:
raiseOSError(err) raiseOSError(err)
events = events + {eventUser} events = events + {Event.User}
else: else:
raise newException(ValueError, raise newException(ValueError,
"Unsupported epoll event in queue") "Unsupported epoll event in queue")
@ -1153,7 +1112,7 @@ else:
results[k].events = events results[k].events = events
results[k].data = skey.key.data results[k].data = skey.key.data
if flagOneshot in flags: if Event.flagOneshot in flags:
var epv: epoll_event var epv: epoll_event
try: try:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint,
@ -1305,12 +1264,12 @@ else:
events: set[Event]) = events: set[Event]) =
mixin withSelectLock mixin withSelectLock
s.withSelectLock(): s.withSelectLock():
if eventRead in events: if Event.Read in events:
if s.rSet.fd_count == FD_SETSIZE: if s.rSet.fd_count == FD_SETSIZE:
raise newException(ValueError, "Maximum numbers of fds exceeded") raise newException(ValueError, "Maximum numbers of fds exceeded")
iFD_SET(fd, s.rSet) iFD_SET(fd, s.rSet)
inc(s.count) inc(s.count)
if eventWrite in events: if Event.Write in events:
if s.wSet.fd_count == FD_SETSIZE: if s.wSet.fd_count == FD_SETSIZE:
raise newException(ValueError, "Maximum numbers of fds exceeded") raise newException(ValueError, "Maximum numbers of fds exceeded")
iFD_SET(fd, s.wSet) iFD_SET(fd, s.wSet)
@ -1320,7 +1279,7 @@ else:
proc registerHandle*[T](s: Selector[T], fd: SocketHandle, proc registerHandle*[T](s: Selector[T], fd: SocketHandle,
events: set[Event], data: T) = events: set[Event], data: T) =
var fdi = int(fd) var fdi = int(fd)
var flags = {flagHandle} + events var flags = {Event.flagHandle} + events
var nkey = SelectorKey[T](ident: fdi, flags: flags) var nkey = SelectorKey[T](ident: fdi, flags: flags)
nkey.key.fd = fdi nkey.key.fd = fdi
nkey.key.data = data nkey.key.data = data
@ -1333,21 +1292,21 @@ else:
events: set[Event]) = events: set[Event]) =
s.withSelectLock(): s.withSelectLock():
withValue(s.fds, fd, skey) do: withValue(s.fds, fd, skey) do:
if flagHandle in skey.flags: if Event.flagHandle in skey.flags:
var oe = skey.flags var oe = skey.flags
var ne = events + {flagHandle} var ne = events + {Event.flagHandle}
if oe != ne: if oe != ne:
if (eventRead in oe) and (eventRead notin ne): if (Event.Read in oe) and (Event.Read notin ne):
iFD_CLR(fd, s.rSet) iFD_CLR(fd, s.rSet)
dec(s.count) dec(s.count)
if (eventWrite in oe) and (eventWrite notin ne): if (Event.Write in oe) and (Event.Write notin ne):
iFD_CLR(fd, s.wSet) iFD_CLR(fd, s.wSet)
iFD_CLR(fd, s.eSet) iFD_CLR(fd, s.eSet)
dec(s.count) dec(s.count)
if (eventRead notin oe) and (eventRead in ne): if (Event.Read notin oe) and (Event.Read in ne):
iFD_SET(fd, s.rSet) iFD_SET(fd, s.rSet)
inc(s.count) inc(s.count)
if (eventWrite notin oe) and (eventWrite in ne): if (Event.Write notin oe) and (Event.Write in ne):
iFD_SET(fd, s.wSet) iFD_SET(fd, s.wSet)
iFD_SET(fd, s.eSet) iFD_SET(fd, s.eSet)
inc(s.count) inc(s.count)
@ -1384,17 +1343,17 @@ else:
proc unregister*[T](s: Selector[T], fd: SocketHandle) = proc unregister*[T](s: Selector[T], fd: SocketHandle) =
s.withSelectLock(): s.withSelectLock():
s.fds.withValue(fd, skey) do: s.fds.withValue(fd, skey) do:
if eventRead in skey.flags: if Event.Read in skey.flags:
iFD_CLR(fd, s.rSet) iFD_CLR(fd, s.rSet)
dec(s.count) dec(s.count)
if eventWrite in skey.flags: if Event.Write in skey.flags:
iFD_CLR(fd, s.wSet) iFD_CLR(fd, s.wSet)
iFD_CLR(fd, s.eSet) iFD_CLR(fd, s.eSet)
dec(s.count) dec(s.count)
s.fds.del(fd) s.fds.del(fd)
proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) =
var flags = {flagUser, eventRead} var flags = {Event.flagUser, Event.Read}
var nkey = SelectorKey[T](ident: ev.rsock.int, flags: flags) var nkey = SelectorKey[T](ident: ev.rsock.int, flags: flags)
nkey.key.fd = ev.rsock.int nkey.key.fd = ev.rsock.int
nkey.key.data = data nkey.key.data = data
@ -1488,13 +1447,13 @@ else:
while i < rset.fd_count: while i < rset.fd_count:
let fd = rset.fd_array[i] let fd = rset.fd_array[i]
if iFD_ISSET(fd, rset): if iFD_ISSET(fd, rset):
var events = {eventRead} var events = {Event.Read}
if iFD_ISSET(fd, eset): events = events + {eventError} if iFD_ISSET(fd, eset): events = events + {Event.Error}
if iFD_ISSET(fd, wset): events = events + {eventWrite} if iFD_ISSET(fd, wset): events = events + {Event.Write}
s.fds.withValue(fd, skey) do: s.fds.withValue(fd, skey) do:
if flagHandle in skey.flags: if Event.flagHandle in skey.flags:
skey.key.events = events skey.key.events = events
elif flagUser in skey.flags: elif Event.flagUser in skey.flags:
var data: int = 0 var data: int = 0
if winlean.recv(fd, cast[pointer](addr(data)), if winlean.recv(fd, cast[pointer](addr(data)),
sizeof(int).cint, 0) != sizeof(int): sizeof(int).cint, 0) != sizeof(int):
@ -1505,7 +1464,7 @@ else:
# someone already consumed event data # someone already consumed event data
inc(i) inc(i)
continue continue
skey.key.events = {eventUser} skey.key.events = {Event.User}
results[rindex].fd = skey.key.fd results[rindex].fd = skey.key.fd
results[rindex].data = skey.key.data results[rindex].data = skey.key.data
results[rindex].events = skey.key.events results[rindex].events = skey.key.events
@ -1516,9 +1475,9 @@ else:
while i < wset.fd_count: while i < wset.fd_count:
let fd = wset.fd_array[i] let fd = wset.fd_array[i]
if iFD_ISSET(fd, wset): if iFD_ISSET(fd, wset):
var events = {eventWrite} var events = {Event.Write}
if not iFD_ISSET(fd, rset): if not iFD_ISSET(fd, rset):
if iFD_ISSET(fd, eset): events = events + {eventError} if iFD_ISSET(fd, eset): events = events + {Event.Error}
s.fds.withValue(fd, skey) do: s.fds.withValue(fd, skey) do:
skey.key.events = events skey.key.events = events
results[rindex].fd = skey.key.fd results[rindex].fd = skey.key.fd
@ -1545,7 +1504,6 @@ else:
else: else:
const MAX_POLL_RESULT_EVENTS = 64 const MAX_POLL_RESULT_EVENTS = 64
when hasThreadSupport:
type type
SelectorImpl[T] = object SelectorImpl[T] = object
maxFD : uint maxFD : uint
@ -1553,27 +1511,16 @@ else:
fds: ptr SharedArray[SelectorKey[T]] fds: ptr SharedArray[SelectorKey[T]]
pollfds: ptr SharedArray[TPollFd] pollfds: ptr SharedArray[TPollFd]
count: int count: int
lock: Lock
else:
type
SelectorImpl[T] = object
maxFD : uint
pollcnt: int
fds: seq[SelectorKey[T]]
pollfds: seq[TPollFd]
count: int
when hasThreadSupport: when hasThreadSupport:
type Selector*[T] = ptr SelectorImpl[T] lock: Lock
else:
type Selector*[T] = ref SelectorImpl[T] Selector*[T] = ptr SelectorImpl[T]
type
SelectEventImpl = object SelectEventImpl = object
rfd: cint rfd: cint
wfd: cint wfd: cint
type SelectEvent* = ptr SelectEventImpl SelectEvent* = ptr SelectEventImpl
when hasThreadSupport: when hasThreadSupport:
template withPollLock[T](s: Selector[T], body: untyped) = template withPollLock[T](s: Selector[T], body: untyped) =
@ -1590,16 +1537,12 @@ else:
proc newSelector*[T](): Selector[T] = proc newSelector*[T](): Selector[T] =
var maxFD = getMaxFds() var maxFD = getMaxFds()
when hasThreadSupport:
result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T])))
result.maxFD = maxFD.uint result.maxFD = maxFD.uint
result.fds = allocSharedArray[SelectorKey[T]](maxFD) result.fds = allocSharedArray[SelectorKey[T]](maxFD)
result.pollfds = allocSharedArray[TPollFd](maxFD) result.pollfds = allocSharedArray[TPollFd](maxFD)
when hasThreadSupport:
initLock(result.lock) initLock(result.lock)
else:
result = Selector[T](maxFD: maxFD.uint)
result.fds = newSeq[SelectorKey[T]](maxFD)
result.pollfds = newSeq[TPollFd](maxFD)
proc close*[T](s: Selector[T]) = proc close*[T](s: Selector[T]) =
when hasThreadSupport: when hasThreadSupport:
@ -1611,8 +1554,8 @@ else:
template pollAdd[T](s: Selector[T], sock: cint, events: set[Event]) = template pollAdd[T](s: Selector[T], sock: cint, events: set[Event]) =
withPollLock(s): withPollLock(s):
var pollev: cshort = 0 var pollev: cshort = 0
if eventRead in events: pollev = pollev or POLLIN if Event.Read in events: pollev = pollev or POLLIN
if eventWrite in events: pollev = pollev or POLLOUT if Event.Write in events: pollev = pollev or POLLOUT
s.pollfds[s.pollcnt].fd = cint(sock) s.pollfds[s.pollcnt].fd = cint(sock)
s.pollfds[s.pollcnt].events = pollev s.pollfds[s.pollcnt].events = pollev
inc(s.count) inc(s.count)
@ -1622,8 +1565,8 @@ else:
withPollLock(s): withPollLock(s):
var i = 0 var i = 0
var pollev: cshort = 0 var pollev: cshort = 0
if eventRead in events: pollev = pollev or POLLIN if Event.Read in events: pollev = pollev or POLLIN
if eventWrite in events: pollev = pollev or POLLOUT if Event.Write in events: pollev = pollev or POLLOUT
while i < s.pollcnt: while i < s.pollcnt:
if s.pollfds[i].fd == sock: if s.pollfds[i].fd == sock:
@ -1658,11 +1601,9 @@ else:
events: set[Event], data: T) = events: set[Event], data: T) =
var fdi = int(fd) var fdi = int(fd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident == 0: doAssert(s.fds[fdi].ident == 0)
setKey(s, fdi, fdi, {flagHandle} + events, 0, data) setKey(s, fdi, fdi, {Event.flagHandle} + events, 0, data)
s.pollAdd(fdi.cint, events) s.pollAdd(fdi.cint, events)
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -1670,21 +1611,16 @@ else:
events: set[Event]) = events: set[Event]) =
var fdi = int(fd) var fdi = int(fd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident != 0:
var oe = s.fds[fdi].flags var oe = s.fds[fdi].flags
if flagHandle in oe: doAssert(s.fds[fdi].ident != 0)
var ne = events + {flagHandle} doAssert(Event.flagHandle in oe)
var ne = events + {Event.flagHandle}
if ne != oe: if ne != oe:
if events != {}: if events != {}:
s.pollUpdate(fd.cint, events) s.pollUpdate(fd.cint, events)
else: else:
s.pollRemove(fd.cint) s.pollRemove(fd.cint)
s.fds[fdi].flags = ne s.fds[fdi].flags = ne
else:
raise newException(ValueError,
"Could not update non-handle descriptor")
else:
raise newException(ValueError, "Re-use of non closed descriptor")
else: else:
raise newException(ValueError, "Maximum file descriptors exceeded") raise newException(ValueError, "Maximum file descriptors exceeded")
@ -1702,15 +1638,10 @@ else:
proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) = proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) =
var fdi = int(ev.rfd) var fdi = int(ev.rfd)
if fdi.uint < s.maxFD: doAssert(s.fds[fdi].ident == 0)
if s.fds[fdi].ident == 0: var events = {Event.flagUser, Event.Read}
var events = {flagUser, eventRead}
setKey(s, fdi, fdi, events, 0, data) setKey(s, fdi, fdi, events, 0, data)
s.pollAdd(fdi.cint, events) s.pollAdd(fdi.cint, events)
else:
raise newException(ValueError, "Re-use of non-closed descriptor")
else:
raise newException(ValueError, "Maximum file descriptors exceeded")
proc flush*[T](s: Selector[T]) = discard proc flush*[T](s: Selector[T]) = discard
@ -1727,7 +1658,7 @@ else:
proc unregister*[T](s: Selector[T], ev: SelectEvent) = proc unregister*[T](s: Selector[T], ev: SelectEvent) =
var fdi = int(ev.rfd) var fdi = int(ev.rfd)
if fdi.uint < s.maxFD: if fdi.uint < s.maxFD:
if s.fds[fdi].ident != 0 and (flagUser in s.fds[fdi].flags): if s.fds[fdi].ident != 0 and (Event.flagUser in s.fds[fdi].flags):
clearKey(s, fdi) clearKey(s, fdi)
s.pollRemove(fdi.cint) s.pollRemove(fdi.cint)
@ -1769,16 +1700,16 @@ else:
if revents != 0: if revents != 0:
var events: set[Event] = {} var events: set[Event] = {}
if (revents and POLLIN) != 0: if (revents and POLLIN) != 0:
events = events + {eventRead} events = events + {Event.Read}
if (revents and POLLOUT) != 0: if (revents and POLLOUT) != 0:
events = events + {eventWrite} events = events + {Event.Write}
if (revents and POLLERR) != 0 or (revents and POLLHUP) != 0 or if (revents and POLLERR) != 0 or (revents and POLLHUP) != 0 or
(revents and POLLNVAL) != 0: (revents and POLLNVAL) != 0:
events = events + {eventError} events = events + {Event.Error}
var skey = addr(s.fds[fd]) var skey = addr(s.fds[fd])
if flagUser in skey.flags: if Event.flagUser in skey.flags:
if eventRead in events: if Event.Read in events:
var data: int = 0 var data: int = 0
if posix.read(fd, addr data, sizeof(int)) != sizeof(int): if posix.read(fd, addr data, sizeof(int)) != sizeof(int):
let err = osLastError() let err = osLastError()
@ -1788,7 +1719,7 @@ else:
# someone already consumed event data # someone already consumed event data
inc(i) inc(i)
continue continue
events = {eventUser} events = {Event.User}
results[rindex].fd = fd results[rindex].fd = fd
results[rindex].events = events results[rindex].events = events

22
tests/async/tioselectors.nim
View file

@ -38,8 +38,8 @@ when not defined(windows):
var client_socket = create_test_socket() var client_socket = create_test_socket()
var server_socket = create_test_socket() var server_socket = create_test_socket()
registerHandle(selector, server_socket, {eventRead}, 0) registerHandle(selector, server_socket, {Event.Read}, 0)
registerHandle(selector, client_socket, {eventWrite}, 0) registerHandle(selector, client_socket, {Event.Write}, 0)
var option : int32 = 1 var option : int32 = 1
if setsockopt(server_socket, cint(SOL_SOCKET), cint(SO_REUSEADDR), if setsockopt(server_socket, cint(SOL_SOCKET), cint(SO_REUSEADDR),
@ -67,13 +67,13 @@ when not defined(windows):
cast[ptr SockAddr](addr(sockAddress)), cast[ptr SockAddr](addr(sockAddress)),
addr(addrLen)) addr(addrLen))
assert(server2_socket != osInvalidSocket) assert(server2_socket != osInvalidSocket)
selector.registerHandle(server2_socket, {eventRead}, 0) selector.registerHandle(server2_socket, {Event.Read}, 0)
if posix.send(client_socket, addr(client_message[0]), if posix.send(client_socket, addr(client_message[0]),
len(client_message), 0) == -1: len(client_message), 0) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
selector.updateHandle(client_socket, {eventRead}) selector.updateHandle(client_socket, {Event.Read})
var rc2 = selector.select(100) var rc2 = selector.select(100)
assert(len(rc2) == 1) assert(len(rc2) == 1)
@ -90,13 +90,13 @@ when not defined(windows):
break break
assert(test1) assert(test1)
selector.updateHandle(server2_socket, {eventWrite}) selector.updateHandle(server2_socket, {Event.Write})
var rc3 = selector.select(0) var rc3 = selector.select(0)
assert(len(rc3) == 1) assert(len(rc3) == 1)
if posix.send(server2_socket, addr(server_message[0]), if posix.send(server2_socket, addr(server_message[0]),
len(server_message), 0) == -1: len(server_message), 0) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
selector.updateHandle(server2_socket, {eventRead}) selector.updateHandle(server2_socket, {Event.Read})
var rc4 = selector.select(100) var rc4 = selector.select(100)
assert(len(rc4) == 1) assert(len(rc4) == 1)
@ -238,7 +238,7 @@ when not defined(windows):
var event = newEvent() var event = newEvent()
for i in 0..high(thr): for i in 0..high(thr):
createThread(thr[i], event_wait_thread, event) createThread(thr[i], event_wait_thread, event)
selector.registerHandle(sock, {eventRead}, 1) selector.registerHandle(sock, {Event.Read}, 1)
discard selector.select(500) discard selector.select(500)
selector.unregister(sock) selector.unregister(sock)
event.setEvent() event.setEvent()
@ -273,8 +273,8 @@ else:
var client_socket = create_test_socket() var client_socket = create_test_socket()
var server_socket = create_test_socket() var server_socket = create_test_socket()
selector.registerHandle(server_socket, {eventRead}, 0) selector.registerHandle(server_socket, {Event.Read}, 0)
selector.registerHandle(client_socket, {eventWrite}, 0) selector.registerHandle(client_socket, {Event.Write}, 0)
var option : int32 = 1 var option : int32 = 1
if setsockopt(server_socket, cint(SOL_SOCKET), cint(SO_REUSEADDR), if setsockopt(server_socket, cint(SOL_SOCKET), cint(SO_REUSEADDR),
@ -305,13 +305,13 @@ else:
cast[ptr SockAddr](addr(sockAddress)), cast[ptr SockAddr](addr(sockAddress)),
addr(addrLen)) addr(addrLen))
assert(server2_socket != osInvalidSocket) assert(server2_socket != osInvalidSocket)
selector.registerHandle(server2_socket, {eventRead}, 0) selector.registerHandle(server2_socket, {Event.Read}, 0)
if send(client_socket, cast[pointer](addr(client_message[0])), if send(client_socket, cast[pointer](addr(client_message[0])),
cint(len(client_message)), 0) == -1: cint(len(client_message)), 0) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
selector.updateHandle(client_socket, {eventRead}) selector.updateHandle(client_socket, {Event.Read})
var rc2 = selector.select(100) var rc2 = selector.select(100)
assert(len(rc2) == 1) assert(len(rc2) == 1)