debian-packages/vibe-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Convert Waitable from struct to pure template to avoid heap closures.

Browse source 
Also fixes a case where ThreadLocalWaiter.emitSingle overlaps a call to .emit.
This commit is contained in:
Sönke Ludwig 2017-07-21 01:43:13 +02:00
parent e32d818873
commit 9fe9783443
4 changed files with 186 additions and 153 deletions
Download patch file Download diff file Expand all files Collapse all files

13
source/vibe/core/core.d
View file

@ -1010,14 +1010,17 @@ struct FileDescriptorEvent {
assert((which & m_trigger) == Trigger.read, "Waiting for write event not yet supported.");
Waitable!(IOCallback,
bool got_data;
alias readwaiter = Waitable!(IOCallback,
cb => eventDriver.sockets.waitForData(m_socket, cb),
cb => eventDriver.sockets.cancelRead(m_socket)
) readwaiter;
cb => eventDriver.sockets.cancelRead(m_socket),
(StreamSocketFD fd, IOStatus st, size_t nb) { got_data = st == IOStatus.ok; }
);
asyncAwaitAny!true(timeout, readwaiter);
asyncAwaitAny!(true, readwaiter)(timeout);
return !readwaiter.cancelled;
return got_data;
}
}

51
source/vibe/core/net.d
View file

