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

Fix #5128, #5184. (#5214)

Browse source 
* Fix #5128, #5184.
Removed flush() procedure from ioselectors.nim
Changed methods of work with application-driven data

* Make cache switch for kqueue, update test for it.

* Fix registerProcess bug returns wrong id.
Fix tupcoming_async test to compile with upcoming again.
Change socket() as unique identifier to dup(socket) as unique identifier.
This commit is contained in:
Eugene Kabanov 2017-01-16 15:01:40 +02:00 • committed by Andreas Rumpf
commit 108f5e688e
8 changed files with 674 additions and 479 deletions
Download patch file Download diff file Expand all files Collapse all files

72
lib/pure/ioselectors.nim
View file

@ -40,7 +40,6 @@ const ioselSupportedPlatform* = defined(macosx) or defined(freebsd) or
const bsdPlatform = defined(macosx) or defined(freebsd) or const bsdPlatform = defined(macosx) or defined(freebsd) or
defined(netbsd) or defined(openbsd) defined(netbsd) or defined(openbsd)
when defined(nimdoc): when defined(nimdoc):
type type
Selector*[T] = ref object Selector*[T] = ref object
@ -64,11 +63,10 @@ when defined(nimdoc):
VnodeRename, ## NOTE_RENAME (BSD specific, file renamed) VnodeRename, ## NOTE_RENAME (BSD specific, file renamed)
VnodeRevoke ## NOTE_REVOKE (BSD specific, file revoke occurred) VnodeRevoke ## NOTE_REVOKE (BSD specific, file revoke occurred)
ReadyKey*[T] = object ReadyKey* = object
## An object which holds result for descriptor ## An object which holds result for descriptor
fd* : int ## file/socket descriptor fd* : int ## file/socket descriptor
events*: set[Event] ## set of events events*: set[Event] ## set of events
data*: T ## application-defined data
SelectEvent* = object SelectEvent* = object
## An object which holds user defined event ## An object which holds user defined event
@ -142,15 +140,8 @@ when defined(nimdoc):
proc unregister*[T](s: Selector[T], fd: int|SocketHandle|cint) = proc unregister*[T](s: Selector[T], fd: int|SocketHandle|cint) =
## Unregisters file/socket descriptor ``fd`` from selector ``s``. ## Unregisters file/socket descriptor ``fd`` from selector ``s``.
proc flush*[T](s: Selector[T]) =
## Flushes all changes was made to kernel pool/queue.
## This function is useful only for BSD and MacOS, because
## kqueue supports bulk changes to be made.
## On Linux/Windows and other Posix compatible operation systems,
## ``flush`` is alias for `discard`.
proc selectInto*[T](s: Selector[T], timeout: int, proc selectInto*[T](s: Selector[T], timeout: int,
results: var openarray[ReadyKey[T]]): int = results: var openarray[ReadyKey]): int =
## Process call waiting for events registered in selector ``s``. ## Process call waiting for events registered in selector ``s``.
## The ``timeout`` argument specifies the minimum number of milliseconds ## The ``timeout`` argument specifies the minimum number of milliseconds
## the function will be blocked, if no events are not ready. Specifying a ## the function will be blocked, if no events are not ready. Specifying a
@ -159,7 +150,7 @@ when defined(nimdoc):
## ##
## Function returns number of triggered events. ## Function returns number of triggered events.
proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] =
## Process call waiting for events registered in selector ``s``. ## Process call waiting for events registered in selector ``s``.
## The ``timeout`` argument specifies the minimum number of milliseconds ## The ``timeout`` argument specifies the minimum number of milliseconds
## the function will be blocked, if no events are not ready. Specifying a ## the function will be blocked, if no events are not ready. Specifying a
@ -167,13 +158,23 @@ when defined(nimdoc):
## ##
## Function returns sequence of triggered events. ## Function returns sequence of triggered events.
proc getData*[T](s: Selector[T], fd: SocketHandle|int): T =
## Retrieves application-defined ``data`` associated with descriptor ``fd``.
## If specified descriptor ``fd`` is not registered, empty/default value
## will be returned.
proc setData*[T](s: Selector[T], fd: SocketHandle|int, data: var T): bool =
## Associate application-defined ``data`` with descriptor ``fd``.
##
## Returns ``true``, if data was succesfully updated, ``false`` otherwise.
template isEmpty*[T](s: Selector[T]): bool = template isEmpty*[T](s: Selector[T]): bool =
## Returns ``true``, if there no registered events or descriptors ## Returns ``true``, if there no registered events or descriptors
## in selector. ## in selector.
template withData*[T](s: Selector[T], fd: SocketHandle, value, template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
body: untyped) = body: untyped) =
## retrieves the application-data assigned with descriptor ``fd`` ## Retrieves the application-data assigned with descriptor ``fd``
## to ``value``. This ``value`` can be modified in the scope of ## to ``value``. This ``value`` can be modified in the scope of
## the ``withData`` call. ## the ``withData`` call.
## ##
@ -184,9 +185,9 @@ when defined(nimdoc):
## value.uid = 1000 ## value.uid = 1000
## ##
template withData*[T](s: Selector[T], fd: SocketHandle, value, template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
body1, body2: untyped) = body1, body2: untyped) =
## retrieves the application-data assigned with descriptor ``fd`` ## Retrieves the application-data assigned with descriptor ``fd``
## to ``value``. This ``value`` can be modified in the scope of ## to ``value``. This ``value`` can be modified in the scope of
## the ``withData`` call. ## the ``withData`` call.
## ##
@ -215,55 +216,68 @@ else:
type type
Event* {.pure.} = enum Event* {.pure.} = enum
Read, Write, Timer, Signal, Process, Vnode, User, Error, Oneshot, Read, Write, Timer, Signal, Process, Vnode, User, Error, Oneshot,
VnodeWrite, VnodeDelete, VnodeExtend, VnodeAttrib, VnodeLink, Finished, VnodeWrite, VnodeDelete, VnodeExtend, VnodeAttrib, VnodeLink,
VnodeRename, VnodeRevoke VnodeRename, VnodeRevoke
ReadyKey*[T] = object type
IOSelectorsException* = object of Exception
ReadyKey* = object
fd* : int fd* : int
events*: set[Event] events*: set[Event]
data*: T
SelectorKey[T] = object SelectorKey[T] = object
ident: int ident: int
events: set[Event] events: set[Event]
param: int param: int
key: ReadyKey[T] data: T
proc raiseIOSelectorsError[T](message: T) =
var msg = ""
when T is string:
msg.add(message)
elif T is OSErrorCode:
msg.add(osErrorMsg(message) & " (code: " & $int(message) & ")")
else:
msg.add("Internal Error\n")
var err = newException(IOSelectorsException, msg)
raise err
when not defined(windows): when not defined(windows):
import posix import posix
proc setNonBlocking(fd: cint) {.inline.} = proc setNonBlocking(fd: cint) {.inline.} =
var x = fcntl(fd, F_GETFL, 0) var x = fcntl(fd, F_GETFL, 0)
if x == -1: if x == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
else: else:
var mode = x or O_NONBLOCK var mode = x or O_NONBLOCK
if fcntl(fd, F_SETFL, mode) == -1: if fcntl(fd, F_SETFL, mode) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
template setKey(s, pident, pkeyfd, pevents, pparam, pdata) = template setKey(s, pident, pevents, pparam, pdata: untyped) =
var skey = addr(s.fds[pident]) var skey = addr(s.fds[pident])
skey.ident = pident skey.ident = pident
skey.events = pevents skey.events = pevents
skey.param = pparam skey.param = pparam
skey.key.fd = pkeyfd skey.data = data
skey.key.data = pdata
when ioselSupportedPlatform: when ioselSupportedPlatform:
template blockSignals(newmask: var Sigset, oldmask: var Sigset) = template blockSignals(newmask: var Sigset, oldmask: var Sigset) =
when hasThreadSupport: when hasThreadSupport:
if posix.pthread_sigmask(SIG_BLOCK, newmask, oldmask) == -1: if posix.pthread_sigmask(SIG_BLOCK, newmask, oldmask) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
else: else:
if posix.sigprocmask(SIG_BLOCK, newmask, oldmask) == -1: if posix.sigprocmask(SIG_BLOCK, newmask, oldmask) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
template unblockSignals(newmask: var Sigset, oldmask: var Sigset) = template unblockSignals(newmask: var Sigset, oldmask: var Sigset) =
when hasThreadSupport: when hasThreadSupport:
if posix.pthread_sigmask(SIG_UNBLOCK, newmask, oldmask) == -1: if posix.pthread_sigmask(SIG_UNBLOCK, newmask, oldmask) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
else: else:
if posix.sigprocmask(SIG_UNBLOCK, newmask, oldmask) == -1: if posix.sigprocmask(SIG_UNBLOCK, newmask, oldmask) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
when defined(linux): when defined(linux):
include ioselects/ioselectors_epoll include ioselects/ioselectors_epoll

249
lib/pure/ioselects/ioselectors_epoll.nim
View file

