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

Queue: Leave only enqueue/dequeue/empty/opApply

Browse Source
This commit is contained in:
Eugen Wissner
2017-01-11 18:24:50 +01:00
parent 291920b479
commit ab930657b6
7 changed files with 70 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -159,7 +159,7 @@ class EpollLoop : SelectorLoop
}
else if (io.output.length)
{
swapPendings.insertBack(io);
swapPendings.enqueue(io);
}
}
else if (events[i].events & EPOLLOUT)

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

@@ -218,11 +218,11 @@ class IOCPLoop : Loop
auto transport = MmapPool.instance.make!IOCPStreamTransport(socket);
auto io = MmapPool.instance.make!IOWatcher(transport, connection.protocol);
connection.incoming.insertBack(io);
connection.incoming.enqueue(io);
reify(io, EventMask(Event.none), EventMask(Event.read, Event.write));
swapPendings.insertBack(connection);
swapPendings.enqueue(connection);
listener.beginAccept(overlapped);
break;
case OverlappedSocketEvent.read:
@@ -264,7 +264,7 @@ class IOCPLoop : Loop
{
transport.socket.beginReceive(io.output[], overlapped);
}
swapPendings.insertBack(io);
swapPendings.enqueue(io);
}
break;
case OverlappedSocketEvent.write:

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

@@ -271,7 +271,7 @@ class KqueueLoop : SelectorLoop
}
else if (io.output.length)
{
swapPendings.insertBack(io);
swapPendings.enqueue(io);
}
}
else if (events[i].filter == EVFILT_WRITE)

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

@@ -248,12 +248,12 @@ abstract class SelectorLoop : Loop
}
reify(io, EventMask(Event.none), EventMask(Event.read, Event.write));
connection.incoming.insertBack(io);
connection.incoming.enqueue(io);
}
if (!connection.incoming.empty)
{
swapPendings.insertBack(connection);
swapPendings.enqueue(connection);
}
}
}

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

@@ -184,8 +184,10 @@ abstract class Loop
poll();
// Invoke pendings
swapPendings.each!((ref p) @nogc => p.invoke());
foreach (ref w; *swapPendings)
{
w.invoke();
}
swap(pendings, swapPendings);
}
while (!done_);
@@ -283,7 +285,7 @@ abstract class Loop
defaultAllocator.dispose(watcher.socket);
MmapPool.instance.dispose(watcher.transport);
watcher.exception = exception;
swapPendings.insertBack(watcher);
swapPendings.enqueue(watcher);
}
/**