@ -53,7 +53,7 @@ NetworkAddress resolveHost(string host, ushort address_family, bool use_dns = tr
enforce(use_dns, "Malformed IP address string.");
NetworkAddress res;
bool success = false;
Waitable!(DNSLookupCallback,
alias waitable = Waitable!(DNSLookupCallback,
cb => eventDriver.dns.lookupHost(host, cb),
(cb, id) => eventDriver.dns.cancelLookup(id),
(DNSLookupID, DNSStatus status, scope RefAddress[] addrs) {
@ -63,9 +63,9 @@ NetworkAddress resolveHost(string host, ushort address_family, bool use_dns = tr
success = true;
}
}
) waitable;
);
asyncAwaitAny!true(waitable);
asyncAwaitAny!(true, waitable);
enforce(success, "Failed to lookup host '"~host~"'.");
return res;
@ -542,22 +542,27 @@ mixin(tracer);
if (m_context.readBuffer.length > 0) return true;
auto mode = timeout <= 0.seconds ? IOMode.immediate : IOMode.once;
Waitable!(IOCallback,
bool cancelled;
IOStatus status;
size_t nbytes;
alias waiter = Waitable!(IOCallback,
cb => eventDriver.sockets.read(m_socket, m_context.readBuffer.peekDst(), mode, cb),
cb => eventDriver.sockets.cancelRead(m_socket)
) waiter;
(cb) { cancelled = true; eventDriver.sockets.cancelRead(m_socket); },
(sock, st, nb) { assert(sock == m_socket); status = st; nbytes = nb; }
);
asyncAwaitAny!true(timeout, waiter);
asyncAwaitAny!(true, waiter)(timeout);
if (waiter.cancelled) return false;
if (cancelled) return false;
logTrace("Socket %s, read %s bytes: %s", waiter.results[0], waiter.results[2], waiter.results[1]);
logTrace("Socket %s, read %s bytes: %s", m_socket, nbytes, status);
assert(m_context.readBuffer.length == 0);
m_context.readBuffer.putN(waiter.results[2]);
switch (waiter.results[1]) {
m_context.readBuffer.putN(nbytes);
switch (status) {
default:
logDebug("Error status when waiting for data: %s", waiter.results[1]);
logDebug("Error status when waiting for data: %s", status);
break;
case IOStatus.ok: break;
case IOStatus.wouldBlock: assert(mode == IOMode.immediate); break;
@ -837,20 +842,21 @@ struct UDPConnection {
IOStatus status;
size_t nbytes;
bool cancelled;
Waitable!(DatagramIOCallback,
alias waitable = Waitable!(DatagramIOCallback,
cb => eventDriver.sockets.send(m_socket, data, IOMode.once, peer_address ? addrc : null, cb),
cb => eventDriver.sockets.cancelSend(m_socket),
(cb) { cancelled = true; eventDriver.sockets.cancelSend(m_socket); },
(DatagramSocketFD, IOStatus status_, size_t nbytes_, scope RefAddress addr)
{
status = status_;
nbytes = nbytes_;
}
) waitable;
);
asyncAwaitAny!true(waitable);
asyncAwaitAny!(true, waitable);
enforce(!waitable.cancelled && status == IOStatus.ok, "Failed to send packet.");
enforce(!cancelled && status == IOStatus.ok, "Failed to send packet.");
enforce(nbytes == data.length, "Packet was only sent partially.");
}
@ -873,10 +879,11 @@ struct UDPConnection {
IOStatus status;
size_t nbytes;
bool cancelled;
Waitable!(DatagramIOCallback,
alias waitable = Waitable!(DatagramIOCallback,
cb => eventDriver.sockets.receive(m_socket, buf, IOMode.once, cb),
cb => eventDriver.sockets.cancelReceive(m_socket),
(cb) { cancelled = true; eventDriver.sockets.cancelReceive(m_socket); },
(DatagramSocketFD, IOStatus status_, size_t nbytes_, scope RefAddress addr)
{
status = status_;
@ -886,10 +893,10 @@ struct UDPConnection {
catch (Exception e) logWarn("Failed to store datagram source address: %s", e.msg);
}
}
) waitable;
);
asyncAwaitAny!true(timeout, waitable);
enforce(!waitable.cancelled, "Receive timeout.");
asyncAwaitAny!(true, waitable)(timeout);
enforce(!cancelled, "Receive timeout.");
enforce(status == IOStatus.ok, "Failed to receive packet.");
return buf[0 .. nbytes];
}

64
source/vibe/core/sync.d
View file

@ -1236,15 +1236,14 @@ private final class ThreadLocalWaiter(bool EVENT_TRIGGERED) {
private {
static struct TaskWaiter {
TaskWaiter* prev, next;
Task task;
void delegate() @safe nothrow notifier;
bool cancelled;
void wait(void delegate() @safe nothrow del) @safe nothrow {
assert(notifier is null, "Local waiter is used twice!");
notifier = del;
}
void cancel() @safe nothrow { cancelled = true; notifier = null; }
void cancel() @safe nothrow { notifier = null; }
void emit() @safe nothrow { auto n = notifier; notifier = null; n(); }
}
static if (EVENT_TRIGGERED) {
@ -1301,30 +1300,35 @@ private final class ThreadLocalWaiter(bool EVENT_TRIGGERED) {
target_timeout = now + timeout;
}
Waitable!(typeof(TaskWaiter.notifier),
cb => waiter.wait(cb),
cb => waiter.cancel(),
) waitable;
bool cancelled;
alias waitable = Waitable!(typeof(TaskWaiter.notifier),
(cb) { waiter.wait(cb); },
(cb) { cancelled = true; waiter.cancel(); },
() {}
);
alias ewaitable = Waitable!(EventCallback,
(cb) {
eventDriver.events.wait(evt, cb);
// check for exit condition *after* starting to wait for the event
// to avoid a race condition
if (exit_condition()) {
eventDriver.events.cancelWait(evt, cb);
cb(evt);
}
},
(cb) { eventDriver.events.cancelWait(evt, cb); },
(EventID) {}
);
if (evt != EventID.invalid) {
Waitable!(EventCallback,
(cb) {
eventDriver.events.wait(evt, cb);
// check for exit condition *after* starting to wait for the event
// to avoid a race condition
if (exit_condition()) {
eventDriver.events.cancelWait(evt, cb);
cb(evt);
}
},
cb => eventDriver.events.cancelWait(evt, cb)
) ewaitable;
asyncAwaitAny!interruptible(timeout, waitable, ewaitable);
asyncAwaitAny!(interruptible, waitable, ewaitable)(timeout);
} else {
asyncAwaitAny!interruptible(timeout, waitable);
asyncAwaitAny!(interruptible, waitable)(timeout);
}
if (waitable.cancelled) {
if (cancelled) {
assert(waiter.next !is null, "Cancelled waiter not in queue anymore!?");
return false;
} else {
@ -1363,6 +1367,19 @@ private final class ThreadLocalWaiter(bool EVENT_TRIGGERED) {
bool emitSingle()
@safe nothrow {
if (m_waiters.empty) return false;
TaskWaiter* pivot = () @trusted { return &m_emitPivot; } ();
if (pivot.next) { // another emit in progress?
// shift pivot to the right, so that the other emit call will process another waiter
if (pivot !is m_waiters.back) {
auto n = pivot.next;
m_waiters.remove(pivot);
m_waiters.insertAfter(pivot, n);
}
return true;
}
emitWaiter(m_waiters.front);
return true;
}
@ -1373,8 +1390,7 @@ private final class ThreadLocalWaiter(bool EVENT_TRIGGERED) {
if (w.notifier !is null) {
logTrace("notify task %s %s %s", cast(void*)w, () @trusted { return cast(void*)w.notifier.funcptr; } (), w.notifier.ptr);
w.notifier();
w.notifier = null;
w.emit();
} else logTrace("notify callback is null");
}
}