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

Replace class Queue with the struct Queue

Browse Source
This commit is contained in:
Eugen Wissner
2016-12-13 10:56:29 +01:00
parent 711855474c
commit ab9f96e0c7
10 changed files with 141 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -26,6 +26,13 @@ import core.sys.posix.unistd;
import core.time;
import std.algorithm.comparison;
extern (C) nothrow @nogc
{
int epoll_create1(int flags);
int epoll_ctl (int epfd, int op, int fd, epoll_event *event);
int epoll_wait (int epfd, epoll_event *events, int maxevents, int timeout);
}
class EpollLoop : SelectorLoop
{
protected int fd;
@ -34,7 +41,7 @@ class EpollLoop : SelectorLoop
/**
* Initializes the loop.
*/
this()
this() @nogc
{
if ((fd = epoll_create1(EPOLL_CLOEXEC)) < 0)
{
@ -47,7 +54,7 @@ class EpollLoop : SelectorLoop
/**
* Free loop internals.
*/
~this()
~this() @nogc
{
MmapPool.instance.dispose(events);
close(fd);
@ -63,7 +70,9 @@ class EpollLoop : SelectorLoop
*
* 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) @nogc
in
{
assert(watcher !is null);
@ -97,7 +106,7 @@ class EpollLoop : SelectorLoop
/**
* Does the actual polling.
*/
protected override void poll()
protected override void poll() @nogc
{
// Don't block
immutable timeout = cast(immutable int) blockTime.total!"msecs";

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

@ -37,7 +37,7 @@ class IOCPStreamTransport : StreamTransport
* Params:
* socket = Socket.
*/
this(OverlappedConnectedSocket socket)
this(OverlappedConnectedSocket socket) @nogc
in
{
assert(socket !is null);
@ -53,7 +53,8 @@ class IOCPStreamTransport : StreamTransport
MmapPool.instance.dispose(input);
}
@property inout(OverlappedConnectedSocket) socket() inout pure nothrow @safe @nogc
@property inout(OverlappedConnectedSocket) socket()
inout pure nothrow @safe @nogc
{
return socket_;
}
@ -64,7 +65,7 @@ class IOCPStreamTransport : StreamTransport
* Params:
* data = Data to send.
*/
void write(ubyte[] data)
void write(ubyte[] data) @nogc
{
immutable empty = input.length == 0;
input ~= data;
@ -85,8 +86,8 @@ class IOCPStreamTransport : StreamTransport
}
}
class IOCPLoop : Loop
{
class IOCPLoop : Loop
{
protected HANDLE completionPort;
protected OVERLAPPED overlap;
@ -94,7 +95,7 @@ class IOCPStreamTransport : StreamTransport
/**
* Initializes the loop.
*/
this()
this() @nogc
{
super();
@ -118,7 +119,7 @@ class IOCPStreamTransport : StreamTransport
*/
override protected bool reify(ConnectionWatcher watcher,
EventMask oldEvents,
EventMask events)
EventMask events) @nogc
{
SocketState overlapped;
if (!(oldEvents & Event.accept) && (events & Event.accept))
@ -185,7 +186,7 @@ class IOCPStreamTransport : StreamTransport
/**
* Does the actual polling.
*/
override protected void poll()
override protected void poll() @nogc
{
DWORD lpNumberOfBytes;
ULONG_PTR key;

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

@ -151,12 +151,13 @@ class KqueueLoop : SelectorLoop
* 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
override protected @property inout(uint) maxEvents()
inout const pure nothrow @safe @nogc
{
return cast(uint) events.length;
}
this()
this() @nogc
{
super();
@ -171,14 +172,14 @@ class KqueueLoop : SelectorLoop
/**
* Free loop internals.
*/
~this()
~this() @nogc
{
MmapPool.instance.dispose(events);
MmapPool.instance.dispose(changes);
close(fd);
}
private void set(socket_t socket, short filter, ushort flags)
private void set(socket_t socket, short filter, ushort flags) @nogc
{
if (changes.length <= changeCount)
{
@ -206,7 +207,7 @@ class KqueueLoop : SelectorLoop
*/
override protected bool reify(ConnectionWatcher watcher,
EventMask oldEvents,
EventMask events)
EventMask events) @nogc
{
if (events != oldEvents)
{
@ -233,7 +234,7 @@ class KqueueLoop : SelectorLoop
/**
* Does the actual polling.
*/
protected override void poll()
protected override void poll() @nogc
{
timespec ts;
blockTime.split!("seconds", "nsecs")(ts.tv_sec, ts.tv_nsec);
@ -316,7 +317,7 @@ class KqueueLoop : SelectorLoop
* Returns: The blocking time.
*/
override protected @property inout(Duration) blockTime()
inout @safe pure nothrow
inout @nogc @safe pure nothrow
{
return min(super.blockTime, 1.dur!"seconds");
}
@ -333,7 +334,8 @@ class KqueueLoop : SelectorLoop
* completed or scheduled, $(D_KEYWORD false) otherwise (the
* transport is be destroyed then).
*/
protected override bool feed(SelectorStreamTransport transport, SocketException exception = null)
protected override bool feed(SelectorStreamTransport transport,
SocketException exception = null) @nogc
{
if (!super.feed(transport, exception))
{

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

@ -42,7 +42,7 @@ class SelectorStreamTransport : StreamTransport
* loop = Event loop.
* socket = Socket.
*/
this(SelectorLoop loop, ConnectedSocket socket)
this(SelectorLoop loop, ConnectedSocket socket) @nogc
{
socket_ = socket;
this.loop = loop;
@ -52,7 +52,7 @@ class SelectorStreamTransport : StreamTransport
/**
* Close the transport and deallocate the data buffers.
*/
~this()
~this() @nogc
{
MmapPool.instance.dispose(input);
}
@ -71,7 +71,7 @@ class SelectorStreamTransport : StreamTransport
* Params:
* data = Data to send.
*/
void write(ubyte[] data)
void write(ubyte[] data) @nogc
{
if (!data.length)
{
@ -113,13 +113,13 @@ abstract class SelectorLoop : Loop
/// Pending connections.
protected ConnectionWatcher[] connections;
this()
this() @nogc
{
super();
connections = MmapPool.instance.makeArray!ConnectionWatcher(maxEvents);
}
~this()
~this() @nogc
{
foreach (ref connection; connections)
{
@ -147,7 +147,8 @@ abstract class SelectorLoop : Loop
* completed or scheduled, $(D_KEYWORD false) otherwise (the
* transport will be destroyed then).
*/
protected bool feed(SelectorStreamTransport transport, SocketException exception = null)
protected bool feed(SelectorStreamTransport transport,
SocketException exception = null) @nogc
{
while (transport.input.length && transport.writeReady)
{
@ -186,7 +187,7 @@ abstract class SelectorLoop : Loop
* Params:
* watcher = Watcher.
*/
override void start(ConnectionWatcher watcher)
override void start(ConnectionWatcher watcher) @nogc
{
if (watcher.active)
{
@ -208,7 +209,7 @@ abstract class SelectorLoop : Loop
* Params:
* connection = Connection watcher ready to accept.
*/
package void acceptConnections(ConnectionWatcher connection)
package void acceptConnections(ConnectionWatcher connection) @nogc
in
{
assert(connection !is null);