@ -12,7 +12,7 @@
import posix, times import posix, times
# Maximum number of events that can be returned # Maximum number of events that can be returned
const MAX_EPOLL_RESULT_EVENTS = 64 const MAX_EPOLL_EVENTS = 64
when not defined(android): when not defined(android):
type type
@ -115,7 +115,7 @@ proc newSelector*[T](): Selector[T] =
var maxFD = int(a.rlim_max) var maxFD = int(a.rlim_max)
doAssert(maxFD > 0) doAssert(maxFD > 0)
var epollFD = epoll_create(MAX_EPOLL_RESULT_EVENTS) var epollFD = epoll_create(MAX_EPOLL_EVENTS)
if epollFD < 0: if epollFD < 0:
raiseOsError(osLastError()) raiseOsError(osLastError())
@ -132,15 +132,21 @@ proc newSelector*[T](): Selector[T] =
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()) raiseIOSelectorsError(osLastError())
when hasThreadSupport: when hasThreadSupport:
deallocSharedArray(s.fds) deallocSharedArray(s.fds)
deallocShared(cast[pointer](s)) deallocShared(cast[pointer](s))
template clearKey[T](key: ptr SelectorKey[T]) =
var empty: T
key.ident = 0
key.events = {}
key.data = empty
proc newSelectEvent*(): SelectEvent = proc newSelectEvent*(): SelectEvent =
let fdci = eventfd(0, 0) let fdci = eventfd(0, 0)
if fdci == -1: if fdci == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fdci) setNonBlocking(fdci)
result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl)))
result.efd = fdci result.efd = fdci
@ -148,29 +154,30 @@ proc newSelectEvent*(): SelectEvent =
proc setEvent*(ev: SelectEvent) = proc setEvent*(ev: SelectEvent) =
var data : uint64 = 1 var data : uint64 = 1
if posix.write(ev.efd, addr data, sizeof(uint64)) == -1: if posix.write(ev.efd, addr data, sizeof(uint64)) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
proc close*(ev: SelectEvent) = proc close*(ev: SelectEvent) =
discard posix.close(ev.efd) if posix.close(ev.efd) == -1:
raiseIOSelectorsError(osLastError())
deallocShared(cast[pointer](ev)) deallocShared(cast[pointer](ev))
template checkFd(s, f) = template checkFd(s, f) =
if f >= s.maxFD: if f >= s.maxFD:
raise newException(ValueError, "Maximum file descriptors exceeded") raiseIOSelectorsError("Maximum file descriptors exceeded")
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) =
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
s.setKey(fdi, fdi, events, 0, data) s.setKey(fdi, events, 0, data)
if events != {}: if events != {}:
var epv = epoll_event(events: EPOLLRDHUP) var epv = epoll_event(events: EPOLLRDHUP)
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
if Event.Read in events: epv.events = epv.events or EPOLLIN if Event.Read in events: epv.events = epv.events or EPOLLIN
if Event.Write in events: epv.events = epv.events or EPOLLOUT if Event.Write in events: 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()) raiseIOSelectorsError(osLastError())
inc(s.count) inc(s.count)
proc updateHandle*[T](s: Selector[T], fd: SocketHandle, events: set[Event]) = proc updateHandle*[T](s: Selector[T], fd: SocketHandle, events: set[Event]) =
@ -190,15 +197,15 @@ proc updateHandle*[T](s: Selector[T], fd: SocketHandle, events: set[Event]) =
if pkey.events == {}: if pkey.events == {}:
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()) raiseIOSelectorsError(osLastError())
inc(s.count) inc(s.count)
else: else:
if events != {}: if events != {}:
if epoll_ctl(s.epollFD, EPOLL_CTL_MOD, fdi.cint, addr epv) == -1: if epoll_ctl(s.epollFD, EPOLL_CTL_MOD, fdi.cint, addr epv) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
else: else:
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()) raiseIOSelectorsError(osLastError())
dec(s.count) dec(s.count)
pkey.events = events pkey.events = events
@ -213,51 +220,56 @@ proc unregister*[T](s: Selector[T], fd: int|SocketHandle) =
if pkey.events * {Event.Read, Event.Write} != {}: if pkey.events * {Event.Read, Event.Write} != {}:
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:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
dec(s.count) dec(s.count)
elif Event.Timer in pkey.events: elif Event.Timer in pkey.events:
var epv = epoll_event() if Event.Finished notin pkey.events:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: var epv = epoll_event()
raiseOSError(osLastError()) if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
discard posix.close(fdi.cint) raiseIOSelectorsError(osLastError())
dec(s.count) dec(s.count)
if posix.close(cint(fdi)) == -1:
raiseIOSelectorsError(osLastError())
elif Event.Signal in pkey.events: elif Event.Signal in pkey.events:
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:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
var nmask, omask: Sigset var nmask, omask: Sigset
discard sigemptyset(nmask) discard sigemptyset(nmask)
discard sigemptyset(omask) discard sigemptyset(omask)
discard sigaddset(nmask, cint(s.fds[fdi].param)) discard sigaddset(nmask, cint(s.fds[fdi].param))
unblockSignals(nmask, omask) unblockSignals(nmask, omask)
discard posix.close(fdi.cint)
dec(s.count) dec(s.count)
if posix.close(cint(fdi)) == -1:
raiseIOSelectorsError(osLastError())
elif Event.Process in pkey.events: elif Event.Process in pkey.events:
var epv = epoll_event() if Event.Finished notin pkey.events:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: var epv = epoll_event()
raiseOSError(osLastError()) if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
var nmask, omask: Sigset raiseIOSelectorsError(osLastError())
discard sigemptyset(nmask) var nmask, omask: Sigset
discard sigemptyset(omask) discard sigemptyset(nmask)
discard sigaddset(nmask, SIGCHLD) discard sigemptyset(omask)
unblockSignals(nmask, omask) discard sigaddset(nmask, SIGCHLD)
discard posix.close(fdi.cint) unblockSignals(nmask, omask)
dec(s.count) dec(s.count)
if posix.close(cint(fdi)) == -1:
raiseIOSelectorsError(osLastError())
else: else:
if pkey.events * {Event.Read, Event.Write} != {}: if pkey.events * {Event.Read, Event.Write} != {}:
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:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
dec(s.count) dec(s.count)
elif Event.Timer in pkey.events: elif Event.Timer in pkey.events:
var epv = epoll_event() if Event.Finished notin pkey.events:
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1: var epv = epoll_event()
raiseOSError(osLastError()) if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fdi.cint, addr epv) == -1:
discard posix.close(fdi.cint) raiseIOSelectorsError(osLastError())
dec(s.count) dec(s.count)
if posix.close(cint(fdi)) == -1:
pkey.ident = 0 raiseIOSelectorsError(osLastError())
pkey.events = {} clearKey(pkey)
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)
@ -265,12 +277,11 @@ proc unregister*[T](s: Selector[T], ev: SelectEvent) =
var pkey = addr(s.fds[fdi]) var pkey = addr(s.fds[fdi])
doAssert(pkey.ident != 0) doAssert(pkey.ident != 0)
doAssert(Event.User in pkey.events) doAssert(Event.User in pkey.events)
pkey.ident = 0
pkey.events = {}
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:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
dec(s.count) dec(s.count)
clearKey(pkey)
proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool,
data: T): int {.discardable.} = data: T): int {.discardable.} =
@ -279,7 +290,7 @@ proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool,
old_ts: Itimerspec old_ts: Itimerspec
let fdi = timerfd_create(CLOCK_MONOTONIC, 0).int let fdi = timerfd_create(CLOCK_MONOTONIC, 0).int
if fdi == -1: if fdi == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fdi.cint) setNonBlocking(fdi.cint)
s.checkFd(fdi) s.checkFd(fdi)
@ -302,10 +313,10 @@ proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool,
new_ts.it_value.tv_nsec = new_ts.it_interval.tv_nsec new_ts.it_value.tv_nsec = new_ts.it_interval.tv_nsec
if timerfd_settime(fdi.cint, cint(0), new_ts, old_ts) == -1: if timerfd_settime(fdi.cint, cint(0), new_ts, old_ts) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
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()) raiseIOSelectorsError(osLastError())
s.setKey(fdi, fdi, events, 0, data) s.setKey(fdi, events, 0, data)
inc(s.count) inc(s.count)
result = fdi result = fdi
@ -323,7 +334,7 @@ when not defined(android):
let fdi = signalfd(-1, nmask, 0).int let fdi = signalfd(-1, nmask, 0).int
if fdi == -1: if fdi == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fdi.cint) setNonBlocking(fdi.cint)
s.checkFd(fdi) s.checkFd(fdi)
@ -332,8 +343,8 @@ when not defined(android):
var epv = epoll_event(events: EPOLLIN or EPOLLRDHUP) var epv = epoll_event(events: EPOLLIN or EPOLLRDHUP)
epv.data.u64 = fdi.uint epv.data.u64 = fdi.uint
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()) raiseIOSelectorsError(osLastError())
s.setKey(fdi, signal, {Event.Signal}, signal, data) s.setKey(fdi, {Event.Signal}, signal, data)
inc(s.count) inc(s.count)
result = fdi result = fdi
@ -350,7 +361,7 @@ when not defined(android):
let fdi = signalfd(-1, nmask, 0).int let fdi = signalfd(-1, nmask, 0).int
if fdi == -1: if fdi == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fdi.cint) setNonBlocking(fdi.cint)
s.checkFd(fdi) s.checkFd(fdi)
@ -360,30 +371,26 @@ when not defined(android):
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()) raiseIOSelectorsError(osLastError())
s.setKey(fdi, pid, {Event.Process, Event.Oneshot}, pid, data) s.setKey(fdi, {Event.Process, Event.Oneshot}, pid, data)
inc(s.count) inc(s.count)
result = fdi result = fdi
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)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
s.setKey(fdi, fdi, {Event.User}, 0, data) s.setKey(fdi, {Event.User}, 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()) raiseIOSelectorsError(osLastError())
inc(s.count) inc(s.count)
proc flush*[T](s: Selector[T]) =
discard
proc selectInto*[T](s: Selector[T], timeout: int, proc selectInto*[T](s: Selector[T], timeout: int,
results: var openarray[ReadyKey[T]]): int = results: var openarray[ReadyKey]): int =
var var
resTable: array[MAX_EPOLL_RESULT_EVENTS, epoll_event] resTable: array[MAX_EPOLL_EVENTS, epoll_event]
maxres = MAX_EPOLL_RESULT_EVENTS maxres = MAX_EPOLL_EVENTS
events: set[Event] = {}
i, k: int i, k: int
if maxres > len(results): if maxres > len(results):
@ -395,7 +402,7 @@ proc selectInto*[T](s: Selector[T], timeout: int,
result = 0 result = 0
let err = osLastError() let err = osLastError()
if cint(err) != EINTR: if cint(err) != EINTR:
raiseOSError(err) raiseIOSelectorsError(err)
elif count == 0: elif count == 0:
result = 0 result = 0
else: else:
@ -404,108 +411,126 @@ proc selectInto*[T](s: Selector[T], timeout: int,
while i < count: while i < count:
let fdi = int(resTable[i].data.u64) let fdi = int(resTable[i].data.u64)
let pevents = resTable[i].events let pevents = resTable[i].events
var skey = addr(s.fds[fdi]) var pkey = addr(s.fds[fdi])
doAssert(skey.ident != 0) doAssert(pkey.ident != 0)
events = {} var rkey = ReadyKey(fd: int(fdi), events: {})
if (pevents and EPOLLERR) != 0 or (pevents and EPOLLHUP) != 0: if (pevents and EPOLLERR) != 0 or (pevents and EPOLLHUP) != 0:
events.incl(Event.Error) rkey.events.incl(Event.Error)
if (pevents and EPOLLOUT) != 0: if (pevents and EPOLLOUT) != 0:
events.incl(Event.Write) rkey.events.incl(Event.Write)
when not defined(android): when not defined(android):
if (pevents and EPOLLIN) != 0: if (pevents and EPOLLIN) != 0:
if Event.Read in skey.events: if Event.Read in pkey.events:
events.incl(Event.Read) rkey.events.incl(Event.Read)
elif Event.Timer in skey.events: elif Event.Timer in pkey.events:
var data: uint64 = 0 var data: uint64 = 0
if posix.read(fdi.cint, addr data, sizeof(uint64)) != sizeof(uint64): if posix.read(cint(fdi), addr data,
raiseOSError(osLastError()) sizeof(uint64)) != sizeof(uint64):
events = {Event.Timer} raiseIOSelectorsError(osLastError())
elif Event.Signal in skey.events: rkey.events.incl(Event.Timer)
elif Event.Signal in pkey.events:
var data = SignalFdInfo() var data = SignalFdInfo()
if posix.read(fdi.cint, addr data, if posix.read(cint(fdi), addr data,
sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): sizeof(SignalFdInfo)) != sizeof(SignalFdInfo):
raiseOsError(osLastError()) raiseIOSelectorsError(osLastError())
events = {Event.Signal} rkey.events.incl(Event.Signal)
elif Event.Process in skey.events: elif Event.Process in pkey.events:
var data = SignalFdInfo() var data = SignalFdInfo()
if posix.read(fdi.cint, addr data, if posix.read(cint(fdi), addr data,
sizeof(SignalFdInfo)) != sizeof(SignalFdInfo): sizeof(SignalFdInfo)) != sizeof(SignalFdInfo):
raiseOsError(osLastError()) raiseIOSelectorsError(osLastError())
if cast[int](data.ssi_pid) == skey.param: if cast[int](data.ssi_pid) == pkey.param:
events = {Event.Process} rkey.events.incl(Event.Process)
else: else:
inc(i) inc(i)
continue continue
elif Event.User in skey.events: elif Event.User in pkey.events:
var data: uint64 = 0 var data: uint64 = 0
if posix.read(fdi.cint, addr data, sizeof(uint64)) != sizeof(uint64): if posix.read(cint(fdi), addr data,
sizeof(uint64)) != sizeof(uint64):
let err = osLastError() let err = osLastError()
if err == OSErrorCode(EAGAIN): if err == OSErrorCode(EAGAIN):
inc(i) inc(i)
continue continue
else: else:
raiseOSError(err) raiseIOSelectorsError(err)
events = {Event.User} rkey.events.incl(Event.User)
else: else:
if (pevents and EPOLLIN) != 0: if (pevents and EPOLLIN) != 0:
if Event.Read in skey.events: if Event.Read in pkey.events:
events.incl(Event.Read) rkey.events.incl(Event.Read)
elif Event.Timer in skey.events: elif Event.Timer in pkey.events:
var data: uint64 = 0 var data: uint64 = 0
if posix.read(fdi.cint, addr data, sizeof(uint64)) != sizeof(uint64): if posix.read(cint(fdi), addr data,
raiseOSError(osLastError()) sizeof(uint64)) != sizeof(uint64):
events = {Event.Timer} raiseIOSelectorsError(osLastError())
elif Event.User in skey.events: rkey.events.incl(Event.Timer)
elif Event.User in pkey.events:
var data: uint64 = 0 var data: uint64 = 0
if posix.read(fdi.cint, addr data, sizeof(uint64)) != sizeof(uint64): if posix.read(cint(fdi), addr data,
sizeof(uint64)) != sizeof(uint64):
let err = osLastError() let err = osLastError()
if err == OSErrorCode(EAGAIN): if err == OSErrorCode(EAGAIN):
inc(i) inc(i)
continue continue
else: else:
raiseOSError(err) raiseIOSelectorsError(err)
events = {Event.User} rkey.events.incl(Event.User)
skey.key.events = events if Event.Oneshot in pkey.events:
results[k] = skey.key
inc(k)
if Event.Oneshot in skey.events:
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, cint(fdi), addr epv) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
discard posix.close(fdi.cint) # we will not clear key until it will be unregistered, so
skey.ident = 0 # application can obtain data, but we will decrease counter,
skey.events = {} # because epoll is empty.
dec(s.count) 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) inc(i)
result = k result = k
proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] =
result = newSeq[ReadyKey[T]](MAX_EPOLL_RESULT_EVENTS) result = newSeq[ReadyKey](MAX_EPOLL_EVENTS)
let count = selectInto(s, timeout, result) let count = selectInto(s, timeout, result)
result.setLen(count) result.setLen(count)
template isEmpty*[T](s: Selector[T]): bool = template isEmpty*[T](s: Selector[T]): bool =
(s.count == 0) (s.count == 0)
template withData*[T](s: Selector[T], fd: SocketHandle, value, proc getData*[T](s: Selector[T], fd: SocketHandle|int): T =
let fdi = int(fd)
s.checkFd(fdi)
if s.fds[fdi].ident != 0:
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 s.fds[fdi].ident != 0:
s.fds[fdi].data = data
result = true
template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
body: untyped) = body: untyped) =
mixin checkFd mixin checkFd
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
if s.fds[fdi].ident != 0: if s.fds[fdi].ident != 0:
var value = addr(s.fds[fdi].key.data) var value = addr(s.fds[fdi].data)
body body
template withData*[T](s: Selector[T], fd: SocketHandle, value, body1, template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1,
body2: untyped) = body2: untyped) =
mixin checkFd mixin checkFd
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
if s.fds[fdi].ident != 0: if s.fds[fdi].ident != 0:
var value = addr(s.fds[fdi].key.data) var value = addr(s.fds[fdi].data)
body1 body1
else: else:
body2 body2

