some progress on making async multithreaded
This commit is contained in:
Araq
parent
c5a479d536
commit
21ea8e6913
6 changed files with 504 additions and 243 deletions
|
|
@ -11,11 +11,11 @@
|
|||
|
||||
import tables, os, unsigned, hashes
|
||||
|
||||
when defined(linux):
|
||||
when defined(linux):
|
||||
import posix, epoll
|
||||
elif defined(windows):
|
||||
elif defined(windows):
|
||||
import winlean
|
||||
else:
|
||||
else:
|
||||
import posix
|
||||
|
||||
proc hash*(x: SocketHandle): THash {.borrow.}
|
||||
|
|
@ -25,22 +25,23 @@ type
|
|||
Event* = enum
|
||||
EvRead, EvWrite, EvError
|
||||
|
||||
SelectorKey* = ref object
|
||||
SelectorKey* = object
|
||||
fd*: SocketHandle
|
||||
events*: set[Event] ## The events which ``fd`` listens for.
|
||||
data*: RootRef ## User object.
|
||||
data*: pointer ## User object.
|
||||
|
||||
ReadyInfo* = tuple[key: SelectorKey, events: set[Event]]
|
||||
|
||||
|
||||
|
||||
when defined(nimdoc):
|
||||
type
|
||||
Selector* = ref object
|
||||
## An object which holds file descriptors to be checked for read/write
|
||||
## status.
|
||||
fds: Table[SocketHandle, SelectorKey]
|
||||
|
||||
proc register*(s: Selector, fd: SocketHandle, events: set[Event],
|
||||
data: RootRef): SelectorKey {.discardable.} =
|
||||
data: pointer): SelectorKey {.discardable.} =
|
||||
## Registers file descriptor ``fd`` to selector ``s`` with a set of TEvent
|
||||
## ``events``.
|
||||
|
||||
|
|
@ -57,7 +58,7 @@ when defined(nimdoc):
|
|||
proc select*(s: Selector, timeout: int): seq[ReadyInfo] =
|
||||
## The ``events`` field of the returned ``key`` contains the original events
|
||||
## for which the ``fd`` was bound. This is contrary to the ``events`` field
|
||||
## of the ``TReadyInfo`` tuple which determines which events are ready
|
||||
## of the ``ReadyInfo`` tuple which determines which events are ready
|
||||
## on the ``fd``.
|
||||
|
||||
proc newSelector*(): Selector =
|
||||
|
|
@ -72,11 +73,11 @@ when defined(nimdoc):
|
|||
|
||||
elif defined(linux):
|
||||
type
|
||||
Selector* = ref object
|
||||
Selector* = object
|
||||
epollFD: cint
|
||||
events: array[64, epoll_event]
|
||||
fds: Table[SocketHandle, SelectorKey]
|
||||
|
||||
fds: SharedTable[SocketHandle, SelectorKey]
|
||||
|
||||
proc createEventStruct(events: set[Event], fd: SocketHandle): epoll_event =
|
||||
if EvRead in events:
|
||||
result.events = EPOLLIN
|
||||
|
|
@ -84,21 +85,19 @@ elif defined(linux):
|
|||
result.events = result.events or EPOLLOUT
|
||||
result.events = result.events or EPOLLRDHUP
|
||||
result.data.fd = fd.cint
|
||||
|
||||
|
||||
proc register*(s: Selector, fd: SocketHandle, events: set[Event],
|
||||
data: RootRef): SelectorKey {.discardable.} =
|
||||
data: pointer) =
|
||||
var event = createEventStruct(events, fd)
|
||||
if events != {}:
|
||||
if epoll_ctl(s.epollFD, EPOLL_CTL_ADD, fd, addr(event)) != 0:
|
||||
raiseOSError(osLastError())
|
||||
|
||||
var key = SelectorKey(fd: fd, events: events, data: data)
|
||||
|
||||
|
||||
s.fds[fd] = key
|
||||
result = key
|
||||
|
||||
proc update*(s: Selector, fd: SocketHandle,
|
||||
events: set[Event]): SelectorKey {.discardable.} =
|
||||
|
||||
proc update*(s: var Selector, fd: SocketHandle, events: set[Event]) =
|
||||
if s.fds[fd].events != events:
|
||||
if events == {}:
|
||||
# This fd is idle -- it should not be registered to epoll.
|
||||
|
|
@ -108,7 +107,7 @@ elif defined(linux):
|
|||
# are therefore constantly ready. (leading to 100% CPU usage).
|
||||
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fd, nil) != 0:
|
||||
raiseOSError(osLastError())
|
||||
s.fds[fd].events = events
|
||||
s.fds.mget(fd).events = events
|
||||
else:
|
||||
var event = createEventStruct(events, fd)
|
||||
if s.fds[fd].events == {}:
|
||||
|
|
@ -119,22 +118,20 @@ elif defined(linux):
|
|||
else:
|
||||
if epoll_ctl(s.epollFD, EPOLL_CTL_MOD, fd, addr(event)) != 0:
|
||||
raiseOSError(osLastError())
|
||||
s.fds[fd].events = events
|
||||
|
||||
result = s.fds[fd]
|
||||
|
||||
proc unregister*(s: Selector, fd: SocketHandle): SelectorKey {.discardable.} =
|
||||
s.fds.mget(fd).events = events
|
||||
|
||||
proc unregister*(s: var Selector, fd: SocketHandle) =
|
||||
if epoll_ctl(s.epollFD, EPOLL_CTL_DEL, fd, nil) != 0:
|
||||
let err = osLastError()
|
||||
if err.cint notin {ENOENT, EBADF}: # TODO: Why do we sometimes get an EBADF? Is this normal?
|
||||
if err.cint notin {ENOENT, EBADF}:
|
||||
# TODO: Why do we sometimes get an EBADF? Is this normal?
|
||||
raiseOSError(err)
|
||||
result = s.fds[fd]
|
||||
s.fds.del(fd)
|
||||
|
||||
proc close*(s: Selector) =
|
||||
proc close*(s: var Selector) =
|
||||
deinitSharedTable(s.fds)
|
||||
if s.epollFD.close() != 0: raiseOSError(osLastError())
|
||||
dealloc(addr s.events) # TODO: Test this
|
||||
|
||||
|
||||
proc epollHasFd(s: Selector, fd: SocketHandle): bool =
|
||||
result = true
|
||||
var event = createEventStruct(s.fds[fd].events, fd)
|
||||
|
|
@ -142,9 +139,9 @@ elif defined(linux):
|
|||
let err = osLastError()
|
||||
if err.cint in {ENOENT, EBADF}:
|
||||
return false
|
||||
raiseOSError(osLastError())
|
||||
|
||||
proc select*(s: Selector, timeout: int): seq[ReadyInfo] =
|
||||
raiseOSError(err)
|
||||
|
||||
proc select*(s: var Selector, timeout: int): seq[ReadyInfo] =
|
||||
##
|
||||
## The ``events`` field of the returned ``key`` contains the original events
|
||||
## for which the ``fd`` was bound. This is contrary to the ``events`` field
|
||||
|
|
@ -156,11 +153,11 @@ elif defined(linux):
|
|||
let err = osLastError()
|
||||
if err.cint == EINTR:
|
||||
return @[]
|
||||
raiseOSError(osLastError())
|
||||
raiseOSError(err)
|
||||
if evNum == 0: return @[]
|
||||
for i in 0 .. <evNum:
|
||||
let fd = s.events[i].data.fd.SocketHandle
|
||||
|
||||
|
||||
var evSet: set[Event] = {}
|
||||
if (s.events[i].events and EPOLLERR) != 0 or (s.events[i].events and EPOLLHUP) != 0: evSet = evSet + {EvError}
|
||||
if (s.events[i].events and EPOLLIN) != 0: evSet = evSet + {EvRead}
|
||||
|
|
@ -170,14 +167,12 @@ elif defined(linux):
|
|||
result.add((selectorKey, evSet))
|
||||
|
||||
#echo("Epoll: ", result[i].key.fd, " ", result[i].events, " ", result[i].key.events)
|
||||
|
||||
|
||||
proc newSelector*(): Selector =
|
||||
new result
|
||||
result.epollFD = epoll_create(64)
|
||||
#result.events = cast[array[64, epoll_event]](alloc0(sizeof(epoll_event)*64))
|
||||
result.fds = initTable[SocketHandle, SelectorKey]()
|
||||
if result.epollFD < 0:
|
||||
raiseOSError(osLastError())
|
||||
result.fds = initSharedTable[SocketHandle, SelectorKey]()
|
||||
|
||||
proc contains*(s: Selector, fd: SocketHandle): bool =
|
||||
## Determines whether selector contains a file descriptor.
|
||||
|
|
@ -196,31 +191,27 @@ elif defined(linux):
|
|||
|
||||
elif not defined(nimdoc):
|
||||
# TODO: kqueue for bsd/mac os x.
|
||||
import sharedtables
|
||||
|
||||
type
|
||||
Selector* = ref object
|
||||
fds: Table[SocketHandle, SelectorKey]
|
||||
Selector* = object
|
||||
fds: SharedTable[SocketHandle, SelectorKey]
|
||||
|
||||
proc register*(s: Selector, fd: SocketHandle, events: set[Event],
|
||||
data: RootRef): SelectorKey {.discardable.} =
|
||||
if s.fds.hasKey(fd):
|
||||
proc register*(s: var Selector, fd: SocketHandle, events: set[Event],
|
||||
data: pointer) =
|
||||
let result = SelectorKey(fd: fd, events: events, data: data)
|
||||
if s.fds.hasKeyOrPut(fd, result):
|
||||
raise newException(ValueError, "File descriptor already exists.")
|
||||
var sk = SelectorKey(fd: fd, events: events, data: data)
|
||||
s.fds[fd] = sk
|
||||
result = sk
|
||||
|
||||
proc update*(s: Selector, fd: SocketHandle,
|
||||
events: set[Event]): SelectorKey {.discardable.} =
|
||||
if not s.fds.hasKey(fd):
|
||||
raise newException(ValueError, "File descriptor not found.")
|
||||
proc update*(s: var Selector, fd: SocketHandle, events: set[Event]) =
|
||||
#if not s.fds.hasKey(fd):
|
||||
# raise newException(ValueError, "File descriptor not found.")
|
||||
s.fds.mget(fd).events = events
|
||||
|
||||
s.fds[fd].events = events
|
||||
result = s.fds[fd]
|
||||
|
||||
proc unregister*(s: Selector, fd: SocketHandle): SelectorKey {.discardable.} =
|
||||
result = s.fds[fd]
|
||||
proc unregister*(s: var Selector, fd: SocketHandle) =
|
||||
s.fds.del(fd)
|
||||
|
||||
proc close*(s: Selector) = discard
|
||||
proc close*(s: var Selector) = deinitSharedTable(s.fds)
|
||||
|
||||
proc timeValFromMilliseconds(timeout: int): TimeVal =
|
||||
if timeout != -1:
|
||||
|
|
@ -228,19 +219,19 @@ elif not defined(nimdoc):
|
|||
result.tv_sec = seconds.int32
|
||||
result.tv_usec = ((timeout - seconds * 1000) * 1000).int32
|
||||
|
||||
proc createFdSet(rd, wr: var TFdSet, fds: Table[SocketHandle, SelectorKey],
|
||||
proc createFdSet(rd, wr: var TFdSet, fds: SharedTable[SocketHandle, SelectorKey],
|
||||
m: var int) =
|
||||
FD_ZERO(rd); FD_ZERO(wr)
|
||||
for k, v in pairs(fds):
|
||||
if EvRead in v.events:
|
||||
if EvRead in v.events:
|
||||
m = max(m, int(k))
|
||||
FD_SET(k, rd)
|
||||
if EvWrite in v.events:
|
||||
m = max(m, int(k))
|
||||
FD_SET(k, wr)
|
||||
|
||||
proc getReadyFDs(rd, wr: var TFdSet, fds: Table[SocketHandle, SelectorKey]):
|
||||
seq[ReadyInfo] =
|
||||
|
||||
proc getReadyFDs(rd, wr: var TFdSet,
|
||||
fds: SharedTable[SocketHandle, SelectorKey]): seq[ReadyInfo] =
|
||||
result = @[]
|
||||
for k, v in pairs(fds):
|
||||
var events: set[Event] = {}
|
||||
|
|
@ -250,20 +241,20 @@ elif not defined(nimdoc):
|
|||
events = events + {EvWrite}
|
||||
result.add((v, events))
|
||||
|
||||
proc select(fds: Table[SocketHandle, SelectorKey], timeout = 500):
|
||||
seq[ReadyInfo] =
|
||||
proc select(fds: var SharedTable[SocketHandle, SelectorKey],
|
||||
timeout = 500): seq[ReadyInfo] =
|
||||
var tv {.noInit.}: TimeVal = timeValFromMilliseconds(timeout)
|
||||
|
||||
|
||||
var rd, wr: TFdSet
|
||||
var m = 0
|
||||
createFdSet(rd, wr, fds, m)
|
||||
|
||||
|
||||
var retCode = 0
|
||||
if timeout != -1:
|
||||
retCode = int(select(cint(m+1), addr(rd), addr(wr), nil, addr(tv)))
|
||||
else:
|
||||
retCode = int(select(cint(m+1), addr(rd), addr(wr), nil, nil))
|
||||
|
||||
|
||||
if retCode < 0:
|
||||
raiseOSError(osLastError())
|
||||
elif retCode == 0:
|
||||
|
|
@ -275,8 +266,7 @@ elif not defined(nimdoc):
|
|||
result = select(s.fds, timeout)
|
||||
|
||||
proc newSelector*(): Selector =
|
||||
new result
|
||||
result.fds = initTable[SocketHandle, SelectorKey]()
|
||||
result.fds = initSharedTable[SocketHandle, SelectorKey]()
|
||||
|
||||
proc contains*(s: Selector, fd: SocketHandle): bool =
|
||||
return s.fds.hasKey(fd)
|
||||
|
|
@ -300,17 +290,17 @@ when not defined(testing) and isMainModule and not defined(nimdoc):
|
|||
# Select()
|
||||
import sockets
|
||||
type
|
||||
SockWrapper = ref object of RootObj
|
||||
SockWrapper = object
|
||||
sock: Socket
|
||||
|
||||
|
||||
var sock = socket()
|
||||
if sock == sockets.invalidSocket: raiseOSError(osLastError())
|
||||
#sock.setBlocking(false)
|
||||
sock.connect("irc.freenode.net", Port(6667))
|
||||
|
||||
|
||||
var selector = newSelector()
|
||||
var data = SockWrapper(sock: sock)
|
||||
let key = selector.register(sock.getFD, {EvWrite}, data)
|
||||
let key = selector.register(sock.getFD, {EvWrite}, addr data)
|
||||
var i = 0
|
||||
while true:
|
||||
let ready = selector.select(1000)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue