diff --git a/source/tanya/async/event/epoll.d b/source/tanya/async/event/epoll.d index 7fec708..7717b52 100644 --- a/source/tanya/async/event/epoll.d +++ b/source/tanya/async/event/epoll.d @@ -28,151 +28,151 @@ import std.algorithm.comparison; class EpollLoop : SelectorLoop { - protected int fd; - private epoll_event[] events; + protected int fd; + private epoll_event[] events; - /** - * Initializes the loop. - */ - this() - { - if ((fd = epoll_create1(EPOLL_CLOEXEC)) < 0) - { - throw MmapPool.instance.make!BadLoopException("epoll initialization failed"); - } - super(); - events = MmapPool.instance.makeArray!epoll_event(maxEvents); - } + /** + * Initializes the loop. + */ + this() + { + if ((fd = epoll_create1(EPOLL_CLOEXEC)) < 0) + { + throw MmapPool.instance.make!BadLoopException("epoll initialization failed"); + } + super(); + events = MmapPool.instance.makeArray!epoll_event(maxEvents); + } - /** - * Free loop internals. - */ - ~this() - { - MmapPool.instance.dispose(events); - close(fd); - } + /** + * Free loop internals. + */ + ~this() + { + MmapPool.instance.dispose(events); + close(fd); + } - /** - * 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. - */ - protected override bool reify(ConnectionWatcher watcher, EventMask oldEvents, EventMask events) - in - { - assert(watcher !is null); - } - body - { - int op = EPOLL_CTL_DEL; - epoll_event ev; + /** + * 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. + */ + protected override bool reify(ConnectionWatcher watcher, EventMask oldEvents, EventMask events) + in + { + assert(watcher !is null); + } + body + { + int op = EPOLL_CTL_DEL; + epoll_event ev; - if (events == oldEvents) - { - return true; - } - if (events && oldEvents) - { - op = EPOLL_CTL_MOD; - } - else if (events && !oldEvents) - { - op = EPOLL_CTL_ADD; - } + if (events == oldEvents) + { + return true; + } + if (events && oldEvents) + { + op = EPOLL_CTL_MOD; + } + else if (events && !oldEvents) + { + op = EPOLL_CTL_ADD; + } - ev.data.fd = watcher.socket.handle; - ev.events = (events & (Event.read | Event.accept) ? EPOLLIN | EPOLLPRI : 0) - | (events & Event.write ? EPOLLOUT : 0) - | EPOLLET; + ev.data.fd = watcher.socket.handle; + ev.events = (events & (Event.read | Event.accept) ? EPOLLIN | EPOLLPRI : 0) + | (events & Event.write ? EPOLLOUT : 0) + | 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. - */ - protected override void poll() - { - // Don't block - immutable timeout = cast(immutable int) blockTime.total!"msecs"; - auto eventCount = epoll_wait(fd, events.ptr, maxEvents, timeout); + /** + * Does the actual polling. + */ + protected override void poll() + { + // Don't block + immutable timeout = cast(immutable int) blockTime.total!"msecs"; + auto eventCount = epoll_wait(fd, events.ptr, maxEvents, timeout); - if (eventCount < 0) - { - if (errno != EINTR) - { - throw theAllocator.make!BadLoopException(); - } - return; - } + if (eventCount < 0) + { + if (errno != EINTR) + { + throw theAllocator.make!BadLoopException(); + } + return; + } - for (auto i = 0; i < eventCount; ++i) - { - auto io = cast(IOWatcher) connections[events[i].data.fd]; + for (auto i = 0; i < eventCount; ++i) + { + auto io = cast(IOWatcher) connections[events[i].data.fd]; - if (io is null) - { - acceptConnections(connections[events[i].data.fd]); - } - else if (events[i].events & EPOLLERR) - { - kill(io, null); - } - else if (events[i].events & (EPOLLIN | EPOLLPRI | EPOLLHUP)) - { - auto transport = cast(SelectorStreamTransport) io.transport; - assert(transport !is null); + if (io is null) + { + acceptConnections(connections[events[i].data.fd]); + } + else if (events[i].events & EPOLLERR) + { + kill(io, null); + } + else if (events[i].events & (EPOLLIN | EPOLLPRI | EPOLLHUP)) + { + auto transport = cast(SelectorStreamTransport) io.transport; + assert(transport !is null); - SocketException exception; - try - { - ptrdiff_t received; - do - { - received = transport.socket.receive(io.output[]); - io.output += received; - } - while (received); - } - catch (SocketException e) - { - exception = e; - } - if (transport.socket.disconnected) - { - kill(io, exception); - } - else if (io.output.length) - { - swapPendings.insertBack(io); - } - } - else if (events[i].events & EPOLLOUT) - { - auto transport = cast(SelectorStreamTransport) io.transport; - assert(transport !is null); + SocketException exception; + try + { + ptrdiff_t received; + do + { + received = transport.socket.receive(io.output[]); + io.output += received; + } + while (received); + } + catch (SocketException e) + { + exception = e; + } + if (transport.socket.disconnected) + { + kill(io, exception); + } + else if (io.output.length) + { + swapPendings.insertBack(io); + } + } + else if (events[i].events & EPOLLOUT) + { + auto transport = cast(SelectorStreamTransport) io.transport; + assert(transport !is null); - transport.writeReady = true; - if (transport.input.length) - { - feed(transport); - } - } - } - } + transport.writeReady = true; + if (transport.input.length) + { + feed(transport); + } + } + } + } - /** - * Returns: The blocking time. - */ - override protected @property inout(Duration) blockTime() - inout @safe pure nothrow - { - return min(super.blockTime, 1.dur!"seconds"); - } + /** + * Returns: The blocking time. + */ + override protected @property inout(Duration) blockTime() + inout @safe pure nothrow + { + return min(super.blockTime, 1.dur!"seconds"); + } } diff --git a/source/tanya/async/event/iocp.d b/source/tanya/async/event/iocp.d index bcfda71..935dd5e 100644 --- a/source/tanya/async/event/iocp.d +++ b/source/tanya/async/event/iocp.d @@ -28,267 +28,267 @@ import core.sys.windows.winsock2; class IOCPStreamTransport : StreamTransport { - private OverlappedConnectedSocket socket_; + private OverlappedConnectedSocket socket_; - private WriteBuffer input; + private WriteBuffer input; - /** - * Creates new completion port transport. - * Params: - * socket = Socket. - */ - this(OverlappedConnectedSocket socket) - in - { - assert(socket !is null); - } - body - { - socket_ = socket; - input = MmapPool.instance.make!WriteBuffer(); - } + /** + * Creates new completion port transport. + * Params: + * socket = Socket. + */ + this(OverlappedConnectedSocket socket) + in + { + assert(socket !is null); + } + body + { + socket_ = socket; + input = MmapPool.instance.make!WriteBuffer(); + } - ~this() - { - MmapPool.instance.dispose(input); - } + ~this() + { + MmapPool.instance.dispose(input); + } - @property inout(OverlappedConnectedSocket) socket() inout pure nothrow @safe @nogc - { - return socket_; - } + @property inout(OverlappedConnectedSocket) socket() inout pure nothrow @safe @nogc + { + return socket_; + } - /** - * Write some data to the transport. - * - * Params: - * data = Data to send. - */ - void write(ubyte[] data) - { - immutable empty = input.length == 0; - input ~= data; - if (empty) - { - SocketState overlapped; - try - { - overlapped = MmapPool.instance.make!SocketState; - socket.beginSend(input[], overlapped); - } - catch (SocketException e) - { - MmapPool.instance.dispose(overlapped); - MmapPool.instance.dispose(e); - } - } - } + /** + * Write some data to the transport. + * + * Params: + * data = Data to send. + */ + void write(ubyte[] data) + { + immutable empty = input.length == 0; + input ~= data; + if (empty) + { + SocketState overlapped; + try + { + overlapped = MmapPool.instance.make!SocketState; + socket.beginSend(input[], overlapped); + } + catch (SocketException e) + { + MmapPool.instance.dispose(overlapped); + MmapPool.instance.dispose(e); + } + } + } } -class IOCPLoop : Loop -{ - protected HANDLE completionPort; + class IOCPLoop : Loop + { + protected HANDLE completionPort; - protected OVERLAPPED overlap; + protected OVERLAPPED overlap; - /** - * Initializes the loop. - */ - this() - { - super(); + /** + * Initializes the loop. + */ + this() + { + super(); - completionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); - if (!completionPort) - { - throw theAllocator.make!BadLoopException("Creating completion port failed"); - } - } + completionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); + if (!completionPort) + { + throw theAllocator.make!BadLoopException("Creating completion port failed"); + } + } - /** - * 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. - */ - override protected bool reify(ConnectionWatcher watcher, - EventMask oldEvents, - EventMask events) - { - SocketState overlapped; - if (!(oldEvents & Event.accept) && (events & Event.accept)) - { - auto socket = cast(OverlappedStreamSocket) watcher.socket; - assert(socket !is null); + /** + * 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. + */ + override protected bool reify(ConnectionWatcher watcher, + EventMask oldEvents, + EventMask events) + { + SocketState overlapped; + if (!(oldEvents & Event.accept) && (events & Event.accept)) + { + auto socket = cast(OverlappedStreamSocket) watcher.socket; + assert(socket !is null); - if (CreateIoCompletionPort(cast(HANDLE) socket.handle, - completionPort, - cast(ULONG_PTR) (cast(void*) watcher), - 0) !is completionPort) - { - return false; - } + if (CreateIoCompletionPort(cast(HANDLE) socket.handle, + completionPort, + cast(ULONG_PTR) (cast(void*) watcher), + 0) !is completionPort) + { + return false; + } - try - { - overlapped = MmapPool.instance.make!SocketState; - socket.beginAccept(overlapped); - } - catch (SocketException e) - { - MmapPool.instance.dispose(overlapped); - theAllocator.dispose(e); - return false; - } - } - if (!(oldEvents & Event.read) && (events & Event.read) - || !(oldEvents & Event.write) && (events & Event.write)) - { - auto io = cast(IOWatcher) watcher; - assert(io !is null); + try + { + overlapped = MmapPool.instance.make!SocketState; + socket.beginAccept(overlapped); + } + catch (SocketException e) + { + MmapPool.instance.dispose(overlapped); + theAllocator.dispose(e); + return false; + } + } + if (!(oldEvents & Event.read) && (events & Event.read) + || !(oldEvents & Event.write) && (events & Event.write)) + { + auto io = cast(IOWatcher) watcher; + assert(io !is null); - auto transport = cast(IOCPStreamTransport) io.transport; - assert(transport !is null); + auto transport = cast(IOCPStreamTransport) io.transport; + assert(transport !is null); - if (CreateIoCompletionPort(cast(HANDLE) transport.socket.handle, - completionPort, - cast(ULONG_PTR) (cast(void*) watcher), - 0) !is completionPort) - { - return false; - } + if (CreateIoCompletionPort(cast(HANDLE) transport.socket.handle, + completionPort, + cast(ULONG_PTR) (cast(void*) watcher), + 0) !is completionPort) + { + return false; + } - // Begin to read - if (!(oldEvents & Event.read) && (events & Event.read)) - { - try - { - overlapped = MmapPool.instance.make!SocketState; - transport.socket.beginReceive(io.output[], overlapped); - } - catch (SocketException e) - { - MmapPool.instance.dispose(overlapped); - theAllocator.dispose(e); - return false; - } - } - } - return true; - } + // Begin to read + if (!(oldEvents & Event.read) && (events & Event.read)) + { + try + { + overlapped = MmapPool.instance.make!SocketState; + transport.socket.beginReceive(io.output[], overlapped); + } + catch (SocketException e) + { + MmapPool.instance.dispose(overlapped); + theAllocator.dispose(e); + return false; + } + } + } + return true; + } - /** - * Does the actual polling. - */ - override protected void poll() - { - DWORD lpNumberOfBytes; - ULONG_PTR key; - LPOVERLAPPED overlap; - immutable timeout = cast(immutable int) blockTime.total!"msecs"; + /** + * Does the actual polling. + */ + override protected void poll() + { + DWORD lpNumberOfBytes; + ULONG_PTR key; + LPOVERLAPPED overlap; + immutable timeout = cast(immutable int) blockTime.total!"msecs"; - auto result = GetQueuedCompletionStatus(completionPort, - &lpNumberOfBytes, - &key, - &overlap, - timeout); - if (result == FALSE && overlap == NULL) - { - return; // Timeout - } + auto result = GetQueuedCompletionStatus(completionPort, + &lpNumberOfBytes, + &key, + &overlap, + timeout); + if (result == FALSE && overlap == NULL) + { + return; // Timeout + } - auto overlapped = (cast(SocketState) ((cast(void*) overlap) - 8)); - assert(overlapped !is null); - scope (failure) - { - MmapPool.instance.dispose(overlapped); - } + auto overlapped = (cast(SocketState) ((cast(void*) overlap) - 8)); + assert(overlapped !is null); + scope (failure) + { + MmapPool.instance.dispose(overlapped); + } - switch (overlapped.event) - { - case OverlappedSocketEvent.accept: - auto connection = cast(ConnectionWatcher) (cast(void*) key); - assert(connection !is null); + switch (overlapped.event) + { + case OverlappedSocketEvent.accept: + auto connection = cast(ConnectionWatcher) (cast(void*) key); + assert(connection !is null); - auto listener = cast(OverlappedStreamSocket) connection.socket; - assert(listener !is null); + auto listener = cast(OverlappedStreamSocket) connection.socket; + assert(listener !is null); - auto socket = listener.endAccept(overlapped); - auto transport = MmapPool.instance.make!IOCPStreamTransport(socket); - auto io = MmapPool.instance.make!IOWatcher(transport, connection.protocol); + auto socket = listener.endAccept(overlapped); + auto transport = MmapPool.instance.make!IOCPStreamTransport(socket); + 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); - listener.beginAccept(overlapped); - break; - case OverlappedSocketEvent.read: - auto io = cast(IOWatcher) (cast(void*) key); - assert(io !is null); - if (!io.active) - { - MmapPool.instance.dispose(io); - MmapPool.instance.dispose(overlapped); - return; - } + swapPendings.insertBack(connection); + listener.beginAccept(overlapped); + break; + case OverlappedSocketEvent.read: + auto io = cast(IOWatcher) (cast(void*) key); + assert(io !is null); + if (!io.active) + { + MmapPool.instance.dispose(io); + MmapPool.instance.dispose(overlapped); + return; + } - auto transport = cast(IOCPStreamTransport) io.transport; - assert(transport !is null); + auto transport = cast(IOCPStreamTransport) io.transport; + assert(transport !is null); - int received; - SocketException exception; - try - { - received = transport.socket.endReceive(overlapped); - } - catch (SocketException e) - { - exception = e; - } - if (transport.socket.disconnected) - { - // We want to get one last notification to destroy the watcher - transport.socket.beginReceive(io.output[], overlapped); - kill(io, exception); - } - else if (received > 0) - { - immutable full = io.output.free == received; + int received; + SocketException exception; + try + { + received = transport.socket.endReceive(overlapped); + } + catch (SocketException e) + { + exception = e; + } + if (transport.socket.disconnected) + { + // We want to get one last notification to destroy the watcher + transport.socket.beginReceive(io.output[], overlapped); + kill(io, exception); + } + else if (received > 0) + { + immutable full = io.output.free == received; - io.output += received; - // Receive was interrupted because the buffer is full. We have to continue - if (full) - { - transport.socket.beginReceive(io.output[], overlapped); - } - swapPendings.insertBack(io); - } - break; - case OverlappedSocketEvent.write: - auto io = cast(IOWatcher) (cast(void*) key); - assert(io !is null); + io.output += received; + // Receive was interrupted because the buffer is full. We have to continue + if (full) + { + transport.socket.beginReceive(io.output[], overlapped); + } + swapPendings.insertBack(io); + } + break; + case OverlappedSocketEvent.write: + auto io = cast(IOWatcher) (cast(void*) key); + assert(io !is null); - auto transport = cast(IOCPStreamTransport) io.transport; - assert(transport !is null); + auto transport = cast(IOCPStreamTransport) io.transport; + assert(transport !is null); - transport.input += transport.socket.endSend(overlapped); - if (transport.input.length) - { - transport.socket.beginSend(transport.input[], overlapped); - } - else - { - transport.socket.beginReceive(io.output[], overlapped); - } - break; - default: - assert(false, "Unknown event"); - } - } + transport.input += transport.socket.endSend(overlapped); + if (transport.input.length) + { + transport.socket.beginSend(transport.input[], overlapped); + } + else + { + transport.socket.beginReceive(io.output[], overlapped); + } + break; + default: + assert(false, "Unknown event"); + } + } } diff --git a/source/tanya/async/event/kqueue.d b/source/tanya/async/event/kqueue.d index 773ea12..fa9261f 100644 --- a/source/tanya/async/event/kqueue.d +++ b/source/tanya/async/event/kqueue.d @@ -12,117 +12,117 @@ module tanya.async.event.kqueue; version (OSX) { - version = MissingKevent; + version = MissingKevent; } else version (iOS) { - version = MissingKevent; + version = MissingKevent; } else version (TVOS) { - version = MissingKevent; + version = MissingKevent; } else version (WatchOS) { - version = MissingKevent; + version = MissingKevent; } else version (OpenBSD) { - version = MissingKevent; + version = MissingKevent; } else version (DragonFlyBSD) { - version = MissingKevent; + version = MissingKevent; } version (MissingKevent) { - extern (C): - nothrow: - @nogc: + extern (C): + nothrow: + @nogc: - import core.stdc.stdint; // intptr_t, uintptr_t - import core.sys.posix.time; // timespec + import core.stdc.stdint; // intptr_t, uintptr_t + import core.sys.posix.time; // timespec - enum : short - { - EVFILT_READ = -1, - EVFILT_WRITE = -2, - EVFILT_AIO = -3, /* attached to aio requests */ - EVFILT_VNODE = -4, /* attached to vnodes */ - EVFILT_PROC = -5, /* attached to struct proc */ - EVFILT_SIGNAL = -6, /* attached to struct proc */ - EVFILT_TIMER = -7, /* timers */ - EVFILT_MACHPORT = -8, /* Mach portsets */ - EVFILT_FS = -9, /* filesystem events */ - EVFILT_USER = -10, /* User events */ - EVFILT_VM = -12, /* virtual memory events */ - EVFILT_SYSCOUNT = 11 - } + enum : short + { + EVFILT_READ = -1, + EVFILT_WRITE = -2, + EVFILT_AIO = -3, /* attached to aio requests */ + EVFILT_VNODE = -4, /* attached to vnodes */ + EVFILT_PROC = -5, /* attached to struct proc */ + EVFILT_SIGNAL = -6, /* attached to struct proc */ + EVFILT_TIMER = -7, /* timers */ + EVFILT_MACHPORT = -8, /* Mach portsets */ + EVFILT_FS = -9, /* filesystem events */ + EVFILT_USER = -10, /* User events */ + EVFILT_VM = -12, /* virtual memory events */ + EVFILT_SYSCOUNT = 11 + } - extern(D) void EV_SET(kevent_t* kevp, typeof(kevent_t.tupleof) args) - { - *kevp = kevent_t(args); - } + extern(D) void EV_SET(kevent_t* kevp, typeof(kevent_t.tupleof) args) + { + *kevp = kevent_t(args); + } - struct kevent_t - { - uintptr_t ident; /* identifier for this event */ - short filter; /* filter for event */ - ushort flags; - uint fflags; - intptr_t data; - void *udata; /* opaque user data identifier */ - } + struct kevent_t + { + uintptr_t ident; /* identifier for this event */ + short filter; /* filter for event */ + ushort flags; + uint fflags; + intptr_t data; + void *udata; /* opaque user data identifier */ + } - enum - { - /* actions */ - EV_ADD = 0x0001, /* add event to kq (implies enable) */ - EV_DELETE = 0x0002, /* delete event from kq */ - EV_ENABLE = 0x0004, /* enable event */ - EV_DISABLE = 0x0008, /* disable event (not reported) */ + enum + { + /* actions */ + EV_ADD = 0x0001, /* add event to kq (implies enable) */ + EV_DELETE = 0x0002, /* delete event from kq */ + EV_ENABLE = 0x0004, /* enable event */ + EV_DISABLE = 0x0008, /* disable event (not reported) */ - /* flags */ - EV_ONESHOT = 0x0010, /* only report one occurrence */ - EV_CLEAR = 0x0020, /* clear event state after reporting */ - EV_RECEIPT = 0x0040, /* force EV_ERROR on success, data=0 */ - EV_DISPATCH = 0x0080, /* disable event after reporting */ + /* flags */ + EV_ONESHOT = 0x0010, /* only report one occurrence */ + EV_CLEAR = 0x0020, /* clear event state after reporting */ + EV_RECEIPT = 0x0040, /* force EV_ERROR on success, data=0 */ + EV_DISPATCH = 0x0080, /* disable event after reporting */ - EV_SYSFLAGS = 0xF000, /* reserved by system */ - EV_FLAG1 = 0x2000, /* filter-specific flag */ + EV_SYSFLAGS = 0xF000, /* reserved by system */ + EV_FLAG1 = 0x2000, /* filter-specific flag */ - /* returned values */ - EV_EOF = 0x8000, /* EOF detected */ - EV_ERROR = 0x4000, /* error, data contains errno */ - } + /* returned values */ + EV_EOF = 0x8000, /* EOF detected */ + EV_ERROR = 0x4000, /* error, data contains errno */ + } - int kqueue(); - int kevent(int kq, const kevent_t *changelist, int nchanges, - kevent_t *eventlist, int nevents, - const timespec *timeout); + int kqueue(); + int kevent(int kq, const kevent_t *changelist, int nchanges, + kevent_t *eventlist, int nevents, + const timespec *timeout); } version (OSX) { - version = MacBSD; + version = MacBSD; } else version (iOS) { - version = MacBSD; + version = MacBSD; } else version (FreeBSD) { - version = MacBSD; - public import core.sys.freebsd.sys.event; + version = MacBSD; + public import core.sys.freebsd.sys.event; } else version (OpenBSD) { - version = MacBSD; + version = MacBSD; } else version (DragonFlyBSD) { - version = MacBSD; + version = MacBSD; } version (MacBSD): @@ -142,208 +142,208 @@ import std.algorithm.comparison; class KqueueLoop : SelectorLoop { - protected int fd; - private kevent_t[] events; - private kevent_t[] changes; - private size_t changeCount; + protected int fd; + private kevent_t[] events; + private kevent_t[] changes; + private size_t changeCount; - /** - * Returns: Maximal event count can be got at a time - * (should be supported by the backend). - */ - override protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc - { - return cast(uint) events.length; - } + /** + * Returns: Maximal event count can be got at a time + * (should be supported by the backend). + */ + override protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc + { + return cast(uint) events.length; + } - this() - { - super(); + this() + { + super(); - if ((fd = kqueue()) == -1) - { - throw MmapPool.instance.make!BadLoopException("epoll initialization failed"); - } - events = MmapPool.instance.makeArray!kevent_t(64); - changes = MmapPool.instance.makeArray!kevent_t(64); - } + if ((fd = kqueue()) == -1) + { + throw MmapPool.instance.make!BadLoopException("epoll initialization failed"); + } + events = MmapPool.instance.makeArray!kevent_t(64); + changes = MmapPool.instance.makeArray!kevent_t(64); + } - /** - * Free loop internals. - */ - ~this() - { - MmapPool.instance.dispose(events); - MmapPool.instance.dispose(changes); - close(fd); - } + /** + * Free loop internals. + */ + ~this() + { + MmapPool.instance.dispose(events); + MmapPool.instance.dispose(changes); + close(fd); + } - private void set(socket_t socket, short filter, ushort flags) - { - if (changes.length <= changeCount) - { - MmapPool.instance.resizeArray(changes, changeCount + maxEvents); - } - EV_SET(&changes[changeCount], - cast(ulong) socket, - filter, - flags, - 0U, - 0L, - null); - ++changeCount; - } + private void set(socket_t socket, short filter, ushort flags) + { + if (changes.length <= changeCount) + { + MmapPool.instance.resizeArray(changes, changeCount + maxEvents); + } + EV_SET(&changes[changeCount], + cast(ulong) socket, + filter, + flags, + 0U, + 0L, + null); + ++changeCount; + } - /** - * 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. - */ - override protected bool reify(ConnectionWatcher watcher, - EventMask oldEvents, - EventMask events) - { - if (events != oldEvents) - { - if (oldEvents & Event.read || oldEvents & Event.accept) - { - set(watcher.socket.handle, EVFILT_READ, EV_DELETE); - } - if (oldEvents & Event.write) - { - set(watcher.socket.handle, EVFILT_WRITE, EV_DELETE); - } - } - if (events & (Event.read | events & Event.accept)) - { - set(watcher.socket.handle, EVFILT_READ, EV_ADD | EV_ENABLE); - } - if (events & Event.write) - { - set(watcher.socket.handle, EVFILT_WRITE, EV_ADD | EV_DISPATCH); - } - return true; - } + /** + * 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. + */ + override protected bool reify(ConnectionWatcher watcher, + EventMask oldEvents, + EventMask events) + { + if (events != oldEvents) + { + if (oldEvents & Event.read || oldEvents & Event.accept) + { + set(watcher.socket.handle, EVFILT_READ, EV_DELETE); + } + if (oldEvents & Event.write) + { + set(watcher.socket.handle, EVFILT_WRITE, EV_DELETE); + } + } + if (events & (Event.read | events & Event.accept)) + { + set(watcher.socket.handle, EVFILT_READ, EV_ADD | EV_ENABLE); + } + if (events & Event.write) + { + set(watcher.socket.handle, EVFILT_WRITE, EV_ADD | EV_DISPATCH); + } + return true; + } - /** - * Does the actual polling. - */ - protected override void poll() - { - timespec ts; - blockTime.split!("seconds", "nsecs")(ts.tv_sec, ts.tv_nsec); + /** + * Does the actual polling. + */ + protected override void poll() + { + timespec ts; + blockTime.split!("seconds", "nsecs")(ts.tv_sec, ts.tv_nsec); - if (changeCount > maxEvents) - { - MmapPool.instance.resizeArray(events, changes.length); - } + if (changeCount > maxEvents) + { + MmapPool.instance.resizeArray(events, changes.length); + } - auto eventCount = kevent(fd, changes.ptr, cast(int) changeCount, events.ptr, maxEvents, &ts); - changeCount = 0; + auto eventCount = kevent(fd, changes.ptr, cast(int) changeCount, events.ptr, maxEvents, &ts); + changeCount = 0; - if (eventCount < 0) - { - if (errno != EINTR) - { - throw theAllocator.make!BadLoopException(); - } - return; - } + if (eventCount < 0) + { + if (errno != EINTR) + { + throw theAllocator.make!BadLoopException(); + } + return; + } - for (int i; i < eventCount; ++i) - { - assert(connections.length > events[i].ident); + for (int i; i < eventCount; ++i) + { + assert(connections.length > events[i].ident); - IOWatcher io = cast(IOWatcher) connections[events[i].ident]; - // If it is a ConnectionWatcher. Accept connections. - if (io is null) - { - acceptConnections(connections[events[i].ident]); - } - else if (events[i].flags & EV_ERROR) - { - kill(io, null); - } - else if (events[i].filter == EVFILT_READ) - { - auto transport = cast(SelectorStreamTransport) io.transport; - assert(transport !is null); + IOWatcher io = cast(IOWatcher) connections[events[i].ident]; + // If it is a ConnectionWatcher. Accept connections. + if (io is null) + { + acceptConnections(connections[events[i].ident]); + } + else if (events[i].flags & EV_ERROR) + { + kill(io, null); + } + else if (events[i].filter == EVFILT_READ) + { + auto transport = cast(SelectorStreamTransport) io.transport; + assert(transport !is null); - SocketException exception; - try - { - ptrdiff_t received; - do - { - received = transport.socket.receive(io.output[]); - io.output += received; - } - while (received); - } - catch (SocketException e) - { - exception = e; - } - if (transport.socket.disconnected) - { - kill(io, exception); - } - else if (io.output.length) - { - swapPendings.insertBack(io); - } - } - else if (events[i].filter == EVFILT_WRITE) - { - auto transport = cast(SelectorStreamTransport) io.transport; - assert(transport !is null); + SocketException exception; + try + { + ptrdiff_t received; + do + { + received = transport.socket.receive(io.output[]); + io.output += received; + } + while (received); + } + catch (SocketException e) + { + exception = e; + } + if (transport.socket.disconnected) + { + kill(io, exception); + } + else if (io.output.length) + { + swapPendings.insertBack(io); + } + } + else if (events[i].filter == EVFILT_WRITE) + { + auto transport = cast(SelectorStreamTransport) io.transport; + assert(transport !is null); - transport.writeReady = true; - if (transport.input.length) - { - feed(transport); - } - } - } - } + transport.writeReady = true; + if (transport.input.length) + { + feed(transport); + } + } + } + } - /** - * Returns: The blocking time. - */ - override protected @property inout(Duration) blockTime() - inout @safe pure nothrow - { - return min(super.blockTime, 1.dur!"seconds"); - } + /** + * Returns: The blocking time. + */ + override protected @property inout(Duration) blockTime() + inout @safe pure nothrow + { + return min(super.blockTime, 1.dur!"seconds"); + } - /** - * If the transport couldn't send the data, the further sending should - * be handled by the event loop. - * - * Params: - * transport = Transport. - * exception = Exception thrown on sending. - * - * Returns: $(D_KEYWORD true) if the operation could be successfully - * completed or scheduled, $(D_KEYWORD false) otherwise (the - * transport is be destroyed then). - */ - protected override bool feed(SelectorStreamTransport transport, SocketException exception = null) - { - if (!super.feed(transport, exception)) - { - return false; - } - if (!transport.writeReady) - { - set(transport.socket.handle, EVFILT_WRITE, EV_DISPATCH); - return true; - } - return false; - } + /** + * If the transport couldn't send the data, the further sending should + * be handled by the event loop. + * + * Params: + * transport = Transport. + * exception = Exception thrown on sending. + * + * Returns: $(D_KEYWORD true) if the operation could be successfully + * completed or scheduled, $(D_KEYWORD false) otherwise (the + * transport is be destroyed then). + */ + protected override bool feed(SelectorStreamTransport transport, SocketException exception = null) + { + if (!super.feed(transport, exception)) + { + return false; + } + if (!transport.writeReady) + { + set(transport.socket.handle, EVFILT_WRITE, EV_DISPATCH); + return true; + } + return false; + } } diff --git a/source/tanya/async/event/selector.d b/source/tanya/async/event/selector.d index e9a103d..c35a782 100644 --- a/source/tanya/async/event/selector.d +++ b/source/tanya/async/event/selector.d @@ -27,240 +27,240 @@ import core.stdc.errno; */ class SelectorStreamTransport : StreamTransport { - private ConnectedSocket socket_; + private ConnectedSocket socket_; - /// Input buffer. - package WriteBuffer input; + /// Input buffer. + package WriteBuffer input; - private SelectorLoop loop; + private SelectorLoop loop; - /// Received notification that the underlying socket is write-ready. - package bool writeReady; + /// Received notification that the underlying socket is write-ready. + package bool writeReady; - /** - * Params: - * loop = Event loop. - * socket = Socket. - */ - this(SelectorLoop loop, ConnectedSocket socket) - { - socket_ = socket; - this.loop = loop; - input = MmapPool.instance.make!WriteBuffer(); - } + /** + * Params: + * loop = Event loop. + * socket = Socket. + */ + this(SelectorLoop loop, ConnectedSocket socket) + { + socket_ = socket; + this.loop = loop; + input = MmapPool.instance.make!WriteBuffer(); + } - /** - * Close the transport and deallocate the data buffers. - */ - ~this() - { - MmapPool.instance.dispose(input); - } + /** + * Close the transport and deallocate the data buffers. + */ + ~this() + { + MmapPool.instance.dispose(input); + } - /** - * Returns: Transport socket. - */ - inout(ConnectedSocket) socket() inout pure nothrow @safe @nogc - { - return socket_; - } + /** + * Returns: Transport socket. + */ + inout(ConnectedSocket) socket() inout pure nothrow @safe @nogc + { + return socket_; + } - /** - * Write some data to the transport. - * - * Params: - * data = Data to send. - */ - void write(ubyte[] data) - { - if (!data.length) - { - return; - } - // Try to write if the socket is write ready. - if (writeReady) - { - ptrdiff_t sent; - SocketException exception; - try - { - sent = socket.send(data); - if (sent == 0) - { - writeReady = false; - } - } - catch (SocketException e) - { - writeReady = false; - exception = e; - } - if (sent < data.length) - { - input ~= data[sent..$]; - loop.feed(this, exception); - } - } - else - { - input ~= data; - } - } + /** + * Write some data to the transport. + * + * Params: + * data = Data to send. + */ + void write(ubyte[] data) + { + if (!data.length) + { + return; + } + // Try to write if the socket is write ready. + if (writeReady) + { + ptrdiff_t sent; + SocketException exception; + try + { + sent = socket.send(data); + if (sent == 0) + { + writeReady = false; + } + } + catch (SocketException e) + { + writeReady = false; + exception = e; + } + if (sent < data.length) + { + input ~= data[sent..$]; + loop.feed(this, exception); + } + } + else + { + input ~= data; + } + } } abstract class SelectorLoop : Loop { - /// Pending connections. - protected ConnectionWatcher[] connections; + /// Pending connections. + protected ConnectionWatcher[] connections; - this() - { - super(); - connections = MmapPool.instance.makeArray!ConnectionWatcher(maxEvents); - } + this() + { + super(); + connections = MmapPool.instance.makeArray!ConnectionWatcher(maxEvents); + } - ~this() - { - foreach (ref connection; connections) - { - // We want to free only IOWatchers. ConnectionWatcher are created by the - // user and should be freed by himself. - auto io = cast(IOWatcher) connection; - if (io !is null) - { - MmapPool.instance.dispose(io); - connection = null; - } - } - MmapPool.instance.dispose(connections); - } + ~this() + { + foreach (ref connection; connections) + { + // We want to free only IOWatchers. ConnectionWatcher are created by the + // user and should be freed by himself. + auto io = cast(IOWatcher) connection; + if (io !is null) + { + MmapPool.instance.dispose(io); + connection = null; + } + } + MmapPool.instance.dispose(connections); + } - /** - * If the transport couldn't send the data, the further sending should - * be handled by the event loop. - * - * Params: - * transport = Transport. - * exception = Exception thrown on sending. - * - * Returns: $(D_KEYWORD true) if the operation could be successfully - * completed or scheduled, $(D_KEYWORD false) otherwise (the - * transport will be destroyed then). - */ - protected bool feed(SelectorStreamTransport transport, SocketException exception = null) - { - while (transport.input.length && transport.writeReady) - { - try - { - ptrdiff_t sent = transport.socket.send(transport.input[]); - if (sent == 0) - { - transport.writeReady = false; - } - else - { - transport.input += sent; - } - } - catch (SocketException e) - { - exception = e; - transport.writeReady = false; - } - } - if (exception !is null) - { - auto watcher = cast(IOWatcher) connections[transport.socket.handle]; - assert(watcher !is null); + /** + * If the transport couldn't send the data, the further sending should + * be handled by the event loop. + * + * Params: + * transport = Transport. + * exception = Exception thrown on sending. + * + * Returns: $(D_KEYWORD true) if the operation could be successfully + * completed or scheduled, $(D_KEYWORD false) otherwise (the + * transport will be destroyed then). + */ + protected bool feed(SelectorStreamTransport transport, SocketException exception = null) + { + while (transport.input.length && transport.writeReady) + { + try + { + ptrdiff_t sent = transport.socket.send(transport.input[]); + if (sent == 0) + { + transport.writeReady = false; + } + else + { + transport.input += sent; + } + } + catch (SocketException e) + { + exception = e; + transport.writeReady = false; + } + } + if (exception !is null) + { + auto watcher = cast(IOWatcher) connections[transport.socket.handle]; + assert(watcher !is null); - kill(watcher, exception); - return false; - } - return true; - } + kill(watcher, exception); + return false; + } + return true; + } - /** - * Start watching. - * - * Params: - * watcher = Watcher. - */ - override void start(ConnectionWatcher watcher) - { - if (watcher.active) - { - return; - } + /** + * Start watching. + * + * Params: + * watcher = Watcher. + */ + override void start(ConnectionWatcher watcher) + { + if (watcher.active) + { + return; + } - if (connections.length <= watcher.socket) - { - MmapPool.instance.resizeArray(connections, watcher.socket.handle + maxEvents / 2); - } - connections[watcher.socket.handle] = watcher; + if (connections.length <= watcher.socket) + { + MmapPool.instance.resizeArray(connections, watcher.socket.handle + maxEvents / 2); + } + connections[watcher.socket.handle] = watcher; - super.start(watcher); - } + super.start(watcher); + } - /** - * Accept incoming connections. - * - * Params: - * connection = Connection watcher ready to accept. - */ - package void acceptConnections(ConnectionWatcher connection) - in - { - assert(connection !is null); - } - body - { - while (true) - { - ConnectedSocket client; - try - { - client = (cast(StreamSocket) connection.socket).accept(); - } - catch (SocketException e) - { - theAllocator.dispose(e); - break; - } - if (client is null) - { - break; - } + /** + * Accept incoming connections. + * + * Params: + * connection = Connection watcher ready to accept. + */ + package void acceptConnections(ConnectionWatcher connection) + in + { + assert(connection !is null); + } + body + { + while (true) + { + ConnectedSocket client; + try + { + client = (cast(StreamSocket) connection.socket).accept(); + } + catch (SocketException e) + { + theAllocator.dispose(e); + break; + } + if (client is null) + { + break; + } - IOWatcher io; - auto transport = MmapPool.instance.make!SelectorStreamTransport(this, client); + IOWatcher io; + auto transport = MmapPool.instance.make!SelectorStreamTransport(this, client); - if (connections.length > client.handle) - { - io = cast(IOWatcher) connections[client.handle]; - } - else - { - MmapPool.instance.resizeArray(connections, client.handle + maxEvents / 2); - } - if (io is null) - { - io = MmapPool.instance.make!IOWatcher(transport, - connection.protocol); - connections[client.handle] = io; - } - else - { - io(transport, connection.protocol); - } + if (connections.length > client.handle) + { + io = cast(IOWatcher) connections[client.handle]; + } + else + { + MmapPool.instance.resizeArray(connections, client.handle + maxEvents / 2); + } + if (io is null) + { + io = MmapPool.instance.make!IOWatcher(transport, + connection.protocol); + connections[client.handle] = io; + } + else + { + io(transport, connection.protocol); + } - reify(io, EventMask(Event.none), EventMask(Event.read, Event.write)); - connection.incoming.insertBack(io); - } + reify(io, EventMask(Event.none), EventMask(Event.read, Event.write)); + connection.incoming.insertBack(io); + } - if (!connection.incoming.empty) - { - swapPendings.insertBack(connection); - } - } + if (!connection.incoming.empty) + { + swapPendings.insertBack(connection); + } + } } diff --git a/source/tanya/async/iocp.d b/source/tanya/async/iocp.d index 8817e0c..f915e36 100644 --- a/source/tanya/async/iocp.d +++ b/source/tanya/async/iocp.d @@ -21,12 +21,12 @@ import core.sys.windows.windef; */ class State { - /// For internal use by Windows API. - align(1) OVERLAPPED overlapped; + /// For internal use by Windows API. + align(1) OVERLAPPED overlapped; - /// File/socket handle. - HANDLE handle; + /// File/socket handle. + HANDLE handle; - /// For keeping events or event masks. - int event; + /// For keeping events or event masks. + int event; } diff --git a/source/tanya/async/loop.d b/source/tanya/async/loop.d index de1e401..15dfafd 100644 --- a/source/tanya/async/loop.d +++ b/source/tanya/async/loop.d @@ -10,100 +10,105 @@ * * --- * import tanya.async; + * import tanya.memory; * import tanya.network.socket; * * class EchoProtocol : TransmissionControlProtocol * { - * private DuplexTransport transport; + * private DuplexTransport transport; * - * void received(ubyte[] data) - * { - * transport.write(data); - * } + * void received(ubyte[] data) + * { + * transport.write(data); + * } * - * void connected(DuplexTransport transport) - * { - * this.transport = transport; - * } + * void connected(DuplexTransport transport) + * { + * this.transport = transport; + * } * - * void disconnected(SocketException e = null) - * { - * } + * void disconnected(SocketException e = null) + * { + * } * } * * 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) - * { - * auto sock = new OverlappedStreamSocket(AddressFamily.INET); - * } - * else - * { - * auto sock = new StreamSocket(AddressFamily.INET); - * sock.blocking = false; - * } + * version (Windows) + * { + * auto sock = theAllocator.make!OverlappedStreamSocket(AddressFamily.INET); + * } + * else + * { + * auto sock = theAllocator.make!StreamSocket(AddressFamily.INET); + * sock.blocking = false; + * } * - * sock.bind(address); - * sock.listen(5); + * sock.bind(address); + * sock.listen(5); * - * auto io = new ConnectionWatcher(sock); - * io.setProtocol!EchoProtocol; + * auto io = theAllocator.make!ConnectionWatcher(sock); + * io.setProtocol!EchoProtocol; * - * defaultLoop.start(io); - * defaultLoop.run(); + * defaultLoop.start(io); + * defaultLoop.run(); * - * sock.shutdown(); + * sock.shutdown(); + * theAllocator.dispose(io); + * theAllocator.dispose(sock); + * theAllocator.dispose(address); * } * --- */ 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 std.algorithm.iteration; import std.algorithm.mutation; 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) { } else version (linux) { - import tanya.async.event.epoll; - version = Epoll; + import tanya.async.event.epoll; + version = Epoll; } else version (Windows) { - import tanya.async.event.iocp; - version = IOCP; + import tanya.async.event.iocp; + version = IOCP; } else version (OSX) { - version = Kqueue; + version = Kqueue; } else version (iOS) { - version = Kqueue; + version = Kqueue; } else version (FreeBSD) { - version = Kqueue; + version = Kqueue; } else version (OpenBSD) { - version = Kqueue; + version = Kqueue; } else version (DragonFlyBSD) { - version = Kqueue; + version = Kqueue; } /** @@ -111,11 +116,11 @@ else version (DragonFlyBSD) */ enum Event : uint { - none = 0x00, /// No events. - read = 0x01, /// Non-blocking read call. - write = 0x02, /// Non-blocking write call. - accept = 0x04, /// Connection made. - error = 0x80000000, /// Sent when an error occurs. + none = 0x00, /// No events. + read = 0x01, /// Non-blocking read call. + write = 0x02, /// Non-blocking write call. + accept = 0x04, /// Connection made. + error = 0x80000000, /// Sent when an error occurs. } alias EventMask = BitFlags!Event; @@ -125,161 +130,170 @@ alias EventMask = BitFlags!Event; */ abstract class Loop { - /// Pending watchers. - protected PendingQueue!Watcher pendings; + /// Pending watchers. + protected Queue!Watcher pendings; - protected PendingQueue!Watcher swapPendings; + protected Queue!Watcher swapPendings; - /** - * Returns: Maximal event count can be got at a time - * (should be supported by the backend). - */ - protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc - { - return 128U; - } + /** + * Returns: Maximal event count can be got at a time + * (should be supported by the backend). + */ + protected @property inout(uint) maxEvents() inout const pure nothrow @safe @nogc + { + return 128U; + } - /** - * Initializes the loop. - */ - this() - { - pendings = MmapPool.instance.make!(PendingQueue!Watcher); - swapPendings = MmapPool.instance.make!(PendingQueue!Watcher); - } + /** + * Initializes the loop. + */ + this() + { + pendings = MmapPool.instance.make!(Queue!Watcher); + swapPendings = MmapPool.instance.make!(Queue!Watcher); + } - /** - * Frees loop internals. - */ - ~this() - { - MmapPool.instance.dispose(pendings); - MmapPool.instance.dispose(swapPendings); - } + /** + * Frees loop internals. + */ + ~this() + { + foreach (w; pendings) + { + MmapPool.instance.dispose(w); + } + MmapPool.instance.dispose(pendings); - /** - * Starts the loop. - */ - void run() - { - done_ = false; - do - { - poll(); + foreach (w; swapPendings) + { + MmapPool.instance.dispose(w); + } + MmapPool.instance.dispose(swapPendings); + } - // Invoke pendings - swapPendings.each!((ref p) => p.invoke()); + /** + * Starts the loop. + */ + void run() + { + done_ = false; + do + { + poll(); - swap(pendings, swapPendings); - } - while (!done_); - } + // Invoke pendings + swapPendings.each!((ref p) => p.invoke()); - /** - * Break out of the loop. - */ - void unloop() @safe pure nothrow - { - done_ = true; - } + swap(pendings, swapPendings); + } + while (!done_); + } - /** - * Start watching. - * - * Params: - * watcher = Watcher. - */ - void start(ConnectionWatcher watcher) - { - if (watcher.active) - { - return; - } - watcher.active = true; - reify(watcher, EventMask(Event.none), EventMask(Event.accept)); - } + /** + * Break out of the loop. + */ + void unloop() @safe pure nothrow + { + done_ = true; + } - /** - * Stop watching. - * - * Params: - * watcher = Watcher. - */ - void stop(ConnectionWatcher watcher) - { - if (!watcher.active) - { - return; - } - watcher.active = false; + /** + * Start watching. + * + * Params: + * watcher = Watcher. + */ + void start(ConnectionWatcher watcher) + { + if (watcher.active) + { + return; + } + 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; - /** - * 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); + reify(watcher, EventMask(Event.accept), EventMask(Event.none)); + } - /** - * Returns: The blocking time. - */ - protected @property inout(Duration) blockTime() - inout @safe pure nothrow - { - // Don't block if we have to do. - return swapPendings.empty ? blockTime_ : Duration.zero; - } + /** + * 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); - /** - * Sets the blocking time for IO watchers. - * - * Params: - * blockTime = The blocking time. Cannot be larger than - * $(D_PSYMBOL maxBlockTime). - */ - protected @property void blockTime(in Duration blockTime) @safe pure nothrow - in - { - assert(blockTime <= 1.dur!"hours", "Too long to wait."); - assert(!blockTime.isNegative); - } - body - { - blockTime_ = blockTime; - } + /** + * Returns: The blocking time. + */ + protected @property inout(Duration) blockTime() + inout @safe pure nothrow + { + // Don't block if we have to do. + return swapPendings.empty ? blockTime_ : Duration.zero; + } - /** - * Kills the watcher and closes the connection. - */ - 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); - } + /** + * Sets the blocking time for IO watchers. + * + * Params: + * blockTime = The blocking time. Cannot be larger than + * $(D_PSYMBOL maxBlockTime). + */ + protected @property void blockTime(in Duration blockTime) @safe pure nothrow + in + { + assert(blockTime <= 1.dur!"hours", "Too long to wait."); + assert(!blockTime.isNegative); + } + body + { + blockTime_ = blockTime; + } - /** - * Does the actual polling. - */ - abstract protected void poll(); + /** + * Kills the watcher and closes the connection. + */ + 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. - protected Duration blockTime_ = 1.dur!"minutes"; + /// Whether the event loop should be stopped. + private bool done_; + + /// Maximal block time. + protected Duration blockTime_ = 1.dur!"minutes"; } /** @@ -287,18 +301,17 @@ abstract class Loop */ class BadLoopException : Exception { -@nogc: - /** - * Params: - * file = The file where the exception occurred. - * line = The line number where the exception occurred. - * next = The previous exception in the chain of exceptions, if any. - */ - this(string file = __FILE__, size_t line = __LINE__, Throwable next = null) - pure @safe nothrow const - { - super("Event loop cannot be initialized.", file, line, next); - } + /** + * Params: + * file = The file where the exception occurred. + * line = The line number where the exception occurred. + * next = The previous exception in the chain of exceptions, if any. + */ + this(string file = __FILE__, size_t line = __LINE__, Throwable next = null) + pure nothrow const @safe @nogc + { + super("Event loop cannot be initialized.", file, line, next); + } } /** @@ -310,24 +323,24 @@ class BadLoopException : Exception */ @property Loop defaultLoop() { - if (defaultLoop_ !is null) - { - return defaultLoop_; - } - version (Epoll) - { - defaultLoop_ = MmapPool.instance.make!EpollLoop; - } - else version (IOCP) - { - defaultLoop_ = MmapPool.instance.make!IOCPLoop; - } - else version (Kqueue) - { - import tanya.async.event.kqueue; - defaultLoop_ = MmapPool.instance.make!KqueueLoop; - } - return defaultLoop_; + if (defaultLoop_ !is null) + { + return defaultLoop_; + } + version (Epoll) + { + defaultLoop_ = MmapPool.instance.make!EpollLoop; + } + else version (IOCP) + { + defaultLoop_ = MmapPool.instance.make!IOCPLoop; + } + else version (Kqueue) + { + import tanya.async.event.kqueue; + defaultLoop_ = MmapPool.instance.make!KqueueLoop; + } + return defaultLoop_; } /** @@ -339,145 +352,16 @@ class BadLoopException : Exception * your implementation to this property. * * Params: - * loop = The event loop. + * loop = The event loop. */ @property void defaultLoop(Loop loop) in { - assert(loop !is null); + assert(loop !is null); } body { - defaultLoop_ = loop; + defaultLoop_ = loop; } 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; -} diff --git a/source/tanya/async/protocol.d b/source/tanya/async/protocol.d index 62c0ab7..959937f 100644 --- a/source/tanya/async/protocol.d +++ b/source/tanya/async/protocol.d @@ -18,28 +18,28 @@ import tanya.async.transport; */ interface Protocol { - /** - * Params: - * data = Read data. - */ - void received(ubyte[] data); + /** + * Params: + * data = Read data. + */ + void received(ubyte[] data); - /** - * Called when a connection is made. - * - * Params: - * transport = Protocol transport. - */ - void connected(DuplexTransport transport); + /** + * Called when a connection is made. + * + * Params: + * transport = Protocol transport. + */ + void connected(DuplexTransport transport); - /** - * Called when a connection is lost. - * - * Params: - * exception = $(D_PSYMBOL Exception) if an error caused - * the disconnect, $(D_KEYWORD null) otherwise. - */ - void disconnected(SocketException exception = null); + /** + * Called when a connection is lost. + * + * Params: + * exception = $(D_PSYMBOL Exception) if an error caused + * the disconnect, $(D_KEYWORD null) otherwise. + */ + void disconnected(SocketException exception = null); } /** diff --git a/source/tanya/async/transport.d b/source/tanya/async/transport.d index 6156815..8191678 100644 --- a/source/tanya/async/transport.d +++ b/source/tanya/async/transport.d @@ -31,13 +31,13 @@ interface ReadTransport : Transport */ interface WriteTransport : Transport { - /** - * Write some data to the transport. - * - * Params: - * data = Data to send. - */ - void write(ubyte[] data); + /** + * Write some data to the transport. + * + * Params: + * data = Data to send. + */ + void write(ubyte[] data); } /** @@ -52,7 +52,7 @@ interface DuplexTransport : ReadTransport, WriteTransport */ interface SocketTransport : Transport { - @property inout(Socket) socket() inout pure nothrow @safe @nogc; + @property inout(Socket) socket() inout pure nothrow @safe @nogc; } /** diff --git a/source/tanya/async/watcher.d b/source/tanya/async/watcher.d index 260c6a3..9a9e441 100644 --- a/source/tanya/async/watcher.d +++ b/source/tanya/async/watcher.d @@ -10,23 +10,24 @@ */ module tanya.async.watcher; +import std.functional; +import std.exception; import tanya.async.loop; import tanya.async.protocol; import tanya.async.transport; import tanya.container.buffer; +import tanya.container.queue; import tanya.memory; import tanya.memory.mmappool; import tanya.network.socket; -import std.functional; -import std.exception; version (Windows) { - import core.sys.windows.basetyps; - import core.sys.windows.mswsock; - import core.sys.windows.winbase; - import core.sys.windows.windef; - import core.sys.windows.winsock2; + import core.sys.windows.basetyps; + import core.sys.windows.mswsock; + import core.sys.windows.winbase; + import core.sys.windows.windef; + import core.sys.windows.winsock2; } /** @@ -35,85 +36,89 @@ version (Windows) */ abstract class Watcher { - /// Whether the watcher is active. - bool active; + /// Whether the watcher is active. + bool active; - /** - * Invoke some action on event. - */ - void invoke(); + /** + * Invoke some action on event. + */ + void invoke(); } class ConnectionWatcher : Watcher { - /// Watched socket. - private Socket socket_; + /// Watched socket. + private Socket socket_; - /// Protocol factory. - protected Protocol delegate() protocolFactory; + /// Protocol factory. + protected Protocol delegate() protocolFactory; - package PendingQueue!IOWatcher incoming; + package Queue!IOWatcher incoming; - /** - * Params: - * socket = Socket. - */ - this(Socket socket) - { - socket_ = socket; - incoming = MmapPool.instance.make!(PendingQueue!IOWatcher); - } + /** + * Params: + * socket = Socket. + */ + this(Socket socket) + { + socket_ = socket; + incoming = MmapPool.instance.make!(Queue!IOWatcher); + } - /// Ditto. - protected this() - { - } + /// Ditto. + protected this() + { + } - ~this() - { - MmapPool.instance.dispose(incoming); - } + ~this() + { + foreach (w; incoming) + { + MmapPool.instance.dispose(w); + } + MmapPool.instance.dispose(incoming); + } - /* - * Params: - * P = Protocol should be used. - */ - void setProtocol(P : Protocol)() - { - this.protocolFactory = () => cast(Protocol) MmapPool.instance.make!P; - } + /* + * Params: + * P = Protocol should be used. + */ + void setProtocol(P : Protocol)() + { + this.protocolFactory = () => cast(Protocol) MmapPool.instance.make!P; + } - /** - * Returns: Socket. - */ - @property inout(Socket) socket() inout pure nothrow @nogc - { - return socket_; - } + /** + * Returns: Socket. + */ + @property inout(Socket) socket() inout pure nothrow @nogc + { + return socket_; + } - /** - * Returns: New protocol instance. - */ - @property Protocol protocol() - in - { - assert(protocolFactory !is null, "Protocol isn't set."); - } - body - { - return protocolFactory(); - } + /** + * Returns: New protocol instance. + */ + @property Protocol protocol() + in + { + assert(protocolFactory !is null, "Protocol isn't set."); + } + body + { + return protocolFactory(); + } - /** - * Invokes new connection callback. - */ - override void invoke() - { - foreach (io; incoming) - { - io.protocol.connected(cast(DuplexTransport) io.transport); - } - } + /** + * Invokes new connection callback. + */ + override void invoke() + { + foreach (io; incoming) + { + io.protocol.connected(cast(DuplexTransport) io.transport); + } + } } /** @@ -122,121 +127,121 @@ class ConnectionWatcher : Watcher */ class IOWatcher : ConnectionWatcher { - /// If an exception was thrown the transport should be already invalid. - private union - { - StreamTransport transport_; - SocketException exception_; - } + /// If an exception was thrown the transport should be already invalid. + private union + { + StreamTransport transport_; + SocketException exception_; + } - private Protocol protocol_; + private Protocol protocol_; - /** - * Returns: Underlying output buffer. - */ - package ReadBuffer output; + /** + * Returns: Underlying output buffer. + */ + package ReadBuffer output; - /** - * Params: - * transport = Transport. - * protocol = New instance of the application protocol. - */ - this(StreamTransport transport, Protocol protocol) - in - { - assert(transport !is null); - assert(protocol !is null); - } - body - { - super(); - transport_ = transport; - protocol_ = protocol; - output = MmapPool.instance.make!ReadBuffer(); - active = true; - } + /** + * Params: + * transport = Transport. + * protocol = New instance of the application protocol. + */ + this(StreamTransport transport, Protocol protocol) + in + { + assert(transport !is null); + assert(protocol !is null); + } + body + { + super(); + transport_ = transport; + protocol_ = protocol; + output = MmapPool.instance.make!ReadBuffer(); + active = true; + } - /** - * Destroys the watcher. - */ - protected ~this() - { - MmapPool.instance.dispose(output); - MmapPool.instance.dispose(protocol_); - } + /** + * Destroys the watcher. + */ + protected ~this() + { + MmapPool.instance.dispose(output); + MmapPool.instance.dispose(protocol_); + } - /** - * Assigns a transport. - * - * Params: - * transport = Transport. - * protocol = Application protocol. - * - * Returns: $(D_KEYWORD this). - */ - IOWatcher opCall(StreamTransport transport, Protocol protocol) pure nothrow @nogc - in - { - assert(transport !is null); - assert(protocol !is null); - } - body - { - transport_ = transport; - protocol_ = protocol; - active = true; - return this; - } + /** + * Assigns a transport. + * + * Params: + * transport = Transport. + * protocol = Application protocol. + * + * Returns: $(D_KEYWORD this). + */ + IOWatcher opCall(StreamTransport transport, Protocol protocol) pure nothrow @nogc + in + { + assert(transport !is null); + assert(protocol !is null); + } + body + { + transport_ = transport; + protocol_ = protocol; + active = true; + return this; + } - /** - * Returns: Transport used by this watcher. - */ - @property inout(StreamTransport) transport() inout pure nothrow @nogc - { - return transport_; - } + /** + * Returns: Transport used by this watcher. + */ + @property inout(StreamTransport) transport() inout pure nothrow @nogc + { + return transport_; + } - /** - * Sets an exception occurred during a read/write operation. - * - * Params: - * exception = Thrown exception. - */ - @property void exception(SocketException exception) pure nothrow @nogc - { - exception_ = exception; - } + /** + * Sets an exception occurred during a read/write operation. + * + * Params: + * exception = Thrown exception. + */ + @property void exception(SocketException exception) pure nothrow @nogc + { + exception_ = exception; + } - /** - * Returns: Application protocol. - */ - override @property Protocol protocol() pure nothrow @safe @nogc - { - return protocol_; - } + /** + * Returns: Application protocol. + */ + override @property Protocol protocol() pure nothrow @safe @nogc + { + return protocol_; + } - /** - * Returns: Socket. - */ - override @property inout(Socket) socket() inout pure nothrow @nogc - { - return transport.socket; - } + /** + * Returns: Socket. + */ + override @property inout(Socket) socket() inout pure nothrow @nogc + { + return transport.socket; + } - /** - * Invokes the watcher callback. - */ - override void invoke() - { - if (output.length) - { - protocol.received(output[0..$]); - output.clear(); - } - else - { - protocol.disconnected(exception_); - active = false; - } - } + /** + * Invokes the watcher callback. + */ + override void invoke() + { + if (output.length) + { + protocol.received(output[0..$]); + output.clear(); + } + else + { + protocol.disconnected(exception_); + active = false; + } + } } diff --git a/source/tanya/container/queue.d b/source/tanya/container/queue.d index ef09da2..cee163d 100644 --- a/source/tanya/container/queue.d +++ b/source/tanya/container/queue.d @@ -121,7 +121,7 @@ class Queue(T) /** * 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; } @@ -170,7 +170,8 @@ class Queue(T) } /** - * $(D_KEYWORD foreach) iteration. + * $(D_KEYWORD foreach) iteration. The elements will be automatically + * dequeued. * * Params: * dg = $(D_KEYWORD foreach) body.