415
lib/pure/ioselects/ioselectors_kqueue.nim
View file

@ -12,15 +12,16 @@
import posix, times, kqueue import posix, times, kqueue
const const
# Maximum number of cached changes.
MAX_KQUEUE_CHANGE_EVENTS = 64
# Maximum number of events that can be returned. # Maximum number of events that can be returned.
MAX_KQUEUE_RESULT_EVENTS = 64 MAX_KQUEUE_EVENTS = 64
# SIG_IGN and SIG_DFL declared in posix.nim as variables, but we need them # SIG_IGN and SIG_DFL declared in posix.nim as variables, but we need them
# to be constants and GC-safe. # to be constants and GC-safe.
SIG_DFL = cast[proc(x: cint) {.noconv,gcsafe.}](0) SIG_DFL = cast[proc(x: cint) {.noconv,gcsafe.}](0)
SIG_IGN = cast[proc(x: cint) {.noconv,gcsafe.}](1) SIG_IGN = cast[proc(x: cint) {.noconv,gcsafe.}](1)
when defined(kqcache):
const CACHE_EVENTS = true
when defined(macosx) or defined(freebsd): when defined(macosx) or defined(freebsd):
when defined(macosx): when defined(macosx):
const MAX_DESCRIPTORS_ID = 29 # KERN_MAXFILESPERPROC (MacOS) const MAX_DESCRIPTORS_ID = 29 # KERN_MAXFILESPERPROC (MacOS)
@ -45,68 +46,88 @@ when hasThreadSupport:
SelectorImpl[T] = object SelectorImpl[T] = object
kqFD : cint kqFD : cint
maxFD : int maxFD : int
changesTable: array[MAX_KQUEUE_CHANGE_EVENTS, KEvent] changes: seq[KEvent]
changesCount: int
fds: ptr SharedArray[SelectorKey[T]] fds: ptr SharedArray[SelectorKey[T]]
count: int count: int
changesLock: Lock changesLock: Lock
sock: cint
Selector*[T] = ptr SelectorImpl[T] Selector*[T] = ptr SelectorImpl[T]
else: else:
type type
SelectorImpl[T] = object SelectorImpl[T] = object
kqFD : cint kqFD : cint
maxFD : int maxFD : int
changesTable: array[MAX_KQUEUE_CHANGE_EVENTS, KEvent] changes: seq[KEvent]
changesCount: int
fds: seq[SelectorKey[T]] fds: seq[SelectorKey[T]]
count: int count: int
sock: cint
Selector*[T] = ref SelectorImpl[T] Selector*[T] = ref SelectorImpl[T]
type type
SelectEventImpl = object SelectEventImpl = object
rfd: cint rfd: cint
wfd: cint wfd: cint
# SelectEvent is declared as `ptr` to be placed in `shared memory`,
# so you can share one SelectEvent handle between threads. SelectEvent* = ptr SelectEventImpl
type 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, s.sock)
if result == -1:
raiseIOSelectorsError(osLastError())
proc newSelector*[T](): Selector[T] = proc newSelector*[T](): Selector[T] =
var maxFD = 0.cint var maxFD = 0.cint
var size = csize(sizeof(cint)) var size = csize(sizeof(cint))
var namearr = [1.cint, MAX_DESCRIPTORS_ID.cint] var namearr = [1.cint, MAX_DESCRIPTORS_ID.cint]
# Obtain maximum number of file descriptors for process # Obtain maximum number of opened file descriptors for process
if sysctl(addr(namearr[0]), 2, cast[pointer](addr maxFD), addr size, if sysctl(addr(namearr[0]), 2, cast[pointer](addr maxFD), addr size,
nil, 0) != 0: nil, 0) != 0:
raiseOsError(osLastError()) raiseIOSelectorsError(osLastError())
var kqFD = kqueue() var kqFD = kqueue()
if kqFD < 0: if kqFD < 0:
raiseOsError(osLastError()) raiseIOSelectorsError(osLastError())
when hasThreadSupport: when hasThreadSupport:
result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T]))) result = cast[Selector[T]](allocShared0(sizeof(SelectorImpl[T])))
result.kqFD = kqFD
result.maxFD = maxFD.int
result.fds = allocSharedArray[SelectorKey[T]](maxFD) result.fds = allocSharedArray[SelectorKey[T]](maxFD)
initLock(result.changesLock) initLock(result.changesLock)
else: else:
result = Selector[T]() result = Selector[T]()
result.kqFD = kqFD
result.maxFD = maxFD.int
result.fds = newSeq[SelectorKey[T]](maxFD) result.fds = newSeq[SelectorKey[T]](maxFD)
result.kqFD = kqFD
result.maxFD = maxFD.int
result.changes = newSeqOfCap[KEvent](MAX_KQUEUE_EVENTS)
# 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.
result.sock = posix.socket(posix.AF_INET, posix.SOCK_STREAM,
posix.IPPROTO_TCP).cint
if result.sock == -1:
raiseIOSelectorsError(osLastError())
proc close*[T](s: Selector[T]) = proc close*[T](s: Selector[T]) =
if posix.close(s.kqFD) != 0: if posix.close(s.kqFD) != 0:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
when hasThreadSupport: when hasThreadSupport:
deinitLock(s.changesLock) deinitLock(s.changesLock)
deallocSharedArray(s.fds) deallocSharedArray(s.fds)
deallocShared(cast[pointer](s)) deallocShared(cast[pointer](s))
template clearKey[T](key: ptr SelectorKey[T]) =
var empty: T
key.ident = 0
key.events = {}
key.data = empty
proc newSelectEvent*(): SelectEvent = proc newSelectEvent*(): SelectEvent =
var fds: array[2, cint] var fds: array[2, cint]
if posix.pipe(fds) == -1: if posix.pipe(fds) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fds[0]) setNonBlocking(fds[0])
setNonBlocking(fds[1]) setNonBlocking(fds[1])
result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl)))
@ -116,16 +137,18 @@ proc newSelectEvent*(): SelectEvent =
proc setEvent*(ev: SelectEvent) = proc setEvent*(ev: SelectEvent) =
var data: uint64 = 1 var data: uint64 = 1
if posix.write(ev.wfd, addr data, sizeof(uint64)) != sizeof(uint64): if posix.write(ev.wfd, addr data, sizeof(uint64)) != sizeof(uint64):
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
proc close*(ev: SelectEvent) = proc close*(ev: SelectEvent) =
discard posix.close(cint(ev.rfd)) if posix.close(cint(ev.rfd)) == -1:
discard posix.close(cint(ev.wfd)) raiseIOSelectorsError(osLastError())
if posix.close(cint(ev.wfd)) == -1:
raiseIOSelectorsError(osLastError())
deallocShared(cast[pointer](ev)) deallocShared(cast[pointer](ev))
template checkFd(s, f) = template checkFd(s, f) =
if f >= s.maxFD: if f >= s.maxFD:
raise newException(ValueError, "Maximum file descriptors exceeded") raiseIOSelectorsError("Maximum file descriptors exceeded!")
when hasThreadSupport: when hasThreadSupport:
template withChangeLock[T](s: Selector[T], body: untyped) = template withChangeLock[T](s: Selector[T], body: untyped) =
@ -144,31 +167,40 @@ template modifyKQueue[T](s: Selector[T], nident: uint, nfilter: cshort,
nudata: pointer) = nudata: pointer) =
mixin withChangeLock mixin withChangeLock
s.withChangeLock(): s.withChangeLock():
s.changesTable[s.changesCount] = KEvent(ident: nident, s.changes.add(KEvent(ident: nident,
filter: nfilter, flags: nflags, filter: nfilter, flags: nflags,
fflags: nfflags, data: ndata, fflags: nfflags, data: ndata,
udata: nudata) udata: nudata))
inc(s.changesCount)
if s.changesCount == MAX_KQUEUE_CHANGE_EVENTS: when not declared(CACHE_EVENTS):
if kevent(s.kqFD, addr(s.changesTable[0]), cint(s.changesCount), template flushKQueue[T](s: Selector[T]) =
nil, 0, nil) == -1: mixin withChangeLock
raiseOSError(osLastError()) s.withChangeLock():
s.changesCount = 0 let length = cint(len(s.changes))
if length > 0:
if kevent(s.kqFD, addr(s.changes[0]), length,
nil, 0, nil) == -1:
raiseIOSelectorsError(osLastError())
s.changes.setLen(0)
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) =
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
s.setKey(fdi, fdi, events, 0, data) s.setKey(fdi, events, 0, data)
if events != {}: if events != {}:
if Event.Read in events: if Event.Read in events:
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_READ, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
if Event.Write in events: if Event.Write in events:
modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_ADD, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_WRITE, EV_ADD, 0, 0, nil)
inc(s.count) inc(s.count)
when not declared(CACHE_EVENTS):
flushKQueue(s)
proc updateHandle*[T](s: Selector[T], fd: SocketHandle, proc updateHandle*[T](s: Selector[T], fd: SocketHandle,
events: set[Event]) = events: set[Event]) =
let maskEvents = {Event.Timer, Event.Signal, Event.Process, Event.Vnode, let maskEvents = {Event.Timer, Event.Signal, Event.Process, Event.Vnode,
@ -192,40 +224,41 @@ proc updateHandle*[T](s: Selector[T], fd: SocketHandle,
if (Event.Write notin pkey.events) and (Event.Write in events): if (Event.Write notin pkey.events) and (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)
when not declared(CACHE_EVENTS):
flushKQueue(s)
pkey.events = events pkey.events = events
proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool, proc registerTimer*[T](s: Selector[T], timeout: int, oneshot: bool,
data: T): int {.discardable.} = data: T): int {.discardable.} =
var fdi = posix.socket(posix.AF_INET, posix.SOCK_STREAM, let fdi = getUnique(s)
posix.IPPROTO_TCP).int
if fdi == -1:
raiseOsError(osLastError())
s.checkFd(fdi) s.checkFd(fdi)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
let events = if oneshot: {Event.Timer, Event.Oneshot} else: {Event.Timer} let events = if oneshot: {Event.Timer, Event.Oneshot} else: {Event.Timer}
let flags: cushort = if oneshot: EV_ONESHOT or EV_ADD else: EV_ADD let flags: cushort = if oneshot: EV_ONESHOT or EV_ADD else: EV_ADD
s.setKey(fdi, fdi, events, 0, data) s.setKey(fdi, events, 0, data)
# EVFILT_TIMER on Open/Net(BSD) has granularity of only milliseconds, # EVFILT_TIMER on Open/Net(BSD) has granularity of only milliseconds,
# but MacOS and FreeBSD allow use `0` as `fflags` to use milliseconds # but MacOS and FreeBSD allow use `0` as `fflags` to use milliseconds
# too # too
modifyKQueue(s, fdi.uint, EVFILT_TIMER, flags, 0, cint(timeout), nil) modifyKQueue(s, fdi.uint, EVFILT_TIMER, flags, 0, cint(timeout), nil)
when not declared(CACHE_EVENTS):
flushKQueue(s)
inc(s.count) inc(s.count)
result = fdi result = fdi
proc registerSignal*[T](s: Selector[T], signal: int, proc registerSignal*[T](s: Selector[T], signal: int,
data: T): int {.discardable.} = data: T): int {.discardable.} =
var fdi = posix.socket(posix.AF_INET, posix.SOCK_STREAM, let fdi = getUnique(s)
posix.IPPROTO_TCP).int
if fdi == -1:
raiseOsError(osLastError())
s.checkFd(fdi) s.checkFd(fdi)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
s.setKey(fdi, signal, {Event.Signal}, signal, data) s.setKey(fdi, {Event.Signal}, signal, data)
var nmask, omask: Sigset var nmask, omask: Sigset
discard sigemptyset(nmask) discard sigemptyset(nmask)
discard sigemptyset(omask) discard sigemptyset(omask)
@ -233,33 +266,44 @@ proc registerSignal*[T](s: Selector[T], signal: int,
blockSignals(nmask, omask) blockSignals(nmask, omask)
# to be compatible with linux semantic we need to "eat" signals # to be compatible with linux semantic we need to "eat" signals
posix.signal(cint(signal), SIG_IGN) posix.signal(cint(signal), SIG_IGN)
modifyKQueue(s, signal.uint, EVFILT_SIGNAL, EV_ADD, 0, 0, modifyKQueue(s, signal.uint, EVFILT_SIGNAL, EV_ADD, 0, 0,
cast[pointer](fdi)) cast[pointer](fdi))
when not declared(CACHE_EVENTS):
flushKQueue(s)
inc(s.count) inc(s.count)
result = fdi result = fdi
proc registerProcess*[T](s: Selector[T], pid: int, proc registerProcess*[T](s: Selector[T], pid: int,
data: T): int {.discardable.} = data: T): int {.discardable.} =
var fdi = posix.socket(posix.AF_INET, posix.SOCK_STREAM, let fdi = getUnique(s)
posix.IPPROTO_TCP).int
if fdi == -1:
raiseOsError(osLastError())
s.checkFd(fdi) s.checkFd(fdi)
doAssert(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, {Event.Process, Event.Oneshot}, pid, data) setKey(s, fdi, {Event.Process, Event.Oneshot}, 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))
when not declared(CACHE_EVENTS):
flushKQueue(s)
inc(s.count) inc(s.count)
result = fdi result = fdi
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
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
setKey(s, fdi, fdi, {Event.User}, 0, data) setKey(s, fdi, {Event.User}, 0, data)
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_READ, EV_ADD, 0, 0, nil)
when not declared(CACHE_EVENTS):
flushKQueue(s)
inc(s.count) inc(s.count)
template processVnodeEvents(events: set[Event]): cuint = template processVnodeEvents(events: set[Event]): cuint =
@ -281,9 +325,14 @@ template processVnodeEvents(events: set[Event]): cuint =
proc registerVnode*[T](s: Selector[T], fd: cint, events: set[Event], data: T) = proc registerVnode*[T](s: Selector[T], fd: cint, events: set[Event], data: T) =
let fdi = fd.int let fdi = fd.int
setKey(s, fdi, fdi, {Event.Vnode} + events, 0, data) setKey(s, fdi, {Event.Vnode} + events, 0, data)
var fflags = processVnodeEvents(events) var fflags = processVnodeEvents(events)
modifyKQueue(s, fdi.uint, EVFILT_VNODE, EV_ADD or EV_CLEAR, fflags, 0, nil) modifyKQueue(s, fdi.uint, EVFILT_VNODE, EV_ADD or EV_CLEAR, fflags, 0, nil)
when not declared(CACHE_EVENTS):
flushKQueue(s)
inc(s.count) inc(s.count)
proc unregister*[T](s: Selector[T], fd: int|SocketHandle) = proc unregister*[T](s: Selector[T], fd: int|SocketHandle) =
@ -295,38 +344,55 @@ proc unregister*[T](s: Selector[T], fd: int|SocketHandle) =
if pkey.events != {}: if pkey.events != {}:
if pkey.events * {Event.Read, Event.Write} != {}: if pkey.events * {Event.Read, Event.Write} != {}:
if Event.Read in pkey.events: if Event.Read in pkey.events:
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_READ, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
if Event.Write in pkey.events: if Event.Write in pkey.events:
modifyKQueue(s, fdi.uint, EVFILT_WRITE, EV_DELETE, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_WRITE, EV_DELETE, 0, 0, nil)
dec(s.count) dec(s.count)
when not declared(CACHE_EVENTS):
flushKQueue(s)
elif Event.Timer in pkey.events: elif Event.Timer in pkey.events:
discard posix.close(cint(pkey.key.fd)) if Event.Finished notin pkey.events:
modifyKQueue(s, fdi.uint, EVFILT_TIMER, EV_DELETE, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_TIMER, EV_DELETE, 0, 0, nil)
dec(s.count) when not declared(CACHE_EVENTS):
flushKQueue(s)
dec(s.count)
if posix.close(cint(pkey.ident)) == -1:
raiseIOSelectorsError(osLastError())
elif Event.Signal in pkey.events: elif Event.Signal in pkey.events:
var nmask, omask: Sigset var nmask, omask: Sigset
var signal = cint(pkey.param) let signal = cint(pkey.param)
discard sigemptyset(nmask) discard sigemptyset(nmask)
discard sigemptyset(omask) discard sigemptyset(omask)
discard sigaddset(nmask, signal) discard sigaddset(nmask, signal)
unblockSignals(nmask, omask) unblockSignals(nmask, omask)
posix.signal(signal, SIG_DFL) posix.signal(signal, SIG_DFL)
discard posix.close(cint(pkey.key.fd)) modifyKQueue(s, uint(pkey.param), EVFILT_SIGNAL, EV_DELETE, 0, 0, nil)
modifyKQueue(s, fdi.uint, EVFILT_SIGNAL, EV_DELETE, 0, 0, nil) when not declared(CACHE_EVENTS):
flushKQueue(s)
dec(s.count) dec(s.count)
if posix.close(cint(pkey.ident)) == -1:
raiseIOSelectorsError(osLastError())
elif Event.Process in pkey.events: elif Event.Process in pkey.events:
discard posix.close(cint(pkey.key.fd)) if Event.Finished notin pkey.events:
modifyKQueue(s, fdi.uint, EVFILT_PROC, EV_DELETE, 0, 0, nil) modifyKQueue(s, uint(pkey.param), EVFILT_PROC, EV_DELETE, 0, 0, nil)
dec(s.count) when not declared(CACHE_EVENTS):
flushKQueue(s)
dec(s.count)
if posix.close(cint(pkey.ident)) == -1:
raiseIOSelectorsError(osLastError())
elif Event.Vnode in pkey.events: elif Event.Vnode in pkey.events:
modifyKQueue(s, fdi.uint, EVFILT_VNODE, EV_DELETE, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_VNODE, EV_DELETE, 0, 0, nil)
when not declared(CACHE_EVENTS):
flushKQueue(s)
dec(s.count) dec(s.count)
elif Event.User in pkey.events: elif Event.User in pkey.events:
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) modifyKQueue(s, uint(fdi), EVFILT_READ, EV_DELETE, 0, 0, nil)
when not declared(CACHE_EVENTS):
flushKQueue(s)
dec(s.count) dec(s.count)
pkey.ident = 0
pkey.events = {} clearKey(pkey)
proc unregister*[T](s: Selector[T], ev: SelectEvent) = proc unregister*[T](s: Selector[T], ev: SelectEvent) =
let fdi = int(ev.rfd) let fdi = int(ev.rfd)
@ -334,26 +400,20 @@ proc unregister*[T](s: Selector[T], ev: SelectEvent) =
var pkey = addr(s.fds[fdi]) var pkey = addr(s.fds[fdi])
doAssert(pkey.ident != 0) doAssert(pkey.ident != 0)
doAssert(Event.User in pkey.events) doAssert(Event.User in pkey.events)
pkey.ident = 0
pkey.events = {} modifyKQueue(s, uint(fdi), EVFILT_READ, EV_DELETE, 0, 0, nil)
modifyKQueue(s, fdi.uint, EVFILT_READ, EV_DELETE, 0, 0, nil) when not declared(CACHE_EVENTS):
flushKQueue(s)
clearKey(pkey)
dec(s.count) dec(s.count)
proc flush*[T](s: Selector[T]) =
s.withChangeLock():
var tv = Timespec()
if kevent(s.kqFD, addr(s.changesTable[0]), cint(s.changesCount),
nil, 0, addr tv) == -1:
raiseOSError(osLastError())
s.changesCount = 0
proc selectInto*[T](s: Selector[T], timeout: int, proc selectInto*[T](s: Selector[T], timeout: int,
results: var openarray[ReadyKey[T]]): int = results: var openarray[ReadyKey]): int =
var var
tv: Timespec tv: Timespec
resTable: array[MAX_KQUEUE_RESULT_EVENTS, KEvent] resTable: array[MAX_KQUEUE_EVENTS, KEvent]
ptv = addr tv ptv = addr tv
maxres = MAX_KQUEUE_RESULT_EVENTS maxres = MAX_KQUEUE_EVENTS
if timeout != -1: if timeout != -1:
if timeout >= 1000: if timeout >= 1000:
@ -369,116 +429,147 @@ proc selectInto*[T](s: Selector[T], timeout: int,
maxres = len(results) maxres = len(results)
var count = 0 var count = 0
s.withChangeLock(): when not declared(CACHE_EVENTS):
count = kevent(s.kqFD, addr(s.changesTable[0]), cint(s.changesCount), count = kevent(s.kqFD, nil, cint(0), addr(resTable[0]), cint(maxres), ptv)
addr(resTable[0]), cint(maxres), ptv) else:
s.changesCount = 0 s.withChangeLock():
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: if count < 0:
result = 0 result = 0
let err = osLastError() let err = osLastError()
if cint(err) != EINTR: if cint(err) != EINTR:
raiseOSError(err) raiseIOSelectorsError(err)
elif count == 0: elif count == 0:
result = 0 result = 0
else: else:
var i = 0 var i = 0
var k = 0 var k = 0 # do not delete this, because `continue` used in cycle.
var pkey: ptr SelectorKey[T] var pkey: ptr SelectorKey[T]
while i < count: while i < count:
let kevent = addr(resTable[i]) let kevent = addr(resTable[i])
if (kevent.flags and EV_ERROR) == 0: var rkey = ReadyKey(fd: int(kevent.ident), events: {})
case kevent.filter:
of EVFILT_READ: if (kevent.flags and EV_ERROR) != 0:
pkey = addr(s.fds[kevent.ident.int]) rkey.events = {Event.Error}
pkey.key.events = {Event.Read}
if Event.User in pkey.events: case kevent.filter:
var data: uint64 = 0 of EVFILT_READ:
if posix.read(kevent.ident.cint, addr data, pkey = addr(s.fds[int(kevent.ident)])
sizeof(uint64)) != sizeof(uint64): rkey.events.incl(Event.Read)
let err = osLastError() if Event.User in pkey.events:
if err == OSErrorCode(EAGAIN): var data: uint64 = 0
# someone already consumed event data if posix.read(cint(kevent.ident), addr data,
inc(i) sizeof(uint64)) != sizeof(uint64):
continue let err = osLastError()
else: if err == OSErrorCode(EAGAIN):
raiseOSError(osLastError()) # someone already consumed event data
pkey.key.events = {Event.User} inc(i)
of EVFILT_WRITE: continue
pkey = addr(s.fds[kevent.ident.int]) else:
pkey.key.events = {Event.Write} raiseIOSelectorsError(err)
of EVFILT_TIMER: rkey.events = {Event.User}
pkey = addr(s.fds[kevent.ident.int]) of EVFILT_WRITE:
if Event.Oneshot in pkey.events: pkey = addr(s.fds[int(kevent.ident)])
if posix.close(cint(pkey.ident)) == -1: rkey.events.incl(Event.Write)
raiseOSError(osLastError()) rkey.events = {Event.Write}
pkey.ident = 0 of EVFILT_TIMER:
pkey.events = {} pkey = addr(s.fds[int(kevent.ident)])
dec(s.count) if Event.Oneshot in pkey.events:
pkey.key.events = {Event.Timer} # we will not clear key until it will be unregistered, so
of EVFILT_VNODE: # application can obtain data, but we will decrease counter,
pkey = addr(s.fds[kevent.ident.int]) # because kqueue is empty.
pkey.key.events = {Event.Vnode}
if (kevent.fflags and NOTE_DELETE) != 0:
pkey.key.events.incl(Event.VnodeDelete)
if (kevent.fflags and NOTE_WRITE) != 0:
pkey.key.events.incl(Event.VnodeWrite)
if (kevent.fflags and NOTE_EXTEND) != 0:
pkey.key.events.incl(Event.VnodeExtend)
if (kevent.fflags and NOTE_ATTRIB) != 0:
pkey.key.events.incl(Event.VnodeAttrib)
if (kevent.fflags and NOTE_LINK) != 0:
pkey.key.events.incl(Event.VnodeLink)
if (kevent.fflags and NOTE_RENAME) != 0:
pkey.key.events.incl(Event.VnodeRename)
if (kevent.fflags and NOTE_REVOKE) != 0:
pkey.key.events.incl(Event.VnodeRevoke)
of EVFILT_SIGNAL:
pkey = addr(s.fds[cast[int](kevent.udata)])
pkey.key.events = {Event.Signal}
of EVFILT_PROC:
pkey = addr(s.fds[cast[int](kevent.udata)])
if posix.close(cint(pkey.ident)) == -1:
raiseOSError(osLastError())
pkey.ident = 0
pkey.events = {}
dec(s.count) dec(s.count)
pkey.key.events = {Event.Process} # we are marking key with `Finished` event, to avoid double decrease.
else: pkey.events.incl(Event.Finished)
raise newException(ValueError, "Unsupported kqueue filter in queue") 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:
pkey = addr(s.fds[cast[int](kevent.udata)])
raiseIOSelectorsError("Unsupported kqueue filter in queue!")
if (kevent.flags and EV_EOF) != 0: if (kevent.flags and EV_EOF) != 0:
pkey.key.events.incl(Event.Error) rkey.events.incl(Event.Error)
results[k] = pkey.key results[k] = rkey
inc(k) inc(k)
inc(i) inc(i)
result = k result = k
proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] =
result = newSeq[ReadyKey[T]](MAX_KQUEUE_RESULT_EVENTS) result = newSeq[ReadyKey](MAX_KQUEUE_EVENTS)
let count = selectInto(s, timeout, result) let count = selectInto(s, timeout, result)
result.setLen(count) result.setLen(count)
template isEmpty*[T](s: Selector[T]): bool = template isEmpty*[T](s: Selector[T]): bool =
(s.count == 0) (s.count == 0)
template withData*[T](s: Selector[T], fd: SocketHandle, value, proc getData*[T](s: Selector[T], fd: SocketHandle|int): T =
body: untyped) = let fdi = int(fd)
s.checkFd(fdi)
if s.fds[fdi].ident != 0:
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 s.fds[fdi].ident != 0:
s.fds[fdi].data = data
result = true
template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
body: untyped) =
mixin checkFd mixin checkFd
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
if s.fds[fdi].ident != 0: if s.fds[fdi].ident != 0:
var value = addr(s.fds[fdi].key.data) var value = addr(s.fds[fdi].data)
body body
template withData*[T](s: Selector[T], fd: SocketHandle, value, body1, template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1,
body2: untyped) = body2: untyped) =
mixin checkFd mixin checkFd
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
if s.fds[fdi].ident != 0: if s.fds[fdi].ident != 0:
var value = addr(s.fds[fdi].key.data) var value = addr(s.fds[fdi].data)
body1 body1
else: else:
body2 body2

