OSS/tanya
1
0
Fork 0
Code Issues 1 Pull Requests Releases Activity

Switch to container.queue. Remove PendingQueue

Browse Source
This commit is contained in:
Eugen Wissner 2016-12-02 19:18:37 +01:00
parent 1123d01e6c
commit f7f92e7906
10 changed files with 1289 additions and 1399 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

268
source/tanya/async/event/epoll.d
View File

@ -28,151 +28,151 @@ import std.algorithm.comparison;
class EpollLoop : SelectorLoop class EpollLoop : SelectorLoop
{ {
protected int fd; protected int fd;
private epoll_event[] events; private epoll_event[] events;
/** /**
* Initializes the loop. * Initializes the loop.
*/ */
this() this()
{ {
if ((fd = epoll_create1(EPOLL_CLOEXEC)) < 0) if ((fd = epoll_create1(EPOLL_CLOEXEC)) < 0)
{ {
throw MmapPool.instance.make!BadLoopException("epoll initialization failed"); throw MmapPool.instance.make!BadLoopException("epoll initialization failed");
} }
super(); super();
events = MmapPool.instance.makeArray!epoll_event(maxEvents); events = MmapPool.instance.makeArray!epoll_event(maxEvents);
} }
/** /**
* Free loop internals. * Free loop internals.
*/ */
~this() ~this()
{ {
MmapPool.instance.dispose(events); MmapPool.instance.dispose(events);
close(fd); close(fd);
} }
/** /**
* Should be called if the backend configuration changes. * Should be called if the backend configuration changes.
* *
* Params: * Params:
* watcher = Watcher. * watcher = Watcher.
* oldEvents = The events were already set. * oldEvents = The events were already set.
* events = The events should be set. * events = The events should be set.
* *
* Returns: $(D_KEYWORD true) if the operation was successful. * Returns: $(D_KEYWORD true) if the operation was successful.
*/ */
protected override bool reify(ConnectionWatcher watcher, EventMask oldEvents, EventMask events) protected override bool reify(ConnectionWatcher watcher, EventMask oldEvents, EventMask events)
in in
{ {
assert(watcher !is null); assert(watcher !is null);
} }
body body
{ {
int op = EPOLL_CTL_DEL; int op = EPOLL_CTL_DEL;
epoll_event ev; epoll_event ev;
if (events == oldEvents) if (events == oldEvents)
{ {
return true; return true;
} }
if (events && oldEvents) if (events && oldEvents)
{ {
op = EPOLL_CTL_MOD; op = EPOLL_CTL_MOD;
} }
else if (events && !oldEvents) else if (events && !oldEvents)
{ {
op = EPOLL_CTL_ADD; op = EPOLL_CTL_ADD;
} }
ev.data.fd = watcher.socket.handle; ev.data.fd = watcher.socket.handle;
ev.events = (events & (Event.read | Event.accept) ? EPOLLIN | EPOLLPRI : 0) ev.events = (events & (Event.read | Event.accept) ? EPOLLIN | EPOLLPRI : 0)
| (events & Event.write ? EPOLLOUT : 0) | (events & Event.write ? EPOLLOUT : 0)
| EPOLLET; | EPOLLET;
return epoll_ctl(fd, op, watcher.socket.handle, &ev) == 0; return epoll_ctl(fd, op, watcher.socket.handle, &ev) == 0;
} }
/** /**
* Does the actual polling. * Does the actual polling.
*/ */
protected override void poll() protected override void poll()
{ {
// Don't block // Don't block
immutable timeout = cast(immutable int) blockTime.total!"msecs"; immutable timeout = cast(immutable int) blockTime.total!"msecs";
auto eventCount = epoll_wait(fd, events.ptr, maxEvents, timeout); auto eventCount = epoll_wait(fd, events.ptr, maxEvents, timeout);
if (eventCount < 0) if (eventCount < 0)
{ {
if (errno != EINTR) if (errno != EINTR)
{ {
throw theAllocator.make!BadLoopException(); throw theAllocator.make!BadLoopException();
} }
return; return;
} }
for (auto i = 0; i < eventCount; ++i) for (auto i = 0; i < eventCount; ++i)
{ {
auto io = cast(IOWatcher) connections[events[i].data.fd]; auto io = cast(IOWatcher) connections[events[i].data.fd];
if (io is null) if (io is null)
{ {
acceptConnections(connections[events[i].data.fd]); acceptConnections(connections[events[i].data.fd]);
} }
else if (events[i].events & EPOLLERR) else if (events[i].events & EPOLLERR)
{ {
kill(io, null); kill(io, null);
} }
else if (events[i].events & (EPOLLIN | EPOLLPRI | EPOLLHUP)) else if (events[i].events & (EPOLLIN | EPOLLPRI | EPOLLHUP))
{ {
auto transport = cast(SelectorStreamTransport) io.transport; auto transport = cast(SelectorStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
SocketException exception; SocketException exception;
try try
{ {
ptrdiff_t received; ptrdiff_t received;
do do
{ {
received = transport.socket.receive(io.output[]); received = transport.socket.receive(io.output[]);
io.output += received; io.output += received;
} }
while (received); while (received);
} }
catch (SocketException e) catch (SocketException e)
{ {
exception = e; exception = e;
} }
if (transport.socket.disconnected) if (transport.socket.disconnected)
{ {
kill(io, exception); kill(io, exception);
} }
else if (io.output.length) else if (io.output.length)
{ {
swapPendings.insertBack(io); swapPendings.insertBack(io);
} }
} }
else if (events[i].events & EPOLLOUT) else if (events[i].events & EPOLLOUT)
{ {
auto transport = cast(SelectorStreamTransport) io.transport; auto transport = cast(SelectorStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
transport.writeReady = true; transport.writeReady = true;
if (transport.input.length) if (transport.input.length)
{ {
feed(transport); feed(transport);
} }
} }
} }
} }
/** /**
* Returns: The blocking time. * Returns: The blocking time.
*/ */
override protected @property inout(Duration) blockTime() override protected @property inout(Duration) blockTime()
inout @safe pure nothrow inout @safe pure nothrow
{ {
return min(super.blockTime, 1.dur!"seconds"); return min(super.blockTime, 1.dur!"seconds");
} }
} }

466
source/tanya/async/event/iocp.d
View File

@ -28,267 +28,267 @@ import core.sys.windows.winsock2;
class IOCPStreamTransport : StreamTransport class IOCPStreamTransport : StreamTransport
{ {
private OverlappedConnectedSocket socket_; private OverlappedConnectedSocket socket_;
private WriteBuffer input; private WriteBuffer input;
/** /**
* Creates new completion port transport. * Creates new completion port transport.
* Params: * Params:
* socket = Socket. * socket = Socket.
*/ */
this(OverlappedConnectedSocket socket) this(OverlappedConnectedSocket socket)
in in
{ {
assert(socket !is null); assert(socket !is null);
} }
body body
{ {
socket_ = socket; socket_ = socket;
input = MmapPool.instance.make!WriteBuffer(); input = MmapPool.instance.make!WriteBuffer();
} }
~this() ~this()
{ {
MmapPool.instance.dispose(input); MmapPool.instance.dispose(input);
} }
@property inout(OverlappedConnectedSocket) socket() inout pure nothrow @safe @nogc @property inout(OverlappedConnectedSocket) socket() inout pure nothrow @safe @nogc
{ {
return socket_; return socket_;
} }
/** /**
* Write some data to the transport. * Write some data to the transport.
* *
* Params: * Params:
* data = Data to send. * data = Data to send.
*/ */
void write(ubyte[] data) void write(ubyte[] data)
{ {
immutable empty = input.length == 0; immutable empty = input.length == 0;
input ~= data; input ~= data;
if (empty) if (empty)
{ {
SocketState overlapped; SocketState overlapped;
try try
{ {
overlapped = MmapPool.instance.make!SocketState; overlapped = MmapPool.instance.make!SocketState;
socket.beginSend(input[], overlapped); socket.beginSend(input[], overlapped);
} }
catch (SocketException e) catch (SocketException e)
{ {
MmapPool.instance.dispose(overlapped); MmapPool.instance.dispose(overlapped);
MmapPool.instance.dispose(e); MmapPool.instance.dispose(e);
} }
} }
} }
} }
class IOCPLoop : Loop class IOCPLoop : Loop
{ {
protected HANDLE completionPort; protected HANDLE completionPort;
protected OVERLAPPED overlap; protected OVERLAPPED overlap;
/** /**
* Initializes the loop. * Initializes the loop.
*/ */
this() this()
{ {
super(); super();
completionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); completionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
if (!completionPort) if (!completionPort)
{ {
throw theAllocator.make!BadLoopException("Creating completion port failed"); throw theAllocator.make!BadLoopException("Creating completion port failed");
} }
} }
/** /**
* Should be called if the backend configuration changes. * Should be called if the backend configuration changes.
* *
* Params: * Params:
* watcher = Watcher. * watcher = Watcher.
* oldEvents = The events were already set. * oldEvents = The events were already set.
* events = The events should be set. * events = The events should be set.
* *
* Returns: $(D_KEYWORD true) if the operation was successful. * Returns: $(D_KEYWORD true) if the operation was successful.
*/ */
override protected bool reify(ConnectionWatcher watcher, override protected bool reify(ConnectionWatcher watcher,
EventMask oldEvents, EventMask oldEvents,
EventMask events) EventMask events)
{ {
SocketState overlapped; SocketState overlapped;
if (!(oldEvents & Event.accept) && (events & Event.accept)) if (!(oldEvents & Event.accept) && (events & Event.accept))
{ {
auto socket = cast(OverlappedStreamSocket) watcher.socket; auto socket = cast(OverlappedStreamSocket) watcher.socket;
assert(socket !is null); assert(socket !is null);
if (CreateIoCompletionPort(cast(HANDLE) socket.handle, if (CreateIoCompletionPort(cast(HANDLE) socket.handle,
completionPort, completionPort,
cast(ULONG_PTR) (cast(void*) watcher), cast(ULONG_PTR) (cast(void*) watcher),
0) !is completionPort) 0) !is completionPort)
{ {
return false; return false;
} }
try try
{ {
overlapped = MmapPool.instance.make!SocketState; overlapped = MmapPool.instance.make!SocketState;
socket.beginAccept(overlapped); socket.beginAccept(overlapped);
} }
catch (SocketException e) catch (SocketException e)
{ {
MmapPool.instance.dispose(overlapped); MmapPool.instance.dispose(overlapped);
theAllocator.dispose(e); theAllocator.dispose(e);
return false; return false;
} }
} }
if (!(oldEvents & Event.read) && (events & Event.read) if (!(oldEvents & Event.read) && (events & Event.read)
|| !(oldEvents & Event.write) && (events & Event.write)) || !(oldEvents & Event.write) && (events & Event.write))
{ {
auto io = cast(IOWatcher) watcher; auto io = cast(IOWatcher) watcher;
assert(io !is null); assert(io !is null);
auto transport = cast(IOCPStreamTransport) io.transport; auto transport = cast(IOCPStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
if (CreateIoCompletionPort(cast(HANDLE) transport.socket.handle, if (CreateIoCompletionPort(cast(HANDLE) transport.socket.handle,
completionPort, completionPort,
cast(ULONG_PTR) (cast(void*) watcher), cast(ULONG_PTR) (cast(void*) watcher),
0) !is completionPort) 0) !is completionPort)
{ {
return false; return false;
} }
// Begin to read // Begin to read
if (!(oldEvents & Event.read) && (events & Event.read)) if (!(oldEvents & Event.read) && (events & Event.read))
{ {
try try
{ {
overlapped = MmapPool.instance.make!SocketState; overlapped = MmapPool.instance.make!SocketState;
transport.socket.beginReceive(io.output[], overlapped); transport.socket.beginReceive(io.output[], overlapped);
} }
catch (SocketException e) catch (SocketException e)
{ {
MmapPool.instance.dispose(overlapped); MmapPool.instance.dispose(overlapped);
theAllocator.dispose(e); theAllocator.dispose(e);
return false; return false;
} }
} }
} }
return true; return true;
} }
/** /**
* Does the actual polling. * Does the actual polling.
*/ */
override protected void poll() override protected void poll()
{ {
DWORD lpNumberOfBytes; DWORD lpNumberOfBytes;
ULONG_PTR key; ULONG_PTR key;
LPOVERLAPPED overlap; LPOVERLAPPED overlap;
immutable timeout = cast(immutable int) blockTime.total!"msecs"; immutable timeout = cast(immutable int) blockTime.total!"msecs";
auto result = GetQueuedCompletionStatus(completionPort, auto result = GetQueuedCompletionStatus(completionPort,
&lpNumberOfBytes, &lpNumberOfBytes,
&key, &key,
&overlap, &overlap,
timeout); timeout);
if (result == FALSE && overlap == NULL) if (result == FALSE && overlap == NULL)
{ {
return; // Timeout return; // Timeout
} }
auto overlapped = (cast(SocketState) ((cast(void*) overlap) - 8)); auto overlapped = (cast(SocketState) ((cast(void*) overlap) - 8));
assert(overlapped !is null); assert(overlapped !is null);
scope (failure) scope (failure)
{ {
MmapPool.instance.dispose(overlapped); MmapPool.instance.dispose(overlapped);
} }
switch (overlapped.event) switch (overlapped.event)
{ {
case OverlappedSocketEvent.accept: case OverlappedSocketEvent.accept:
auto connection = cast(ConnectionWatcher) (cast(void*) key); auto connection = cast(ConnectionWatcher) (cast(void*) key);
assert(connection !is null); assert(connection !is null);
auto listener = cast(OverlappedStreamSocket) connection.socket; auto listener = cast(OverlappedStreamSocket) connection.socket;
assert(listener !is null); assert(listener !is null);
auto socket = listener.endAccept(overlapped); auto socket = listener.endAccept(overlapped);
auto transport = MmapPool.instance.make!IOCPStreamTransport(socket); auto transport = MmapPool.instance.make!IOCPStreamTransport(socket);
auto io = MmapPool.instance.make!IOWatcher(transport, connection.protocol); auto io = MmapPool.instance.make!IOWatcher(transport, connection.protocol);
connection.incoming.insertBack(io); connection.incoming.insertBack(io);
reify(io, EventMask(Event.none), EventMask(Event.read, Event.write)); reify(io, EventMask(Event.none), EventMask(Event.read, Event.write));
swapPendings.insertBack(connection); swapPendings.insertBack(connection);
listener.beginAccept(overlapped); listener.beginAccept(overlapped);
break; break;
case OverlappedSocketEvent.read: case OverlappedSocketEvent.read:
auto io = cast(IOWatcher) (cast(void*) key); auto io = cast(IOWatcher) (cast(void*) key);
assert(io !is null); assert(io !is null);
if (!io.active) if (!io.active)
{ {
MmapPool.instance.dispose(io); MmapPool.instance.dispose(io);
MmapPool.instance.dispose(overlapped); MmapPool.instance.dispose(overlapped);
return; return;
} }
auto transport = cast(IOCPStreamTransport) io.transport; auto transport = cast(IOCPStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
int received; int received;
SocketException exception; SocketException exception;
try try
{ {
received = transport.socket.endReceive(overlapped); received = transport.socket.endReceive(overlapped);
} }
catch (SocketException e) catch (SocketException e)
{ {
exception = e; exception = e;
} }
if (transport.socket.disconnected) if (transport.socket.disconnected)
{ {
// We want to get one last notification to destroy the watcher // We want to get one last notification to destroy the watcher
transport.socket.beginReceive(io.output[], overlapped); transport.socket.beginReceive(io.output[], overlapped);
kill(io, exception); kill(io, exception);
} }
else if (received > 0) else if (received > 0)
{ {
immutable full = io.output.free == received; immutable full = io.output.free == received;
io.output += received; io.output += received;
// Receive was interrupted because the buffer is full. We have to continue // Receive was interrupted because the buffer is full. We have to continue
if (full) if (full)
{ {
transport.socket.beginReceive(io.output[], overlapped); transport.socket.beginReceive(io.output[], overlapped);
} }
swapPendings.insertBack(io); swapPendings.insertBack(io);
} }
break; break;
case OverlappedSocketEvent.write: case OverlappedSocketEvent.write:
auto io = cast(IOWatcher) (cast(void*) key); auto io = cast(IOWatcher) (cast(void*) key);
assert(io !is null); assert(io !is null);
auto transport = cast(IOCPStreamTransport) io.transport; auto transport = cast(IOCPStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
transport.input += transport.socket.endSend(overlapped); transport.input += transport.socket.endSend(overlapped);
if (transport.input.length) if (transport.input.length)
{ {
transport.socket.beginSend(transport.input[], overlapped); transport.socket.beginSend(transport.input[], overlapped);
} }
else else
{ {
transport.socket.beginReceive(io.output[], overlapped); transport.socket.beginReceive(io.output[], overlapped);
} }
break; break;
default: default:
assert(false, "Unknown event"); assert(false, "Unknown event");
} }
} }
} }

510
source/tanya/async/event/kqueue.d
View File

@ -12,117 +12,117 @@ module tanya.async.event.kqueue;
version (OSX) version (OSX)
{ {
version = MissingKevent; version = MissingKevent;
} }
else version (iOS) else version (iOS)
{ {
version = MissingKevent; version = MissingKevent;
} }
else version (TVOS) else version (TVOS)
{ {
version = MissingKevent; version = MissingKevent;
} }
else version (WatchOS) else version (WatchOS)
{ {
version = MissingKevent; version = MissingKevent;
} }
else version (OpenBSD) else version (OpenBSD)
{ {
version = MissingKevent; version = MissingKevent;
} }
else version (DragonFlyBSD) else version (DragonFlyBSD)
{ {
version = MissingKevent; version = MissingKevent;
} }
version (MissingKevent) version (MissingKevent)
{ {
extern (C): extern (C):
nothrow: nothrow:
@nogc: @nogc:
import core.stdc.stdint; // intptr_t, uintptr_t import core.stdc.stdint; // intptr_t, uintptr_t
import core.sys.posix.time; // timespec import core.sys.posix.time; // timespec
enum : short enum : short
{ {
EVFILT_READ = -1, EVFILT_READ = -1,
EVFILT_WRITE = -2, EVFILT_WRITE = -2,
EVFILT_AIO = -3, /* attached to aio requests */ EVFILT_AIO = -3, /* attached to aio requests */
EVFILT_VNODE = -4, /* attached to vnodes */ EVFILT_VNODE = -4, /* attached to vnodes */
EVFILT_PROC = -5, /* attached to struct proc */ EVFILT_PROC = -5, /* attached to struct proc */
EVFILT_SIGNAL = -6, /* attached to struct proc */ EVFILT_SIGNAL = -6, /* attached to struct proc */
EVFILT_TIMER = -7, /* timers */ EVFILT_TIMER = -7, /* timers */
EVFILT_MACHPORT = -8, /* Mach portsets */ EVFILT_MACHPORT = -8, /* Mach portsets */
EVFILT_FS = -9, /* filesystem events */ EVFILT_FS = -9, /* filesystem events */
EVFILT_USER = -10, /* User events */ EVFILT_USER = -10, /* User events */
EVFILT_VM = -12, /* virtual memory events */ EVFILT_VM = -12, /* virtual memory events */
EVFILT_SYSCOUNT = 11 EVFILT_SYSCOUNT = 11
} }
extern(D) void EV_SET(kevent_t* kevp, typeof(kevent_t.tupleof) args) extern(D) void EV_SET(kevent_t* kevp, typeof(kevent_t.tupleof) args)
{ {
*kevp = kevent_t(args); *kevp = kevent_t(args);
} }
struct kevent_t struct kevent_t
{ {
uintptr_t ident; /* identifier for this event */ uintptr_t ident; /* identifier for this event */
short filter; /* filter for event */ short filter; /* filter for event */
ushort flags; ushort flags;
uint fflags; uint fflags;
intptr_t data; intptr_t data;
void *udata; /* opaque user data identifier */ void *udata; /* opaque user data identifier */
} }
enum enum
{ {
/* actions */ /* actions */
EV_ADD = 0x0001, /* add event to kq (implies enable) */ EV_ADD = 0x0001, /* add event to kq (implies enable) */
EV_DELETE = 0x0002, /* delete event from kq */ EV_DELETE = 0x0002, /* delete event from kq */
EV_ENABLE = 0x0004, /* enable event */ EV_ENABLE = 0x0004, /* enable event */
EV_DISABLE = 0x0008, /* disable event (not reported) */ EV_DISABLE = 0x0008, /* disable event (not reported) */
/* flags */ /* flags */
EV_ONESHOT = 0x0010, /* only report one occurrence */ EV_ONESHOT = 0x0010, /* only report one occurrence */
EV_CLEAR = 0x0020, /* clear event state after reporting */ EV_CLEAR = 0x0020, /* clear event state after reporting */
EV_RECEIPT = 0x0040, /* force EV_ERROR on success, data=0 */ EV_RECEIPT = 0x0040, /* force EV_ERROR on success, data=0 */
EV_DISPATCH = 0x0080, /* disable event after reporting */ EV_DISPATCH = 0x0080, /* disable event after reporting */
EV_SYSFLAGS = 0xF000, /* reserved by system */ EV_SYSFLAGS = 0xF000, /* reserved by system */
EV_FLAG1 = 0x2000, /* filter-specific flag */ EV_FLAG1 = 0x2000, /* filter-specific flag */
/* returned values */ /* returned values */
EV_EOF = 0x8000, /* EOF detected */ EV_EOF = 0x8000, /* EOF detected */
EV_ERROR = 0x4000, /* error, data contains errno */ EV_ERROR = 0x4000, /* error, data contains errno */
} }
int kqueue(); int kqueue();
int kevent(int kq, const kevent_t *changelist, int nchanges, int kevent(int kq, const kevent_t *changelist, int nchanges,
kevent_t *eventlist, int nevents, kevent_t *eventlist, int nevents,
const timespec *timeout); const timespec *timeout);
} }
version (OSX) version (OSX)
{ {
version = MacBSD; version = MacBSD;
} }
else version (iOS) else version (iOS)
{ {
version = MacBSD; version = MacBSD;
} }
else version (FreeBSD) else version (FreeBSD)
{ {
version = MacBSD; version = MacBSD;
public import core.sys.freebsd.sys.event; public import core.sys.freebsd.sys.event;
} }
else version (OpenBSD) else version (OpenBSD)
{ {
version = MacBSD; version = MacBSD;
} }
else version (DragonFlyBSD) else version (DragonFlyBSD)
{ {
version = MacBSD; version = MacBSD;
} }
version (MacBSD): version (MacBSD):
@ -142,208 +142,208 @@ import std.algorithm.comparison;
class KqueueLoop : SelectorLoop class KqueueLoop : SelectorLoop
{ {
protected int fd; protected int fd;
private kevent_t[] events; private kevent_t[] events;
private kevent_t[] changes; private kevent_t[] changes;
private size_t changeCount; private size_t changeCount;
/** /**
* Returns: Maximal event count can be got at a time * Returns: Maximal event count can be got at a time
* (should be supported by the backend). * (should be supported by the backend).
*/ */
override protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc override protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc
{ {
return cast(uint) events.length; return cast(uint) events.length;
} }
this() this()
{ {
super(); super();
if ((fd = kqueue()) == -1) if ((fd = kqueue()) == -1)
{ {
throw MmapPool.instance.make!BadLoopException("epoll initialization failed"); throw MmapPool.instance.make!BadLoopException("epoll initialization failed");
} }
events = MmapPool.instance.makeArray!kevent_t(64); events = MmapPool.instance.makeArray!kevent_t(64);
changes = MmapPool.instance.makeArray!kevent_t(64); changes = MmapPool.instance.makeArray!kevent_t(64);
} }
/** /**
* Free loop internals. * Free loop internals.
*/ */
~this() ~this()
{ {
MmapPool.instance.dispose(events); MmapPool.instance.dispose(events);
MmapPool.instance.dispose(changes); MmapPool.instance.dispose(changes);
close(fd); close(fd);
} }
private void set(socket_t socket, short filter, ushort flags) private void set(socket_t socket, short filter, ushort flags)
{ {
if (changes.length <= changeCount) if (changes.length <= changeCount)
{ {
MmapPool.instance.resizeArray(changes, changeCount + maxEvents); MmapPool.instance.resizeArray(changes, changeCount + maxEvents);
} }
EV_SET(&changes[changeCount], EV_SET(&changes[changeCount],
cast(ulong) socket, cast(ulong) socket,
filter, filter,
flags, flags,
0U, 0U,
0L, 0L,
null); null);
++changeCount; ++changeCount;
} }
/** /**
* Should be called if the backend configuration changes. * Should be called if the backend configuration changes.
* *
* Params: * Params:
* watcher = Watcher. * watcher = Watcher.
* oldEvents = The events were already set. * oldEvents = The events were already set.
* events = The events should be set. * events = The events should be set.
* *
* Returns: $(D_KEYWORD true) if the operation was successful. * Returns: $(D_KEYWORD true) if the operation was successful.
*/ */
override protected bool reify(ConnectionWatcher watcher, override protected bool reify(ConnectionWatcher watcher,
EventMask oldEvents, EventMask oldEvents,
EventMask events) EventMask events)
{ {
if (events != oldEvents) if (events != oldEvents)
{ {
if (oldEvents & Event.read || oldEvents & Event.accept) if (oldEvents & Event.read || oldEvents & Event.accept)
{ {
set(watcher.socket.handle, EVFILT_READ, EV_DELETE); set(watcher.socket.handle, EVFILT_READ, EV_DELETE);
} }
if (oldEvents & Event.write) if (oldEvents & Event.write)
{ {
set(watcher.socket.handle, EVFILT_WRITE, EV_DELETE); set(watcher.socket.handle, EVFILT_WRITE, EV_DELETE);
} }
} }
if (events & (Event.read | events & Event.accept)) if (events & (Event.read | events & Event.accept))
{ {
set(watcher.socket.handle, EVFILT_READ, EV_ADD | EV_ENABLE); set(watcher.socket.handle, EVFILT_READ, EV_ADD | EV_ENABLE);
} }
if (events & Event.write) if (events & Event.write)
{ {
set(watcher.socket.handle, EVFILT_WRITE, EV_ADD | EV_DISPATCH); set(watcher.socket.handle, EVFILT_WRITE, EV_ADD | EV_DISPATCH);
} }
return true; return true;
} }
/** /**
* Does the actual polling. * Does the actual polling.
*/ */
protected override void poll() protected override void poll()
{ {
timespec ts; timespec ts;
blockTime.split!("seconds", "nsecs")(ts.tv_sec, ts.tv_nsec); blockTime.split!("seconds", "nsecs")(ts.tv_sec, ts.tv_nsec);
if (changeCount > maxEvents) if (changeCount > maxEvents)
{ {
MmapPool.instance.resizeArray(events, changes.length); MmapPool.instance.resizeArray(events, changes.length);
} }
auto eventCount = kevent(fd, changes.ptr, cast(int) changeCount, events.ptr, maxEvents, &ts); auto eventCount = kevent(fd, changes.ptr, cast(int) changeCount, events.ptr, maxEvents, &ts);
changeCount = 0; changeCount = 0;
if (eventCount < 0) if (eventCount < 0)
{ {
if (errno != EINTR) if (errno != EINTR)
{ {
throw theAllocator.make!BadLoopException(); throw theAllocator.make!BadLoopException();
} }
return; return;
} }
for (int i; i < eventCount; ++i) for (int i; i < eventCount; ++i)
{ {
assert(connections.length > events[i].ident); assert(connections.length > events[i].ident);
IOWatcher io = cast(IOWatcher) connections[events[i].ident]; IOWatcher io = cast(IOWatcher) connections[events[i].ident];
// If it is a ConnectionWatcher. Accept connections. // If it is a ConnectionWatcher. Accept connections.
if (io is null) if (io is null)
{ {
acceptConnections(connections[events[i].ident]); acceptConnections(connections[events[i].ident]);
} }
else if (events[i].flags & EV_ERROR) else if (events[i].flags & EV_ERROR)
{ {
kill(io, null); kill(io, null);
} }
else if (events[i].filter == EVFILT_READ) else if (events[i].filter == EVFILT_READ)
{ {
auto transport = cast(SelectorStreamTransport) io.transport; auto transport = cast(SelectorStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
SocketException exception; SocketException exception;
try try
{ {
ptrdiff_t received; ptrdiff_t received;
do do
{ {
received = transport.socket.receive(io.output[]); received = transport.socket.receive(io.output[]);
io.output += received; io.output += received;
} }
while (received); while (received);
} }
catch (SocketException e) catch (SocketException e)
{ {
exception = e; exception = e;
} }
if (transport.socket.disconnected) if (transport.socket.disconnected)
{ {
kill(io, exception); kill(io, exception);
} }
else if (io.output.length) else if (io.output.length)
{ {
swapPendings.insertBack(io); swapPendings.insertBack(io);
} }
} }
else if (events[i].filter == EVFILT_WRITE) else if (events[i].filter == EVFILT_WRITE)
{ {
auto transport = cast(SelectorStreamTransport) io.transport; auto transport = cast(SelectorStreamTransport) io.transport;
assert(transport !is null); assert(transport !is null);
transport.writeReady = true; transport.writeReady = true;
if (transport.input.length) if (transport.input.length)
{ {
feed(transport); feed(transport);
} }
} }
} }
} }
/** /**
* Returns: The blocking time. * Returns: The blocking time.
*/ */
override protected @property inout(Duration) blockTime() override protected @property inout(Duration) blockTime()
inout @safe pure nothrow inout @safe pure nothrow
{ {
return min(super.blockTime, 1.dur!"seconds"); return min(super.blockTime, 1.dur!"seconds");
} }
/** /**
* If the transport couldn't send the data, the further sending should * If the transport couldn't send the data, the further sending should
* be handled by the event loop. * be handled by the event loop.
* *
* Params: * Params:
* transport = Transport. * transport = Transport.
* exception = Exception thrown on sending. * exception = Exception thrown on sending.
* *
* Returns: $(D_KEYWORD true) if the operation could be successfully * Returns: $(D_KEYWORD true) if the operation could be successfully
* completed or scheduled, $(D_KEYWORD false) otherwise (the * completed or scheduled, $(D_KEYWORD false) otherwise (the
* transport is be destroyed then). * transport is be destroyed then).
*/ */
protected override bool feed(SelectorStreamTransport transport, SocketException exception = null) protected override bool feed(SelectorStreamTransport transport, SocketException exception = null)
{ {
if (!super.feed(transport, exception)) if (!super.feed(transport, exception))
{ {
return false; return false;
} }
if (!transport.writeReady) if (!transport.writeReady)
{ {
set(transport.socket.handle, EVFILT_WRITE, EV_DISPATCH); set(transport.socket.handle, EVFILT_WRITE, EV_DISPATCH);
return true; return true;
} }
return false; return false;
} }
} }

426
source/tanya/async/event/selector.d
View File

@ -27,240 +27,240 @@ import core.stdc.errno;
*/ */
class SelectorStreamTransport : StreamTransport class SelectorStreamTransport : StreamTransport
{ {
private ConnectedSocket socket_; private ConnectedSocket socket_;
/// Input buffer. /// Input buffer.
package WriteBuffer input; package WriteBuffer input;
private SelectorLoop loop; private SelectorLoop loop;
/// Received notification that the underlying socket is write-ready. /// Received notification that the underlying socket is write-ready.
package bool writeReady; package bool writeReady;
/** /**
* Params: * Params:
* loop = Event loop. * loop = Event loop.
* socket = Socket. * socket = Socket.
*/ */
this(SelectorLoop loop, ConnectedSocket socket) this(SelectorLoop loop, ConnectedSocket socket)
{ {
socket_ = socket; socket_ = socket;
this.loop = loop; this.loop = loop;
input = MmapPool.instance.make!WriteBuffer(); input = MmapPool.instance.make!WriteBuffer();
} }
/** /**
* Close the transport and deallocate the data buffers. * Close the transport and deallocate the data buffers.
*/ */
~this() ~this()
{ {
MmapPool.instance.dispose(input); MmapPool.instance.dispose(input);
} }
/** /**
* Returns: Transport socket. * Returns: Transport socket.
*/ */
inout(ConnectedSocket) socket() inout pure nothrow @safe @nogc inout(ConnectedSocket) socket() inout pure nothrow @safe @nogc
{ {
return socket_; return socket_;
} }
/** /**
* Write some data to the transport. * Write some data to the transport.
* *
* Params: * Params:
* data = Data to send. * data = Data to send.
*/ */
void write(ubyte[] data) void write(ubyte[] data)
{ {
if (!data.length) if (!data.length)
{ {
return; return;
} }
// Try to write if the socket is write ready. // Try to write if the socket is write ready.
if (writeReady) if (writeReady)
{ {
ptrdiff_t sent; ptrdiff_t sent;
SocketException exception; SocketException exception;
try try
{ {
sent = socket.send(data); sent = socket.send(data);
if (sent == 0) if (sent == 0)
{ {
writeReady = false; writeReady = false;
} }
} }
catch (SocketException e) catch (SocketException e)
{ {
writeReady = false; writeReady = false;
exception = e; exception = e;
} }
if (sent < data.length) if (sent < data.length)
{ {
input ~= data[sent..$]; input ~= data[sent..$];
loop.feed(this, exception); loop.feed(this, exception);
} }
} }
else else
{ {
input ~= data; input ~= data;
} }
} }
} }
abstract class SelectorLoop : Loop abstract class SelectorLoop : Loop
{ {
/// Pending connections. /// Pending connections.
protected ConnectionWatcher[] connections; protected ConnectionWatcher[] connections;
this() this()
{ {
super(); super();
connections = MmapPool.instance.makeArray!ConnectionWatcher(maxEvents); connections = MmapPool.instance.makeArray!ConnectionWatcher(maxEvents);
} }
~this() ~this()
{ {
foreach (ref connection; connections) foreach (ref connection; connections)
{ {
// We want to free only IOWatchers. ConnectionWatcher are created by the // We want to free only IOWatchers. ConnectionWatcher are created by the
// user and should be freed by himself. // user and should be freed by himself.
auto io = cast(IOWatcher) connection; auto io = cast(IOWatcher) connection;
if (io !is null) if (io !is null)
{ {
MmapPool.instance.dispose(io); MmapPool.instance.dispose(io);
connection = null; connection = null;
} }
} }
MmapPool.instance.dispose(connections); MmapPool.instance.dispose(connections);
} }
/** /**
* If the transport couldn't send the data, the further sending should * If the transport couldn't send the data, the further sending should
* be handled by the event loop. * be handled by the event loop.
* *
* Params: * Params:
* transport = Transport. * transport = Transport.
* exception = Exception thrown on sending. * exception = Exception thrown on sending.
* *
* Returns: $(D_KEYWORD true) if the operation could be successfully * Returns: $(D_KEYWORD true) if the operation could be successfully
* completed or scheduled, $(D_KEYWORD false) otherwise (the * completed or scheduled, $(D_KEYWORD false) otherwise (the
* transport will be destroyed then). * transport will be destroyed then).
*/ */
protected bool feed(SelectorStreamTransport transport, SocketException exception = null) protected bool feed(SelectorStreamTransport transport, SocketException exception = null)
{ {
while (transport.input.length && transport.writeReady) while (transport.input.length && transport.writeReady)
{ {
try try
{ {
ptrdiff_t sent = transport.socket.send(transport.input[]); ptrdiff_t sent = transport.socket.send(transport.input[]);
if (sent == 0) if (sent == 0)
{ {
transport.writeReady = false; transport.writeReady = false;
} }
else else
{ {
transport.input += sent; transport.input += sent;
} }
} }
catch (SocketException e) catch (SocketException e)
{ {
exception = e; exception = e;
transport.writeReady = false; transport.writeReady = false;
} }
} }
if (exception !is null) if (exception !is null)
{ {
auto watcher = cast(IOWatcher) connections[transport.socket.handle]; auto watcher = cast(IOWatcher) connections[transport.socket.handle];
assert(watcher !is null); assert(watcher !is null);
kill(watcher, exception); kill(watcher, exception);
return false; return false;
} }
return true; return true;
} }
/** /**
* Start watching. * Start watching.
* *
* Params: * Params:
* watcher = Watcher. * watcher = Watcher.
*/ */
override void start(ConnectionWatcher watcher) override void start(ConnectionWatcher watcher)
{ {
if (watcher.active) if (watcher.active)
{ {
return; return;
} }
if (connections.length <= watcher.socket) if (connections.length <= watcher.socket)
{ {
MmapPool.instance.resizeArray(connections, watcher.socket.handle + maxEvents / 2); MmapPool.instance.resizeArray(connections, watcher.socket.handle + maxEvents / 2);
} }
connections[watcher.socket.handle] = watcher; connections[watcher.socket.handle] = watcher;
super.start(watcher); super.start(watcher);
} }
/** /**
* Accept incoming connections. * Accept incoming connections.
* *
* Params: * Params:
* connection = Connection watcher ready to accept. * connection = Connection watcher ready to accept.
*/ */
package void acceptConnections(ConnectionWatcher connection) package void acceptConnections(ConnectionWatcher connection)
in in
{ {
assert(connection !is null); assert(connection !is null);
} }
body body
{ {
while (true) while (true)
{ {
ConnectedSocket client; ConnectedSocket client;
try try
{ {
client = (cast(StreamSocket) connection.socket).accept(); client = (cast(StreamSocket) connection.socket).accept();
} }
catch (SocketException e) catch (SocketException e)
{ {
theAllocator.dispose(e); theAllocator.dispose(e);
break; break;
} }
if (client is null) if (client is null)
{ {
break; break;
} }
IOWatcher io; IOWatcher io;
auto transport = MmapPool.instance.make!SelectorStreamTransport(this, client); auto transport = MmapPool.instance.make!SelectorStreamTransport(this, client);
if (connections.length > client.handle) if (connections.length > client.handle)
{ {
io = cast(IOWatcher) connections[client.handle]; io = cast(IOWatcher) connections[client.handle];
} }
else else
{ {
MmapPool.instance.resizeArray(connections, client.handle + maxEvents / 2); MmapPool.instance.resizeArray(connections, client.handle + maxEvents / 2);
} }
if (io is null) if (io is null)
{ {
io = MmapPool.instance.make!IOWatcher(transport, io = MmapPool.instance.make!IOWatcher(transport,
connection.protocol); connection.protocol);
connections[client.handle] = io; connections[client.handle] = io;
} }
else else
{ {
io(transport, connection.protocol); io(transport, connection.protocol);
} }
reify(io, EventMask(Event.none), EventMask(Event.read, Event.write)); reify(io, EventMask(Event.none), EventMask(Event.read, Event.write));
connection.incoming.insertBack(io); connection.incoming.insertBack(io);
} }
if (!connection.incoming.empty) if (!connection.incoming.empty)
{ {
swapPendings.insertBack(connection); swapPendings.insertBack(connection);
} }
} }
} }

12
source/tanya/async/iocp.d
View File

@ -21,12 +21,12 @@ import core.sys.windows.windef;
*/ */
class State class State
{ {
/// For internal use by Windows API. /// For internal use by Windows API.
align(1) OVERLAPPED overlapped; align(1) OVERLAPPED overlapped;
/// File/socket handle. /// File/socket handle.
HANDLE handle; HANDLE handle;
/// For keeping events or event masks. /// For keeping events or event masks.
int event; int event;
} }

582
source/tanya/async/loop.d
View File

@ -10,100 +10,105 @@
* *
* --- * ---
* import tanya.async; * import tanya.async;
* import tanya.memory;
* import tanya.network.socket; * import tanya.network.socket;
* *
* class EchoProtocol : TransmissionControlProtocol * class EchoProtocol : TransmissionControlProtocol
* { * {
* private DuplexTransport transport; * private DuplexTransport transport;
* *
* void received(ubyte[] data) * void received(ubyte[] data)
* { * {
* transport.write(data); * transport.write(data);
* } * }
* *
* void connected(DuplexTransport transport) * void connected(DuplexTransport transport)
* { * {
* this.transport = transport; * this.transport = transport;
* } * }
* *
* void disconnected(SocketException e = null) * void disconnected(SocketException e = null)
* { * {
* } * }
* } * }
* *
* void main() * void main()
* { * {
* auto address = new InternetAddress("127.0.0.1", cast(ushort) 8192); * auto address = theAllocator.make!InternetAddress("127.0.0.1", cast(ushort) 8192);
* *
* version (Windows) * version (Windows)
* { * {
* auto sock = new OverlappedStreamSocket(AddressFamily.INET); * auto sock = theAllocator.make!OverlappedStreamSocket(AddressFamily.INET);
* } * }
* else * else
* { * {
* auto sock = new StreamSocket(AddressFamily.INET); * auto sock = theAllocator.make!StreamSocket(AddressFamily.INET);
* sock.blocking = false; * sock.blocking = false;
* } * }
* *
* sock.bind(address); * sock.bind(address);
* sock.listen(5); * sock.listen(5);
* *
* auto io = new ConnectionWatcher(sock); * auto io = theAllocator.make!ConnectionWatcher(sock);
* io.setProtocol!EchoProtocol; * io.setProtocol!EchoProtocol;
* *
* defaultLoop.start(io); * defaultLoop.start(io);
* defaultLoop.run(); * defaultLoop.run();
* *
* sock.shutdown(); * sock.shutdown();
* theAllocator.dispose(io);
* theAllocator.dispose(sock);
* theAllocator.dispose(address);
* } * }
* --- * ---
*/ */
module tanya.async.loop; module tanya.async.loop;
import tanya.async.protocol;
import tanya.async.transport;
import tanya.async.watcher;
import tanya.container.buffer;
import tanya.memory;
import tanya.memory.mmappool;
import tanya.network.socket;
import core.time; import core.time;
import std.algorithm.iteration; import std.algorithm.iteration;
import std.algorithm.mutation; import std.algorithm.mutation;
import std.typecons; import std.typecons;
import tanya.async.protocol;
import tanya.async.transport;
import tanya.async.watcher;
import tanya.container.buffer;
import tanya.container.queue;
import tanya.memory;
import tanya.memory.mmappool;
import tanya.network.socket;
version (DisableBackends) version (DisableBackends)
{ {
} }
else version (linux) else version (linux)
{ {
import tanya.async.event.epoll; import tanya.async.event.epoll;
version = Epoll; version = Epoll;
} }
else version (Windows) else version (Windows)
{ {
import tanya.async.event.iocp; import tanya.async.event.iocp;
version = IOCP; version = IOCP;
} }
else version (OSX) else version (OSX)
{ {
version = Kqueue; version = Kqueue;
} }
else version (iOS) else version (iOS)
{ {
version = Kqueue; version = Kqueue;
} }
else version (FreeBSD) else version (FreeBSD)
{ {
version = Kqueue; version = Kqueue;
} }
else version (OpenBSD) else version (OpenBSD)
{ {
version = Kqueue; version = Kqueue;
} }
else version (DragonFlyBSD) else version (DragonFlyBSD)
{ {
version = Kqueue; version = Kqueue;
} }
/** /**
@ -111,11 +116,11 @@ else version (DragonFlyBSD)
*/ */
enum Event : uint enum Event : uint
{ {
none = 0x00, /// No events. none = 0x00, /// No events.
read = 0x01, /// Non-blocking read call. read = 0x01, /// Non-blocking read call.
write = 0x02, /// Non-blocking write call. write = 0x02, /// Non-blocking write call.
accept = 0x04, /// Connection made. accept = 0x04, /// Connection made.
error = 0x80000000, /// Sent when an error occurs. error = 0x80000000, /// Sent when an error occurs.
} }
alias EventMask = BitFlags!Event; alias EventMask = BitFlags!Event;
@ -125,161 +130,170 @@ alias EventMask = BitFlags!Event;
*/ */
abstract class Loop abstract class Loop
{ {
/// Pending watchers. /// Pending watchers.
protected PendingQueue!Watcher pendings; protected Queue!Watcher pendings;
protected PendingQueue!Watcher swapPendings; protected Queue!Watcher swapPendings;
/** /**
* Returns: Maximal event count can be got at a time * Returns: Maximal event count can be got at a time
* (should be supported by the backend). * (should be supported by the backend).
*/ */
protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc
{ {
return 128U; return 128U;
} }
/** /**
* Initializes the loop. * Initializes the loop.
*/ */
this() this()
{ {
pendings = MmapPool.instance.make!(PendingQueue!Watcher); pendings = MmapPool.instance.make!(Queue!Watcher);
swapPendings = MmapPool.instance.make!(PendingQueue!Watcher); swapPendings = MmapPool.instance.make!(Queue!Watcher);
} }
/** /**
* Frees loop internals. * Frees loop internals.
*/ */
~this() ~this()
{ {
MmapPool.instance.dispose(pendings); foreach (w; pendings)
MmapPool.instance.dispose(swapPendings); {
} MmapPool.instance.dispose(w);
}
MmapPool.instance.dispose(pendings);
/** foreach (w; swapPendings)
* Starts the loop. {
*/ MmapPool.instance.dispose(w);
void run() }
{ MmapPool.instance.dispose(swapPendings);
done_ = false; }
do
{
poll();
// Invoke pendings /**
swapPendings.each!((ref p) => p.invoke()); * Starts the loop.
*/
void run()
{
done_ = false;
do
{
poll();
swap(pendings, swapPendings); // Invoke pendings
} swapPendings.each!((ref p) => p.invoke());
while (!done_);
}
/** swap(pendings, swapPendings);
* Break out of the loop. }
*/ while (!done_);
void unloop() @safe pure nothrow }
{
done_ = true;
}
/** /**
* Start watching. * Break out of the loop.
* */
* Params: void unloop() @safe pure nothrow
* watcher = Watcher. {
*/ done_ = true;
void start(ConnectionWatcher watcher) }
{
if (watcher.active)
{
return;
}
watcher.active = true;
reify(watcher, EventMask(Event.none), EventMask(Event.accept));
}
/** /**
* Stop watching. * Start watching.
* *
* Params: * Params:
* watcher = Watcher. * watcher = Watcher.
*/ */
void stop(ConnectionWatcher watcher) void start(ConnectionWatcher watcher)
{ {
if (!watcher.active) if (watcher.active)
{ {
return; return;
} }
watcher.active = false; watcher.active = true;
reify(watcher, EventMask(Event.none), EventMask(Event.accept));
}
reify(watcher, EventMask(Event.accept), EventMask(Event.none)); /**
} * Stop watching.
*
* Params:
* watcher = Watcher.
*/
void stop(ConnectionWatcher watcher)
{
if (!watcher.active)
{
return;
}
watcher.active = false;
/** reify(watcher, EventMask(Event.accept), EventMask(Event.none));
* Should be called if the backend configuration changes. }
*
* Params:
* watcher = Watcher.
* oldEvents = The events were already set.
* events = The events should be set.
*
* Returns: $(D_KEYWORD true) if the operation was successful.
*/
abstract protected bool reify(ConnectionWatcher watcher,
EventMask oldEvents,
EventMask events);
/** /**
* Returns: The blocking time. * Should be called if the backend configuration changes.
*/ *
protected @property inout(Duration) blockTime() * Params:
inout @safe pure nothrow * watcher = Watcher.
{ * oldEvents = The events were already set.
// Don't block if we have to do. * events = The events should be set.
return swapPendings.empty ? blockTime_ : Duration.zero; *
} * Returns: $(D_KEYWORD true) if the operation was successful.
*/
abstract protected bool reify(ConnectionWatcher watcher,
EventMask oldEvents,
EventMask events);
/** /**
* Sets the blocking time for IO watchers. * Returns: The blocking time.
* */
* Params: protected @property inout(Duration) blockTime()
* blockTime = The blocking time. Cannot be larger than inout @safe pure nothrow
* $(D_PSYMBOL maxBlockTime). {
*/ // Don't block if we have to do.
protected @property void blockTime(in Duration blockTime) @safe pure nothrow return swapPendings.empty ? blockTime_ : Duration.zero;
in }
{
assert(blockTime <= 1.dur!"hours", "Too long to wait.");
assert(!blockTime.isNegative);
}
body
{
blockTime_ = blockTime;
}
/** /**
* Kills the watcher and closes the connection. * Sets the blocking time for IO watchers.
*/ *
protected void kill(IOWatcher watcher, SocketException exception) * Params:
{ * blockTime = The blocking time. Cannot be larger than
watcher.socket.shutdown(); * $(D_PSYMBOL maxBlockTime).
theAllocator.dispose(watcher.socket); */
MmapPool.instance.dispose(watcher.transport); protected @property void blockTime(in Duration blockTime) @safe pure nothrow
watcher.exception = exception; in
swapPendings.insertBack(watcher); {
} assert(blockTime <= 1.dur!"hours", "Too long to wait.");
assert(!blockTime.isNegative);
}
body
{
blockTime_ = blockTime;
}
/** /**
* Does the actual polling. * Kills the watcher and closes the connection.
*/ */
abstract protected void poll(); protected void kill(IOWatcher watcher, SocketException exception)
{
watcher.socket.shutdown();
theAllocator.dispose(watcher.socket);
MmapPool.instance.dispose(watcher.transport);
watcher.exception = exception;
swapPendings.insertBack(watcher);
}
/// Whether the event loop should be stopped. /**
private bool done_; * Does the actual polling.
*/
abstract protected void poll();
/// Maximal block time. /// Whether the event loop should be stopped.
protected Duration blockTime_ = 1.dur!"minutes"; private bool done_;
/// Maximal block time.
protected Duration blockTime_ = 1.dur!"minutes";
} }
/** /**
@ -287,18 +301,17 @@ abstract class Loop
*/ */
class BadLoopException : Exception class BadLoopException : Exception
{ {
@nogc: /**
/** * Params:
* Params: * file = The file where the exception occurred.
* file = The file where the exception occurred. * line = The line number where the exception occurred.
* line = The line number where the exception occurred. * next = The previous exception in the chain of exceptions, if any.
* next = The previous exception in the chain of exceptions, if any. */
*/ this(string file = __FILE__, size_t line = __LINE__, Throwable next = null)
this(string file = __FILE__, size_t line = __LINE__, Throwable next = null) pure nothrow const @safe @nogc
pure @safe nothrow const {
{ super("Event loop cannot be initialized.", file, line, next);
super("Event loop cannot be initialized.", file, line, next); }
}
} }
/** /**
@ -310,24 +323,24 @@ class BadLoopException : Exception
*/ */
@property Loop defaultLoop() @property Loop defaultLoop()
{ {
if (defaultLoop_ !is null) if (defaultLoop_ !is null)
{ {
return defaultLoop_; return defaultLoop_;
} }
version (Epoll) version (Epoll)
{ {
defaultLoop_ = MmapPool.instance.make!EpollLoop; defaultLoop_ = MmapPool.instance.make!EpollLoop;
} }
else version (IOCP) else version (IOCP)
{ {
defaultLoop_ = MmapPool.instance.make!IOCPLoop; defaultLoop_ = MmapPool.instance.make!IOCPLoop;
} }
else version (Kqueue) else version (Kqueue)
{ {
import tanya.async.event.kqueue; import tanya.async.event.kqueue;
defaultLoop_ = MmapPool.instance.make!KqueueLoop; defaultLoop_ = MmapPool.instance.make!KqueueLoop;
} }
return defaultLoop_; return defaultLoop_;
} }
/** /**
@ -339,145 +352,16 @@ class BadLoopException : Exception
* your implementation to this property. * your implementation to this property.
* *
* Params: * Params:
* loop = The event loop. * loop = The event loop.
*/ */
@property void defaultLoop(Loop loop) @property void defaultLoop(Loop loop)
in in
{ {
assert(loop !is null); assert(loop !is null);
} }
body body
{ {
defaultLoop_ = loop; defaultLoop_ = loop;
} }
private Loop defaultLoop_; private Loop defaultLoop_;
/**
* Queue.
*
* Params:
* T = Content type.
*/
class PendingQueue(T)
{
/**
* Creates a new $(D_PSYMBOL Queue).
*/
this()
{
}
/**
* Removes all elements from the queue.
*/
~this()
{
foreach (e; this)
{
MmapPool.instance.dispose(e);
}
}
/**
* Returns: First element.
*/
@property ref T front()
in
{
assert(!empty);
}
body
{
return first.next.content;
}
/**
* Inserts a new element.
*
* Params:
* x = New element.
*
* Returns: $(D_KEYWORD this).
*/
typeof(this) insertBack(T x)
{
Entry* temp = MmapPool.instance.make!Entry;
temp.content = x;
if (empty)
{
first.next = rear = temp;
}
else
{
rear.next = temp;
rear = rear.next;
}
return this;
}
alias insert = insertBack;
/**
* Inserts a new element.
*
* Params:
* x = New element.
*
* Returns: $(D_KEYWORD this).
*/
typeof(this) opOpAssign(string Op)(ref T x)
if (Op == "~")
{
return insertBack(x);
}
/**
* Returns: $(D_KEYWORD true) if the queue is empty.
*/
@property bool empty() const @safe pure nothrow
{
return first.next is null;
}
/**
* Move position to the next element.
*
* Returns: $(D_KEYWORD this).
*/
typeof(this) popFront()
in
{
assert(!empty);
}
body
{
auto n = first.next.next;
MmapPool.instance.dispose(first.next);
first.next = n;
return this;
}
/**
* Queue entry.
*/
protected struct Entry
{
/// Queue item content.
T content;
/// Next list item.
Entry* next;
}
/// The first element of the list.
protected Entry first;
/// The last element of the list.
protected Entry* rear;
}

40
source/tanya/async/protocol.d
View File

@ -18,28 +18,28 @@ import tanya.async.transport;
*/ */
interface Protocol interface Protocol
{ {
/** /**
* Params: * Params:
* data = Read data. * data = Read data.
*/ */
void received(ubyte[] data); void received(ubyte[] data);
/** /**
* Called when a connection is made. * Called when a connection is made.
* *
* Params: * Params:
* transport = Protocol transport. * transport = Protocol transport.
*/ */
void connected(DuplexTransport transport); void connected(DuplexTransport transport);
/** /**
* Called when a connection is lost. * Called when a connection is lost.
* *
* Params: * Params:
* exception = $(D_PSYMBOL Exception) if an error caused * exception = $(D_PSYMBOL Exception) if an error caused
* the disconnect, $(D_KEYWORD null) otherwise. * the disconnect, $(D_KEYWORD null) otherwise.
*/ */
void disconnected(SocketException exception = null); void disconnected(SocketException exception = null);
} }
/** /**

16
source/tanya/async/transport.d
View File

@ -31,13 +31,13 @@ interface ReadTransport : Transport
*/ */
interface WriteTransport : Transport interface WriteTransport : Transport
{ {
/** /**
* Write some data to the transport. * Write some data to the transport.
* *
* Params: * Params:
* data = Data to send. * data = Data to send.
*/ */
void write(ubyte[] data); void write(ubyte[] data);
} }
/** /**
@ -52,7 +52,7 @@ interface DuplexTransport : ReadTransport, WriteTransport
*/ */
interface SocketTransport : Transport interface SocketTransport : Transport
{ {
@property inout(Socket) socket() inout pure nothrow @safe @nogc; @property inout(Socket) socket() inout pure nothrow @safe @nogc;
} }
/** /**

363
source/tanya/async/watcher.d
View File

@ -10,23 +10,24 @@
*/ */
module tanya.async.watcher; module tanya.async.watcher;
import std.functional;
import std.exception;
import tanya.async.loop; import tanya.async.loop;
import tanya.async.protocol; import tanya.async.protocol;
import tanya.async.transport; import tanya.async.transport;
import tanya.container.buffer; import tanya.container.buffer;
import tanya.container.queue;
import tanya.memory; import tanya.memory;
import tanya.memory.mmappool; import tanya.memory.mmappool;
import tanya.network.socket; import tanya.network.socket;
import std.functional;
import std.exception;
version (Windows) version (Windows)
{ {
import core.sys.windows.basetyps; import core.sys.windows.basetyps;
import core.sys.windows.mswsock; import core.sys.windows.mswsock;
import core.sys.windows.winbase; import core.sys.windows.winbase;
import core.sys.windows.windef; import core.sys.windows.windef;
import core.sys.windows.winsock2; import core.sys.windows.winsock2;
} }
/** /**
@ -35,85 +36,89 @@ version (Windows)
*/ */
abstract class Watcher abstract class Watcher
{ {
/// Whether the watcher is active. /// Whether the watcher is active.
bool active; bool active;
/** /**
* Invoke some action on event. * Invoke some action on event.
*/ */
void invoke(); void invoke();
} }
class ConnectionWatcher : Watcher class ConnectionWatcher : Watcher
{ {
/// Watched socket. /// Watched socket.
private Socket socket_; private Socket socket_;
/// Protocol factory. /// Protocol factory.
protected Protocol delegate() protocolFactory; protected Protocol delegate() protocolFactory;
package PendingQueue!IOWatcher incoming; package Queue!IOWatcher incoming;
/** /**
* Params: * Params:
* socket = Socket. * socket = Socket.
*/ */
this(Socket socket) this(Socket socket)
{ {
socket_ = socket; socket_ = socket;
incoming = MmapPool.instance.make!(PendingQueue!IOWatcher); incoming = MmapPool.instance.make!(Queue!IOWatcher);
} }
/// Ditto. /// Ditto.
protected this() protected this()
{ {
} }
~this() ~this()
{ {
MmapPool.instance.dispose(incoming); foreach (w; incoming)
} {
MmapPool.instance.dispose(w);
}
MmapPool.instance.dispose(incoming);
}
/* /*
* Params: * Params:
* P = Protocol should be used. * P = Protocol should be used.
*/ */
void setProtocol(P : Protocol)() void setProtocol(P : Protocol)()
{ {
this.protocolFactory = () => cast(Protocol) MmapPool.instance.make!P; this.protocolFactory = () => cast(Protocol) MmapPool.instance.make!P;
} }
/** /**
* Returns: Socket. * Returns: Socket.
*/ */
@property inout(Socket) socket() inout pure nothrow @nogc @property inout(Socket) socket() inout pure nothrow @nogc
{ {
return socket_; return socket_;
} }
/** /**
* Returns: New protocol instance. * Returns: New protocol instance.
*/ */
@property Protocol protocol() @property Protocol protocol()
in in
{ {
assert(protocolFactory !is null, "Protocol isn't set."); assert(protocolFactory !is null, "Protocol isn't set.");
} }
body body
{ {
return protocolFactory(); return protocolFactory();
} }
/** /**
* Invokes new connection callback. * Invokes new connection callback.
*/ */
override void invoke() override void invoke()
{ {
foreach (io; incoming) foreach (io; incoming)
{ {
io.protocol.connected(cast(DuplexTransport) io.transport); io.protocol.connected(cast(DuplexTransport) io.transport);
} }
} }
} }
/** /**
@ -122,121 +127,121 @@ class ConnectionWatcher : Watcher
*/ */
class IOWatcher : ConnectionWatcher class IOWatcher : ConnectionWatcher
{ {
/// If an exception was thrown the transport should be already invalid. /// If an exception was thrown the transport should be already invalid.
private union private union
{ {
StreamTransport transport_; StreamTransport transport_;
SocketException exception_; SocketException exception_;
} }
private Protocol protocol_; private Protocol protocol_;
/** /**
* Returns: Underlying output buffer. * Returns: Underlying output buffer.
*/ */
package ReadBuffer output; package ReadBuffer output;
/** /**
* Params: * Params:
* transport = Transport. * transport = Transport.
* protocol = New instance of the application protocol. * protocol = New instance of the application protocol.
*/ */
this(StreamTransport transport, Protocol protocol) this(StreamTransport transport, Protocol protocol)
in in
{ {
assert(transport !is null); assert(transport !is null);
assert(protocol !is null); assert(protocol !is null);
} }
body body
{ {
super(); super();
transport_ = transport; transport_ = transport;
protocol_ = protocol; protocol_ = protocol;
output = MmapPool.instance.make!ReadBuffer(); output = MmapPool.instance.make!ReadBuffer();
active = true; active = true;
} }
/** /**
* Destroys the watcher. * Destroys the watcher.
*/ */
protected ~this() protected ~this()
{ {
MmapPool.instance.dispose(output); MmapPool.instance.dispose(output);
MmapPool.instance.dispose(protocol_); MmapPool.instance.dispose(protocol_);
} }
/** /**
* Assigns a transport. * Assigns a transport.
* *
* Params: * Params:
* transport = Transport. * transport = Transport.
* protocol = Application protocol. * protocol = Application protocol.
* *
* Returns: $(D_KEYWORD this). * Returns: $(D_KEYWORD this).
*/ */
IOWatcher opCall(StreamTransport transport, Protocol protocol) pure nothrow @nogc IOWatcher opCall(StreamTransport transport, Protocol protocol) pure nothrow @nogc
in in
{ {
assert(transport !is null); assert(transport !is null);
assert(protocol !is null); assert(protocol !is null);
} }
body body
{ {
transport_ = transport; transport_ = transport;
protocol_ = protocol; protocol_ = protocol;
active = true; active = true;
return this; return this;
} }
/** /**
* Returns: Transport used by this watcher. * Returns: Transport used by this watcher.
*/ */
@property inout(StreamTransport) transport() inout pure nothrow @nogc @property inout(StreamTransport) transport() inout pure nothrow @nogc
{ {
return transport_; return transport_;
} }
/** /**
* Sets an exception occurred during a read/write operation. * Sets an exception occurred during a read/write operation.
* *
* Params: * Params:
* exception = Thrown exception. * exception = Thrown exception.
*/ */
@property void exception(SocketException exception) pure nothrow @nogc @property void exception(SocketException exception) pure nothrow @nogc
{ {
exception_ = exception; exception_ = exception;
} }
/** /**
* Returns: Application protocol. * Returns: Application protocol.
*/ */
override @property Protocol protocol() pure nothrow @safe @nogc override @property Protocol protocol() pure nothrow @safe @nogc
{ {
return protocol_; return protocol_;
} }
/** /**
* Returns: Socket. * Returns: Socket.
*/ */
override @property inout(Socket) socket() inout pure nothrow @nogc override @property inout(Socket) socket() inout pure nothrow @nogc
{ {
return transport.socket; return transport.socket;
} }
/** /**
* Invokes the watcher callback. * Invokes the watcher callback.
*/ */
override void invoke() override void invoke()
{ {
if (output.length) if (output.length)
{ {
protocol.received(output[0..$]); protocol.received(output[0..$]);
output.clear(); output.clear();
} }
else else
{ {
protocol.disconnected(exception_); protocol.disconnected(exception_);
active = false; active = false;
} }
} }
} }

5
source/tanya/container/queue.d
View File

@ -121,7 +121,7 @@ class Queue(T)
/** /**
* Returns: $(D_KEYWORD true) if the queue is empty. * Returns: $(D_KEYWORD true) if the queue is empty.
*/ */
@property bool empty() inout const @property bool empty() inout const pure nothrow @safe @nogc
{ {
return first.next is null; return first.next is null;
} }
@ -170,7 +170,8 @@ class Queue(T)
} }
/** /**
* $(D_KEYWORD foreach) iteration. * $(D_KEYWORD foreach) iteration. The elements will be automatically
* dequeued.
* *
* Params: * Params:
* dg = $(D_KEYWORD foreach) body. * dg = $(D_KEYWORD foreach) body.