83
lib/pure/ioselects/ioselectors_poll.nim
View file

@ -12,7 +12,7 @@
import posix, times import posix, times
# Maximum number of events that can be returned # Maximum number of events that can be returned
const MAX_POLL_RESULT_EVENTS = 64 const MAX_POLL_EVENTS = 64
when hasThreadSupport: when hasThreadSupport:
type type
@ -65,7 +65,7 @@ else:
proc newSelector*[T](): Selector[T] = proc newSelector*[T](): Selector[T] =
var a = rlimit() var a = rlimit()
if getrlimit(RLIMIT_NOFILE, a) != 0: if getrlimit(RLIMIT_NOFILE, a) != 0:
raiseOsError(osLastError()) raiseIOSelectorsError(osLastError())
var maxFD = int(a.rlim_max) var maxFD = int(a.rlim_max)
when hasThreadSupport: when hasThreadSupport:
@ -87,6 +87,12 @@ proc close*[T](s: Selector[T]) =
deallocSharedArray(s.pollfds) deallocSharedArray(s.pollfds)
deallocShared(cast[pointer](s)) deallocShared(cast[pointer](s))
template clearKey[T](key: ptr SelectorKey[T]) =
var empty: T
key.ident = 0
key.events = {}
key.data = empty
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
@ -111,7 +117,7 @@ template pollUpdate[T](s: Selector[T], sock: cint, events: set[Event]) =
inc(i) inc(i)
if i == s.pollcnt: if i == s.pollcnt:
raise newException(ValueError, "Descriptor is not registered in queue") raiseIOSelectorsError("Descriptor is not registered in queue")
template pollRemove[T](s: Selector[T], sock: cint) = template pollRemove[T](s: Selector[T], sock: cint) =
withPollLock(s): withPollLock(s):
@ -134,14 +140,14 @@ template pollRemove[T](s: Selector[T], sock: cint) =
template checkFd(s, f) = template checkFd(s, f) =
if f >= s.maxFD: if f >= s.maxFD:
raise newException(ValueError, "Maximum file descriptors exceeded") raiseIOSelectorsError("Descriptor is not registered in queue")
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)
s.checkFd(fdi) s.checkFd(fdi)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
s.setKey(fdi, fdi, events, 0, data) setKey(s, fdi, events, 0, data)
if events != {}: s.pollAdd(fdi.cint, events) if events != {}: s.pollAdd(fdi.cint, events)
proc updateHandle*[T](s: Selector[T], fd: SocketHandle, proc updateHandle*[T](s: Selector[T], fd: SocketHandle,
@ -168,12 +174,10 @@ proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) =
var fdi = int(ev.rfd) var fdi = int(ev.rfd)
doAssert(s.fds[fdi].ident == 0) doAssert(s.fds[fdi].ident == 0)
var events = {Event.User} var events = {Event.User}
setKey(s, fdi, fdi, events, 0, data) setKey(s, fdi, events, 0, data)
events.incl(Event.Read) events.incl(Event.Read)
s.pollAdd(fdi.cint, events) s.pollAdd(fdi.cint, events)
proc flush*[T](s: Selector[T]) = discard
proc unregister*[T](s: Selector[T], fd: int|SocketHandle) = proc unregister*[T](s: Selector[T], fd: int|SocketHandle) =
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
@ -196,7 +200,7 @@ proc unregister*[T](s: Selector[T], ev: SelectEvent) =
proc newSelectEvent*(): SelectEvent = proc newSelectEvent*(): SelectEvent =
var fds: array[2, cint] var fds: array[2, cint]
if posix.pipe(fds) == -1: if posix.pipe(fds) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fds[0]) setNonBlocking(fds[0])
setNonBlocking(fds[1]) setNonBlocking(fds[1])
result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl)))
@ -206,16 +210,18 @@ proc newSelectEvent*(): SelectEvent =
proc setEvent*(ev: SelectEvent) = proc setEvent*(ev: SelectEvent) =
var data: uint64 = 1 var data: uint64 = 1
if posix.write(ev.wfd, addr data, sizeof(uint64)) != sizeof(uint64): if posix.write(ev.wfd, addr data, sizeof(uint64)) != sizeof(uint64):
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
proc close*(ev: SelectEvent) = proc close*(ev: SelectEvent) =
discard posix.close(cint(ev.rfd)) if posix.close(cint(ev.rfd)) == -1:
discard posix.close(cint(ev.wfd)) raiseIOSelectorsError(osLastError())
if posix.close(cint(ev.wfd)) == -1:
raiseIOSelectorsError(osLastError())
deallocShared(cast[pointer](ev)) deallocShared(cast[pointer](ev))
proc selectInto*[T](s: Selector[T], timeout: int, proc selectInto*[T](s: Selector[T], timeout: int,
results: var openarray[ReadyKey[T]]): int = results: var openarray[ReadyKey]): int =
var maxres = MAX_POLL_RESULT_EVENTS var maxres = MAX_POLL_EVENTS
if maxres > len(results): if maxres > len(results):
maxres = len(results) maxres = len(results)
@ -224,10 +230,8 @@ proc selectInto*[T](s: Selector[T], timeout: int,
if count < 0: if count < 0:
result = 0 result = 0
let err = osLastError() let err = osLastError()
if err.cint == EINTR: if cint(err) != EINTR:
discard raiseIOSelectorsError(err)
else:
raiseOSError(osLastError())
elif count == 0: elif count == 0:
result = 0 result = 0
else: else:
@ -238,58 +242,71 @@ proc selectInto*[T](s: Selector[T], timeout: int,
let revents = s.pollfds[i].revents let revents = s.pollfds[i].revents
if revents != 0: if revents != 0:
let fd = s.pollfds[i].fd let fd = s.pollfds[i].fd
var skey = addr(s.fds[fd]) var pkey = addr(s.fds[fd])
skey.key.events = {} var rkey = ReadyKey(fd: int(fd), events: {})
if (revents and POLLIN) != 0: if (revents and POLLIN) != 0:
skey.key.events.incl(Event.Read) rkey.events.incl(Event.Read)
if Event.User in skey.events: if Event.User in pkey.events:
var data: uint64 = 0 var data: uint64 = 0
if posix.read(fd, addr data, sizeof(uint64)) != sizeof(uint64): if posix.read(fd, addr data, sizeof(uint64)) != sizeof(uint64):
let err = osLastError() let err = osLastError()
if err != OSErrorCode(EAGAIN): if err != OSErrorCode(EAGAIN):
raiseOSError(osLastError()) raiseIOSelectorsError(err)
else: else:
# someone already consumed event data # someone already consumed event data
inc(i) inc(i)
continue continue
skey.key.events = {Event.User} rkey.events = {Event.User}
if (revents and POLLOUT) != 0: if (revents and POLLOUT) != 0:
skey.key.events.incl(Event.Write) rkey.events.incl(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:
skey.key.events.incl(Event.Error) rkey.events.incl(Event.Error)
results[rindex] = skey.key results[rindex] = rkey
s.pollfds[i].revents = 0 s.pollfds[i].revents = 0
inc(rindex) inc(rindex)
inc(k) inc(k)
inc(i) inc(i)
result = k result = k
proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] =
result = newSeq[ReadyKey[T]](MAX_POLL_RESULT_EVENTS) result = newSeq[ReadyKey](MAX_POLL_EVENTS)
let count = selectInto(s, timeout, result) let count = selectInto(s, timeout, result)
result.setLen(count) result.setLen(count)
template isEmpty*[T](s: Selector[T]): bool = template isEmpty*[T](s: Selector[T]): bool =
(s.count == 0) (s.count == 0)
template withData*[T](s: Selector[T], fd: SocketHandle, value, proc getData*[T](s: Selector[T], fd: SocketHandle|int): T =
let fdi = int(fd)
s.checkFd(fdi)
if s.fds[fdi].ident != 0:
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 s.fds[fdi].ident != 0:
s.fds[fdi].data = data
result = true
template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
body: untyped) = body: untyped) =
mixin checkFd mixin checkFd
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
if s.fds[fdi].ident != 0: if s.fds[fdi].ident != 0:
var value = addr(s.fds[fdi].key.data) var value = addr(s.fds[fdi].data)
body body
template withData*[T](s: Selector[T], fd: SocketHandle, value, body1, template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1,
body2: untyped) = body2: untyped) =
mixin checkFd mixin checkFd
let fdi = int(fd) let fdi = int(fd)
s.checkFd(fdi) s.checkFd(fdi)
if s.fds[fdi].ident != 0: if s.fds[fdi].ident != 0:
var value = addr(s.fds[fdi].key.data) var value = addr(s.fds[fdi].data)
body1 body1
else: else:
body2 body2

103
lib/pure/ioselects/ioselectors_select.nim
View file

@ -120,35 +120,35 @@ when defined(windows):
saddr.sin_addr.s_addr = INADDR_ANY saddr.sin_addr.s_addr = INADDR_ANY
if bindAddr(ssock, cast[ptr SockAddr](addr(saddr)), if bindAddr(ssock, cast[ptr SockAddr](addr(saddr)),
sizeof(saddr).SockLen) < 0'i32: sizeof(saddr).SockLen) < 0'i32:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
if winlean.listen(ssock, 1) == -1: if winlean.listen(ssock, 1) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
var namelen = sizeof(saddr).SockLen var namelen = sizeof(saddr).SockLen
if getsockname(ssock, cast[ptr SockAddr](addr(saddr)), if getsockname(ssock, cast[ptr SockAddr](addr(saddr)),
addr(namelen)) == -1'i32: addr(namelen)) == -1'i32:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
saddr.sin_addr.s_addr = 0x0100007F saddr.sin_addr.s_addr = 0x0100007F
if winlean.connect(wsock, cast[ptr SockAddr](addr(saddr)), if winlean.connect(wsock, cast[ptr SockAddr](addr(saddr)),
sizeof(saddr).SockLen) == -1: sizeof(saddr).SockLen) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
namelen = sizeof(saddr).SockLen namelen = sizeof(saddr).SockLen
rsock = winlean.accept(ssock, cast[ptr SockAddr](addr(saddr)), rsock = winlean.accept(ssock, cast[ptr SockAddr](addr(saddr)),
cast[ptr SockLen](addr(namelen))) cast[ptr SockLen](addr(namelen)))
if rsock == SocketHandle(-1): if rsock == SocketHandle(-1):
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
if winlean.closesocket(ssock) == -1: if winlean.closesocket(ssock) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
var mode = clong(1) var mode = clong(1)
if ioctlsocket(rsock, FIONBIO, addr(mode)) == -1: if ioctlsocket(rsock, FIONBIO, addr(mode)) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
mode = clong(1) mode = clong(1)
if ioctlsocket(wsock, FIONBIO, addr(mode)) == -1: if ioctlsocket(wsock, FIONBIO, addr(mode)) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl)))
result.rsock = rsock result.rsock = rsock
@ -158,7 +158,7 @@ when defined(windows):
var data: uint64 = 1 var data: uint64 = 1
if winlean.send(ev.wsock, cast[pointer](addr data), if winlean.send(ev.wsock, cast[pointer](addr data),
cint(sizeof(uint64)), 0) != sizeof(uint64): cint(sizeof(uint64)), 0) != sizeof(uint64):
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
proc close*(ev: SelectEvent) = proc close*(ev: SelectEvent) =
discard winlean.closesocket(ev.rsock) discard winlean.closesocket(ev.rsock)
@ -169,7 +169,7 @@ else:
proc newSelectEvent*(): SelectEvent = proc newSelectEvent*(): SelectEvent =
var fds: array[2, cint] var fds: array[2, cint]
if posix.pipe(fds) == -1: if posix.pipe(fds) == -1:
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
setNonBlocking(fds[0]) setNonBlocking(fds[0])
setNonBlocking(fds[1]) setNonBlocking(fds[1])
result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl))) result = cast[SelectEvent](allocShared0(sizeof(SelectEventImpl)))
@ -179,14 +179,17 @@ else:
proc setEvent*(ev: SelectEvent) = proc setEvent*(ev: SelectEvent) =
var data: uint64 = 1 var data: uint64 = 1
if posix.write(cint(ev.wsock), addr data, sizeof(uint64)) != sizeof(uint64): if posix.write(cint(ev.wsock), addr data, sizeof(uint64)) != sizeof(uint64):
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
proc close*(ev: SelectEvent) = proc close*(ev: SelectEvent) =
discard posix.close(cint(ev.rsock)) if posix.close(cint(ev.rsock)) == -1:
discard posix.close(cint(ev.wsock)) raiseIOSelectorsError(osLastError())
if posix.close(cint(ev.wsock)) == -1:
raiseIOSelectorsError(osLastError())
deallocShared(cast[pointer](ev)) deallocShared(cast[pointer](ev))
proc setKey[T](s: Selector[T], fd: SocketHandle, events: set[Event], data: T) = proc setSelectKey[T](s: Selector[T], fd: SocketHandle, events: set[Event],
data: T) =
var i = 0 var i = 0
let fdi = int(fd) let fdi = int(fd)
while i < FD_SETSIZE: while i < FD_SETSIZE:
@ -194,13 +197,11 @@ proc setKey[T](s: Selector[T], fd: SocketHandle, events: set[Event], data: T) =
var pkey = addr(s.fds[i]) var pkey = addr(s.fds[i])
pkey.ident = fdi pkey.ident = fdi
pkey.events = events pkey.events = events
pkey.key.fd = fd.int pkey.data = data
pkey.key.events = {}
pkey.key.data = data
break break
inc(i) inc(i)
if i == FD_SETSIZE: if i == FD_SETSIZE:
raise newException(ValueError, "Maximum numbers of fds exceeded") raiseIOSelectorsError("Maximum numbers of fds exceeded")
proc getKey[T](s: Selector[T], fd: SocketHandle): ptr SelectorKey[T] = proc getKey[T](s: Selector[T], fd: SocketHandle): ptr SelectorKey[T] =
var i = 0 var i = 0
@ -210,24 +211,28 @@ proc getKey[T](s: Selector[T], fd: SocketHandle): ptr SelectorKey[T] =
result = addr(s.fds[i]) result = addr(s.fds[i])
break break
inc(i) inc(i)
doAssert(i < FD_SETSIZE, "Descriptor not registered in queue") if i == FD_SETSIZE:
raiseIOSelectorsError("Descriptor not registered in queue")
proc delKey[T](s: Selector[T], fd: SocketHandle) = proc delKey[T](s: Selector[T], fd: SocketHandle) =
var empty: T
var i = 0 var i = 0
while i < FD_SETSIZE: while i < FD_SETSIZE:
if s.fds[i].ident == fd.int: if s.fds[i].ident == fd.int:
s.fds[i].ident = 0 s.fds[i].ident = 0
s.fds[i].events = {} s.fds[i].events = {}
s.fds[i].data = empty
break break
inc(i) inc(i)
doAssert(i < FD_SETSIZE, "Descriptor not registered in queue") if i == FD_SETSIZE:
raiseIOSelectorsError("Descriptor not registered in queue")
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) =
when not defined(windows): when not defined(windows):
let fdi = int(fd) let fdi = int(fd)
s.withSelectLock(): s.withSelectLock():
s.setKey(fd, events, data) s.setSelectKey(fd, events, data)
when not defined(windows): when not defined(windows):
if fdi > s.maxFD: s.maxFD = fdi if fdi > s.maxFD: s.maxFD = fdi
if Event.Read in events: if Event.Read in events:
@ -242,7 +247,7 @@ proc registerEvent*[T](s: Selector[T], ev: SelectEvent, data: T) =
when not defined(windows): when not defined(windows):
let fdi = int(ev.rsock) let fdi = int(ev.rsock)
s.withSelectLock(): s.withSelectLock():
s.setKey(ev.rsock, {Event.User}, data) s.setSelectKey(ev.rsock, {Event.User}, data)
when not defined(windows): when not defined(windows):
if fdi > s.maxFD: s.maxFD = fdi if fdi > s.maxFD: s.maxFD = fdi
IOFD_SET(ev.rsock, addr s.rSet) IOFD_SET(ev.rsock, addr s.rSet)
@ -292,7 +297,7 @@ proc unregister*[T](s: Selector[T], ev: SelectEvent) =
s.delKey(fd) s.delKey(fd)
proc selectInto*[T](s: Selector[T], timeout: int, proc selectInto*[T](s: Selector[T], timeout: int,
results: var openarray[ReadyKey[T]]): int = results: var openarray[ReadyKey]): int =
var tv = Timeval() var tv = Timeval()
var ptv = addr tv var ptv = addr tv
var rset, wset, eset: FdSet var rset, wset, eset: FdSet
@ -313,11 +318,11 @@ proc selectInto*[T](s: Selector[T], timeout: int,
if count < 0: if count < 0:
result = 0 result = 0
when defined(windows): when defined(windows):
raiseOSError(osLastError()) raiseIOSelectorsError(osLastError())
else: else:
let err = osLastError() let err = osLastError()
if cint(err) != EINTR: if cint(err) != EINTR:
raiseOSError(err) raiseIOSelectorsError(err)
elif count == 0: elif count == 0:
result = 0 result = 0
else: else:
@ -329,7 +334,7 @@ proc selectInto*[T](s: Selector[T], timeout: int,
if s.fds[i].ident != 0: if s.fds[i].ident != 0:
var flag = false var flag = false
var pkey = addr(s.fds[i]) var pkey = addr(s.fds[i])
pkey.key.events = {} var rkey = ReadyKey(fd: int(pkey.ident), events: {})
let fd = SocketHandle(pkey.ident) let fd = SocketHandle(pkey.ident)
if IOFD_ISSET(fd, addr rset) != 0: if IOFD_ISSET(fd, addr rset) != 0:
if Event.User in pkey.events: if Event.User in pkey.events:
@ -338,31 +343,31 @@ proc selectInto*[T](s: Selector[T], timeout: int,
sizeof(uint64).cint, 0) != sizeof(uint64): sizeof(uint64).cint, 0) != sizeof(uint64):
let err = osLastError() let err = osLastError()
if cint(err) != EAGAIN: if cint(err) != EAGAIN:
raiseOSError(err) raiseIOSelectorsError(err)
else: else:
inc(i) inc(i)
inc(k) inc(k)
continue continue
else: else:
flag = true flag = true
pkey.key.events = {Event.User} rkey.events = {Event.User}
else: else:
flag = true flag = true
pkey.key.events = {Event.Read} rkey.events = {Event.Read}
if IOFD_ISSET(fd, addr wset) != 0: if IOFD_ISSET(fd, addr wset) != 0:
pkey.key.events.incl(Event.Write) rkey.events.incl(Event.Write)
if IOFD_ISSET(fd, addr eset) != 0: if IOFD_ISSET(fd, addr eset) != 0:
pkey.key.events.incl(Event.Error) rkey.events.incl(Event.Error)
flag = true flag = true
if flag: if flag:
results[rindex] = pkey.key results[rindex] = rkey
inc(rindex) inc(rindex)
inc(k) inc(k)
inc(i) inc(i)
result = rindex result = rindex
proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey[T]] = proc select*[T](s: Selector[T], timeout: int): seq[ReadyKey] =
result = newSeq[ReadyKey[T]](FD_SETSIZE) result = newSeq[ReadyKey](FD_SETSIZE)
var count = selectInto(s, timeout, result) var count = selectInto(s, timeout, result)
result.setLen(count) result.setLen(count)
@ -383,7 +388,28 @@ else:
template withSelectLock[T](s: Selector[T], body: untyped) = template withSelectLock[T](s: Selector[T], body: untyped) =
body body
template withData*[T](s: Selector[T], fd: SocketHandle, value, proc getData*[T](s: Selector[T], fd: SocketHandle|int): T =
s.withSelectLock():
let fdi = int(fd)
var i = 0
while i < FD_SETSIZE:
if s.fds[i].ident == fdi:
result = s.fds[i].data
break
inc(i)
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) = body: untyped) =
mixin withSelectLock mixin withSelectLock
s.withSelectLock(): s.withSelectLock():
@ -392,13 +418,13 @@ template withData*[T](s: Selector[T], fd: SocketHandle, value,
var i = 0 var i = 0
while i < FD_SETSIZE: while i < FD_SETSIZE:
if s.fds[i].ident == fdi: if s.fds[i].ident == fdi:
value = addr(s.fds[i].key.data) value = addr(s.fds[i].data)
break break
inc(i) inc(i)
if i != FD_SETSIZE: if i != FD_SETSIZE:
body body
template withData*[T](s: Selector[T], fd: SocketHandle, value, template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
body1, body2: untyped) = body1, body2: untyped) =
mixin withSelectLock mixin withSelectLock
s.withSelectLock(): s.withSelectLock():
@ -407,10 +433,11 @@ template withData*[T](s: Selector[T], fd: SocketHandle, value,
var i = 0 var i = 0
while i < FD_SETSIZE: while i < FD_SETSIZE:
if s.fds[i].ident == fdi: if s.fds[i].ident == fdi:
value = addr(s.fds[i].key.data) value = addr(s.fds[i].data)
break break
inc(i) inc(i)
if i != FD_SETSIZE: if i != FD_SETSIZE:
body1 body1
else: else:
body2 body2

177
lib/upcoming/asyncdispatch.nim
View file

@ -526,10 +526,10 @@ when defined(windows) or defined(nimdoc):
proc send*(socket: AsyncFD, buf: pointer, size: int, proc send*(socket: AsyncFD, buf: pointer, size: int,
flags = {SocketFlag.SafeDisconn}): Future[void] = flags = {SocketFlag.SafeDisconn}): Future[void] =
## Sends ``size`` bytes from ``buf`` to ``socket``. The returned future will complete once all ## Sends ``size`` bytes from ``buf`` to ``socket``. The returned future
## data has been sent. ## will complete once all data has been sent.
## **WARNING**: Use it with caution. If ``buf`` refers to GC'ed object, you must use GC_ref/GC_unref calls ## **WARNING**: Use it with caution. If ``buf`` refers to GC'ed object,
## to avoid early freeing of the buffer ## you must use GC_ref/GC_unref calls to avoid early freeing of the buffer.
verifyPresence(socket) verifyPresence(socket)
var retFuture = newFuture[void]("send") var retFuture = newFuture[void]("send")
@ -946,7 +946,8 @@ when defined(windows) or defined(nimdoc):
## receiving notifies. ## receiving notifies.
registerWaitableEvent(fd, cb, FD_WRITE or FD_CONNECT or FD_CLOSE) registerWaitableEvent(fd, cb, FD_WRITE or FD_CONNECT or FD_CLOSE)
template registerWaitableHandle(p, hEvent, flags, pcd, timeout, handleCallback) = template registerWaitableHandle(p, hEvent, flags, pcd, timeout,
handleCallback) =
let handleFD = AsyncFD(hEvent) let handleFD = AsyncFD(hEvent)
pcd.ioPort = p.ioPort pcd.ioPort = p.ioPort
pcd.handleFd = handleFD pcd.handleFd = handleFD
@ -961,7 +962,7 @@ when defined(windows) or defined(nimdoc):
pcd.ovl = ol pcd.ovl = ol
if not registerWaitForSingleObject(addr(pcd.waitFd), hEvent, if not registerWaitForSingleObject(addr(pcd.waitFd), hEvent,
cast[WAITORTIMERCALLBACK](waitableCallback), cast[WAITORTIMERCALLBACK](waitableCallback),
cast[pointer](pcd), timeout.Dword, flags): cast[pointer](pcd), timeout.Dword, flags):
GC_unref(ol) GC_unref(ol)
deallocShared(cast[pointer](pcd)) deallocShared(cast[pointer](pcd))
discard closeHandle(hEvent) discard closeHandle(hEvent)
@ -1098,15 +1099,18 @@ else:
import ioselectors import ioselectors
from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK, from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK,
MSG_NOSIGNAL MSG_NOSIGNAL
const
InitCallbackListSize = 4 # initial size of callbacks sequence,
# associated with file/socket descriptor.
InitDelayedCallbackListSize = 64 # initial size of delayed callbacks
# queue.
type type
AsyncFD* = distinct cint AsyncFD* = distinct cint
Callback = proc (fd: AsyncFD): bool {.closure,gcsafe.} Callback = proc (fd: AsyncFD): bool {.closure,gcsafe.}
DoublyLinkedListRef = ref DoublyLinkedList[Callback]
AsyncData = object AsyncData = object
readCBs: DoublyLinkedListRef readList: seq[Callback]
writeCBs: DoublyLinkedListRef writeList: seq[Callback]
AsyncEvent* = distinct SelectEvent AsyncEvent* = distinct SelectEvent
@ -1117,11 +1121,17 @@ else:
proc `==`*(x, y: AsyncFD): bool {.borrow.} proc `==`*(x, y: AsyncFD): bool {.borrow.}
proc `==`*(x, y: AsyncEvent): bool {.borrow.} proc `==`*(x, y: AsyncEvent): bool {.borrow.}
template newAsyncData(): AsyncData =
AsyncData(
readList: newSeqOfCap[Callback](InitCallbackListSize),
writeList: newSeqOfCap[Callback](InitCallbackListSize)
)
proc newDispatcher*(): PDispatcher = proc newDispatcher*(): PDispatcher =
new result new result
result.selector = newSelector[AsyncData]() result.selector = newSelector[AsyncData]()
result.timers.newHeapQueue() result.timers.newHeapQueue()
result.callbacks = initDeque[proc ()](64) result.callbacks = initDeque[proc ()](InitDelayedCallbackListSize)
var gDisp{.threadvar.}: PDispatcher ## Global dispatcher var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
proc getGlobalDispatcher*(): PDispatcher = proc getGlobalDispatcher*(): PDispatcher =
@ -1130,10 +1140,7 @@ else:
proc register*(fd: AsyncFD) = proc register*(fd: AsyncFD) =
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var data = AsyncData( var data = newAsyncData()
readCBs: DoublyLinkedListRef(),
writeCBs: DoublyLinkedListRef()
)
p.selector.registerHandle(fd.SocketHandle, {}, data) p.selector.registerHandle(fd.SocketHandle, {}, data)
proc newAsyncNativeSocket*(domain: cint, sockType: cint, proc newAsyncNativeSocket*(domain: cint, sockType: cint,
@ -1168,10 +1175,9 @@ else:
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var newEvents = {Event.Read} var newEvents = {Event.Read}
withData(p.selector, fd.SocketHandle, adata) do: withData(p.selector, fd.SocketHandle, adata) do:
adata.readCBs[].append(cb) adata.readList.add(cb)
newEvents.incl(Event.Read) newEvents.incl(Event.Read)
if not isNil(adata.writeCBs.head): if len(adata.writeList) != 0: newEvents.incl(Event.Write)
newEvents.incl(Event.Write)
do: do:
raise newException(ValueError, "File descriptor not registered.") raise newException(ValueError, "File descriptor not registered.")
p.selector.updateHandle(fd.SocketHandle, newEvents) p.selector.updateHandle(fd.SocketHandle, newEvents)
@ -1180,10 +1186,9 @@ else:
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var newEvents = {Event.Write} var newEvents = {Event.Write}
withData(p.selector, fd.SocketHandle, adata) do: withData(p.selector, fd.SocketHandle, adata) do:
adata.writeCBs[].append(cb) adata.writeList.add(cb)
newEvents.incl(Event.Write) newEvents.incl(Event.Write)
if not isNil(adata.readCBs.head): if len(adata.readList) != 0: newEvents.incl(Event.Read)
newEvents.incl(Event.Read)
do: do:
raise newException(ValueError, "File descriptor not registered.") raise newException(ValueError, "File descriptor not registered.")
p.selector.updateHandle(fd.SocketHandle, newEvents) p.selector.updateHandle(fd.SocketHandle, newEvents)
@ -1192,13 +1197,65 @@ else:
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
not p.selector.isEmpty() or p.timers.len != 0 or p.callbacks.len != 0 not p.selector.isEmpty() or p.timers.len != 0 or p.callbacks.len != 0
template processBasicCallbacks(ident, rwlist: untyped) =
# Process pending descriptor's callbacks.
# Invoke every callback stored in `rwlist`, until first one
# returned `false`, which means callback wants to stay
# alive. In such case all remaining callbacks will be added
# to `rwlist` again, in the order they have been inserted.
#
# `rwlist` associated with file descriptor MUST BE emptied before
# dispatching callback (See https://github.com/nim-lang/Nim/issues/5128),
# or it can be possible to fall into endless cycle.
var curList: seq[Callback]
withData(p.selector, ident, adata) do:
shallowCopy(curList, adata.rwlist)
adata.rwlist = newSeqOfCap[Callback](InitCallbackListSize)
let newLength = max(len(curList), InitCallbackListSize)
var newList = newSeqOfCap[Callback](newLength)
for cb in curList:
if len(newList) > 0:
newList.add(cb)
else:
if not cb(fd.AsyncFD):
newList.add(cb)
withData(p.selector, ident, adata) do:
adata.rwlist = newList & adata.rwlist
template processCustomCallbacks(ident: untyped) =
# Process pending custom event callbacks. Custom events are
# {Event.Timer, Event.Signal, Event.Process, Event.Vnode}.
# There can be only one callback registered with one descriptor,
# so there no need to iterate over list.
var curList: seq[Callback]
withData(p.selector, ident, adata) do:
shallowCopy(curList, adata.readList)
adata.readList = newSeqOfCap[Callback](InitCallbackListSize)
let newLength = len(curList)
var newList = newSeqOfCap[Callback](newLength)
var cb = curList[0]
if not cb(fd.AsyncFD):
newList.add(cb)
else:
p.selector.unregister(fd)
withData(p.selector, ident, adata) do:
adata.readList = newList & adata.readList
proc poll*(timeout = 500) = proc poll*(timeout = 500) =
var keys: array[64, ReadyKey[AsyncData]] var keys: array[64, ReadyKey]
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
when ioselSupportedPlatform: when ioselSupportedPlatform:
let customSet = {Event.Timer, Event.Signal, Event.Process, let customSet = {Event.Timer, Event.Signal, Event.Process,
Event.Vnode, Event.User} Event.Vnode}
if p.selector.isEmpty() and p.timers.len == 0 and p.callbacks.len == 0: if p.selector.isEmpty() and p.timers.len == 0 and p.callbacks.len == 0:
raise newException(ValueError, raise newException(ValueError,
@ -1209,45 +1266,23 @@ else:
var i = 0 var i = 0
while i < count: while i < count:
var custom = false var custom = false
var fd = keys[i].fd.SocketHandle let fd = keys[i].fd
let events = keys[i].events let events = keys[i].events
if Event.Read in events or events == {Event.Error}: if Event.Read in events or events == {Event.Error}:
for node in keys[i].data.readCBs[].nodes(): processBasicCallbacks(fd, readList)
let cb = node.value
if cb != nil:
if cb(fd.AsyncFD):
keys[i].data.readCBs[].remove(node)
else:
break
if Event.Write in events or events == {Event.Error}: if Event.Write in events or events == {Event.Error}:
for node in keys[i].data.writeCBs[].nodes(): processBasicCallbacks(fd, writeList)
let cb = node.value
if cb != nil: if Event.User in events or events == {Event.Error}:
if cb(fd.AsyncFD): custom = true
keys[i].data.writeCBs[].remove(node) processBasicCallbacks(fd, readList)
else:
break
when ioselSupportedPlatform: when ioselSupportedPlatform:
if (customSet * events) != {}: if (customSet * events) != {}:
for node in keys[i].data.readCBs[].nodes(): custom = true
let cb = node.value processCustomCallbacks(fd)
doAssert(cb != nil)
custom = true
if cb(fd.AsyncFD):
keys[i].data.readCBs[].remove(node)
p.selector.unregister(fd)
else:
if Event.User in events or events == {Event.Error}:
for node in keys[i].data.readCBs[].nodes():
let cb = node.value
custom = true
if cb != nil:
if cb(fd.AsyncFD):
keys[i].data.readCBs[].remove(node)
p.selector.unregister(fd)
# because state `data` can be modified in callback we need to update # because state `data` can be modified in callback we need to update
# descriptor events with currently registered callbacks. # descriptor events with currently registered callbacks.
@ -1255,11 +1290,11 @@ else:
var update = false var update = false
var newEvents: set[Event] = {} var newEvents: set[Event] = {}
p.selector.withData(fd, adata) do: p.selector.withData(fd, adata) do:
if not isNil(adata.readCBs.head): incl(newEvents, Event.Read) if len(adata.readList) > 0: incl(newEvents, Event.Read)
if not isNil(adata.writeCBs.head): incl(newEvents, Event.Write) if len(adata.writeList) > 0: incl(newEvents, Event.Write)
update = true update = true
if update: if update:
p.selector.updateHandle(fd, newEvents) p.selector.updateHandle(SocketHandle(fd), newEvents)
inc(i) inc(i)
# Timer processing. # Timer processing.
@ -1519,33 +1554,24 @@ else:
## ``oneshot`` - if ``true`` only one event will be dispatched, ## ``oneshot`` - if ``true`` only one event will be dispatched,
## if ``false`` continuous events every ``timeout`` milliseconds. ## if ``false`` continuous events every ``timeout`` milliseconds.
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var data = AsyncData( var data = newAsyncData()
readCBs: DoublyLinkedListRef(), data.readList.add(cb)
writeCBs: DoublyLinkedListRef()
)
data.readCBs[].append(cb)
p.selector.registerTimer(timeout, oneshot, data) p.selector.registerTimer(timeout, oneshot, data)
proc addSignal*(signal: int, cb: Callback) = proc addSignal*(signal: int, cb: Callback) =
## Start watching signal ``signal``, and when signal appears, call the ## Start watching signal ``signal``, and when signal appears, call the
## callback ``cb``. ## callback ``cb``.
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var data = AsyncData( var data = newAsyncData()
readCBs: DoublyLinkedListRef(), data.readList.add(cb)
writeCBs: DoublyLinkedListRef()
)
data.readCBs[].append(cb)
p.selector.registerSignal(signal, data) p.selector.registerSignal(signal, data)
proc addProcess*(pid: int, cb: Callback) = proc addProcess*(pid: int, cb: Callback) =
## Start watching for process exit with pid ``pid``, and then call ## Start watching for process exit with pid ``pid``, and then call
## the callback ``cb``. ## the callback ``cb``.
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var data = AsyncData( var data = newAsyncData()
readCBs: DoublyLinkedListRef(), data.readList.add(cb)
writeCBs: DoublyLinkedListRef()
)
data.readCBs[].append(cb)
p.selector.registerProcess(pid, data) p.selector.registerProcess(pid, data)
proc newAsyncEvent*(): AsyncEvent = proc newAsyncEvent*(): AsyncEvent =
@ -1564,11 +1590,8 @@ else:
## Start watching for event ``ev``, and call callback ``cb``, when ## Start watching for event ``ev``, and call callback ``cb``, when
## ev will be set to signaled state. ## ev will be set to signaled state.
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
var data = AsyncData( var data = newAsyncData()
readCBs: DoublyLinkedListRef(), data.readList.add(cb)
writeCBs: DoublyLinkedListRef()
)
data.readCBs[].append(cb)
p.selector.registerEvent(SelectEvent(ev), data) p.selector.registerEvent(SelectEvent(ev), data)
proc sleepAsync*(ms: int): Future[void] = proc sleepAsync*(ms: int): Future[void] =

53
tests/async/tioselectors.nim
View file

@ -37,9 +37,6 @@ 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, {Event.Read}, 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),
addr(option), sizeof(option).SockLen) < 0: addr(option), sizeof(option).SockLen) < 0:
@ -50,16 +47,19 @@ when not defined(windows):
aiList.ai_addrlen.Socklen) < 0'i32: aiList.ai_addrlen.Socklen) < 0'i32:
dealloc(aiList) dealloc(aiList)
raiseOSError(osLastError()) raiseOSError(osLastError())
discard server_socket.listen() if server_socket.listen() == -1:
raiseOSError(osLastError())
dealloc(aiList) dealloc(aiList)
aiList = getAddrInfo("127.0.0.1", Port(13337)) aiList = getAddrInfo("127.0.0.1", Port(13337))
discard posix.connect(client_socket, aiList.ai_addr, discard posix.connect(client_socket, aiList.ai_addr,
aiList.ai_addrlen.Socklen) aiList.ai_addrlen.Socklen)
registerHandle(selector, server_socket, {Event.Read}, 0)
registerHandle(selector, client_socket, {Event.Write}, 0)
dealloc(aiList) dealloc(aiList)
discard selector.select(100) discard selector.select(100)
var rc1 = selector.select(100)
assert(len(rc1) == 2)
var sockAddress: SockAddr var sockAddress: SockAddr
var addrLen = sizeof(sockAddress).Socklen var addrLen = sizeof(sockAddress).Socklen
@ -126,15 +126,15 @@ when not defined(windows):
var selector = newSelector[int]() var selector = newSelector[int]()
var event = newSelectEvent() var event = newSelectEvent()
selector.registerEvent(event, 1) selector.registerEvent(event, 1)
selector.flush() var rc0 = selector.select(0)
event.setEvent() event.setEvent()
var rc1 = selector.select(0) var rc1 = selector.select(0)
event.setEvent() event.setEvent()
var rc2 = selector.select(0) var rc2 = selector.select(0)
var rc3 = selector.select(0) var rc3 = selector.select(0)
assert(len(rc1) == 1 and len(rc2) == 1 and len(rc3) == 0) assert(len(rc0) == 0 and len(rc1) == 1 and len(rc2) == 1 and len(rc3) == 0)
var ev1 = rc1[0].data var ev1 = selector.getData(rc1[0].fd)
var ev2 = rc2[0].data var ev2 = selector.getData(rc2[0].fd)
assert(ev1 == 1 and ev2 == 1) assert(ev1 == 1 and ev2 == 1)
selector.unregister(event) selector.unregister(event)
event.close() event.close()
@ -150,11 +150,11 @@ when not defined(windows):
var rc2 = selector.select(140) var rc2 = selector.select(140)
assert(len(rc1) == 1 and len(rc2) == 1) assert(len(rc1) == 1 and len(rc2) == 1)
selector.unregister(timer) selector.unregister(timer)
selector.flush() discard selector.select(0)
selector.registerTimer(100, true, 0) selector.registerTimer(100, true, 0)
var rc3 = selector.select(120)
var rc4 = selector.select(120) var rc4 = selector.select(120)
assert(len(rc3) == 1 and len(rc4) == 0) var rc5 = selector.select(120)
assert(len(rc4) == 1 and len(rc5) == 0)
assert(selector.isEmpty()) assert(selector.isEmpty())
selector.close() selector.close()
result = true result = true
@ -193,12 +193,14 @@ when not defined(windows):
var s1 = selector.registerSignal(SIGUSR1, 1) var s1 = selector.registerSignal(SIGUSR1, 1)
var s2 = selector.registerSignal(SIGUSR2, 2) var s2 = selector.registerSignal(SIGUSR2, 2)
var s3 = selector.registerSignal(SIGTERM, 3) var s3 = selector.registerSignal(SIGTERM, 3)
selector.flush() discard selector.select(0)
discard posix.kill(pid, SIGUSR1) discard posix.kill(pid, SIGUSR1)
discard posix.kill(pid, SIGUSR2) discard posix.kill(pid, SIGUSR2)
discard posix.kill(pid, SIGTERM) discard posix.kill(pid, SIGTERM)
var rc = selector.select(0) var rc = selector.select(0)
var cd0 = selector.getData(rc[0].fd)
var cd1 = selector.getData(rc[1].fd)
var cd2 = selector.getData(rc[2].fd)
selector.unregister(s1) selector.unregister(s1)
selector.unregister(s2) selector.unregister(s2)
selector.unregister(s3) selector.unregister(s3)
@ -211,7 +213,7 @@ when not defined(windows):
raiseOSError(osLastError()) raiseOSError(osLastError())
assert(len(rc) == 3) assert(len(rc) == 3)
assert(rc[0].data + rc[1].data + rc[2].data == 6) # 1 + 2 + 3 assert(cd0 + cd1 + cd2 == 6, $(cd0 + cd1 + cd2)) # 1 + 2 + 3
assert(equalMem(addr sigset1o, addr sigset2o, sizeof(Sigset))) assert(equalMem(addr sigset1o, addr sigset2o, sizeof(Sigset)))
assert(selector.isEmpty()) assert(selector.isEmpty())
result = true result = true
@ -286,8 +288,8 @@ when not defined(windows):
events: set[Event] events: set[Event]
proc vnode_test(): bool = proc vnode_test(): bool =
proc validate[T](test: openarray[ReadyKey[T]], proc validate(test: openarray[ReadyKey],
check: openarray[valType]): bool = check: openarray[valType]): bool =
result = false result = false
if len(test) == len(check): if len(test) == len(check):
for checkItem in check: for checkItem in check:
@ -300,7 +302,7 @@ when not defined(windows):
if not result: if not result:
break break
var res: seq[ReadyKey[int]] var res: seq[ReadyKey]
var selector = newSelector[int]() var selector = newSelector[int]()
var events = {Event.VnodeWrite, Event.VnodeDelete, Event.VnodeExtend, var events = {Event.VnodeWrite, Event.VnodeDelete, Event.VnodeExtend,
Event.VnodeAttrib, Event.VnodeLink, Event.VnodeRename, Event.VnodeAttrib, Event.VnodeLink, Event.VnodeRename,
@ -315,7 +317,7 @@ when not defined(windows):
raiseOsError(osLastError()) raiseOsError(osLastError())
selector.registerVnode(dirfd, events, 1) selector.registerVnode(dirfd, events, 1)
selector.flush() discard selector.select(0)
# chmod testDirectory to 0777 # chmod testDirectory to 0777
chmodPath(testDirectory, 0x1FF) chmodPath(testDirectory, 0x1FF)
@ -337,7 +339,6 @@ when not defined(windows):
# open test directory for watching # open test directory for watching
var testfd = openWatch(testDirectory & "/test") var testfd = openWatch(testDirectory & "/test")
selector.registerVnode(testfd, events, 2) selector.registerVnode(testfd, events, 2)
selector.flush()
doAssert(len(selector.select(0)) == 0) doAssert(len(selector.select(0)) == 0)
# rename test directory # rename test directory
@ -381,7 +382,7 @@ when not defined(windows):
testfd = openWatch(testDirectory & "/testfile") testfd = openWatch(testDirectory & "/testfile")
selector.registerVnode(testfd, events, 1) selector.registerVnode(testfd, events, 1)
selector.flush() discard selector.select(0)
# write data to test file # write data to test file
writeFile(testDirectory & "/testfile", "TESTDATA") writeFile(testDirectory & "/testfile", "TESTDATA")
@ -433,7 +434,6 @@ when not defined(windows):
proc event_wait_thread(event: SelectEvent) {.thread.} = proc event_wait_thread(event: SelectEvent) {.thread.} =
var selector = newSelector[int]() var selector = newSelector[int]()
selector.registerEvent(event, 1) selector.registerEvent(event, 1)
selector.flush()
var rc = selector.select(1000) var rc = selector.select(1000)
if len(rc) == 1: if len(rc) == 1:
inc(counter) inc(counter)
@ -573,15 +573,15 @@ else:
var selector = newSelector[int]() var selector = newSelector[int]()
var event = newSelectEvent() var event = newSelectEvent()
selector.registerEvent(event, 1) selector.registerEvent(event, 1)
selector.flush() discard selector.select(0)
event.setEvent() event.setEvent()
var rc1 = selector.select(0) var rc1 = selector.select(0)
event.setEvent() event.setEvent()
var rc2 = selector.select(0) var rc2 = selector.select(0)
var rc3 = selector.select(0) var rc3 = selector.select(0)
assert(len(rc1) == 1 and len(rc2) == 1 and len(rc3) == 0) assert(len(rc1) == 1 and len(rc2) == 1 and len(rc3) == 0)
var ev1 = rc1[0].data var ev1 = selector.getData(rc1[0].fd)
var ev2 = rc2[0].data var ev2 = selector.getData(rc2[0].fd)
assert(ev1 == 1 and ev2 == 1) assert(ev1 == 1 and ev2 == 1)
selector.unregister(event) selector.unregister(event)
event.close() event.close()
@ -595,7 +595,6 @@ else:
proc event_wait_thread(event: SelectEvent) {.thread.} = proc event_wait_thread(event: SelectEvent) {.thread.} =
var selector = newSelector[int]() var selector = newSelector[int]()
selector.registerEvent(event, 1) selector.registerEvent(event, 1)
selector.flush()
var rc = selector.select(500) var rc = selector.select(500)
if len(rc) == 1: if len(rc) == 1:
inc(counter) inc(counter)

1
tests/async/tupcoming_async.nim
View file

@ -1,5 +1,4 @@
discard """ discard """
cmd: "nim c -r -f $file"
output: ''' output: '''
OK OK
OK OK