// F05 loop wakeup & external fds on Wayland (../../f05-loop-wakeup.md): can a
// second thread wake the native event loop, and can arbitrary fds join it?
//
// Wayland's answer is structural: the wl_display connection IS a file
// descriptor, and libwayland's documented thread-safe pump —
// wl_display_prepare_read / poll / wl_display_read_events /
// wl_display_dispatch_pending — is *designed* around the application owning
// the poll(2) call. So "integrate an external fd" is not a workaround on
// Wayland, it is the intended shape: put the wl fd, an eventfd, and a timerfd
// into ONE pollfd array (see pumpOnce, the deliverable of this demo).
//
// What Wayland does NOT have is a protocol-level user event (no XSendEvent /
// PostMessage analogue): a client cannot send itself a message through the
// compositor. The absence is the finding — cross-thread wakeup must be a
// client-owned fd. This demo uses eventfd(2):
//
// 1. A producer pthread writes the eventfd 10×/s for 30 s; each post's
// monotonic timestamp travels through a lock-free ring (the eventfd
// counter is a *sum*, so coalesced posts would corrupt an inline
// timestamp — the side buffer is the correct pattern). The main loop
// logs `wakeup latency_us=… mech=eventfd` per post.
// 2. A timerfd ticking at 7 Hz is the arbitrary-fd probe; its `fd_tick`
// lines interleave with the 60 Hz frame-callback redraws.
// 3. min/median/p99/max latency is printed at exit.
//
// Based on the scaffold (../scaffold/app.d, findings ../../scaffold.md);
// instrumentation contract: ./instrument.d. Headless-safe: no compositor →
// SKIP, exit 0. WSI_AUTO_EXIT is accepted for runner symmetry; the run is
// inherently bounded (producer stops after WSI_F05_WAKEUPS posts, default 300).
module (module) appapp;
import c; // ImportC: <wayland-client.h> + xdg-shell glue + eventfd/timerfd/poll + wsi_* wrappers
import instrument;
import (package) corecore.(module) core.atomicThe atomic module provides basic support for lock-free
concurrent programming.
Use the -preview=nosharedaccess compiler flag to detect
unsafe individual read or write operations on shared data.
Source
core/atomic.d
Examples
int y = 2;
shared int x = y; // OK
//x++; // read modify write error
x.atomicOp!"+="(1); // OK
//y = x; // read error with preview flag
y = x.atomicLoad(); // OK
assert(y == 3);
//x = 5; // write error with preview flag
x.atomicStore(5); // OK
assert(x.atomicLoad() == 5);
atomic : (alias template) app.atomicLoad = core.atomic.atomicLoad(MemoryOrder ms = MemoryOrder.seq, T)(auto ref return scope const T val) if (!is(T == shared(U), U) && !is(T == shared(inout(U)), U) && !is(T == shared(const(U)), U))Loads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad, (alias template) app.atomicStore = core.atomic.atomicStore(MemoryOrder ms = MemoryOrder.seq, T, V)(ref T val, V newval) if (!is(T == shared) && !is(V == shared))Writes 'newval' into 'val'. The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
atomicStore, (enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder;
// glibc 2.42's <pthread.h> is not ImportC-able (linux/types.h __int128), so
// the thread API comes from druntime's POSIX declarations instead (see c.c).
import (package) corecore.(package) core.syssys.(package) core.sys.posixposix.(module) core.sys.posix.pthreadD header file for POSIX.
pthread : (alias) app.pthread_create = int core.sys.posix.pthread.pthread_create(ulong*, scope const(core.sys.posix.sys.types.pthread_attr_t*), extern (C) void* function(void*), void*) nothrow @nogcpthread_create, (alias) app.pthread_join = int core.sys.posix.pthread.pthread_join(ulong, void**) nothrow @nogcpthread_join, pthread_t;
import (package) corecore.(package) core.stdcstdc.(module) core.stdc.stdioD header file for C99 <stdio.h>
pubs.opengroup.org/onlinepubs/009695399/basedefs/stdio.h.html, stdio.h
Source
core/stdc/stdio.d
stdio : (alias) app.printf = int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf;
import (package) corecore.(package) core.stdcstdc.(module) core.stdc.stdlibD header file for C99.
pubs.opengroup.org/onlinepubs/009695399/basedefs/stdlib.h.html, stdlib.h
Source
core/stdc/stdlib.d
stdlib : (alias) app.atoi = int core.stdc.stdlib.atoi(scope const(char*) nptr) nothrow @nogcatoi, (alias) app.getenv = char* core.stdc.stdlib.getenv(scope const(char*) name) nothrow @nogcgetenv, (alias) app.qsort = void core.stdc.stdlib.qsort(void* base, ulong nmemb, ulong size, extern (C) int function(const(void*), const(void*)) compar) nothrow @nogcqsort;
import (package) corecore.(package) core.stdcstdc.(module) core.stdc.stringD header file for C99.
pubs.opengroup.org/onlinepubs/009695399/basedefs/string.h.html, string.h
Source
core/stdc/string.d
string : (alias) app.strcmp = int core.stdc.string.strcmp(scope const(char*) s1, scope const(char*) s2) pure nothrow @nogcstrcmp;
// ----------------------------------------------------------------- tunables
enum int (constant) int app.defaultWidth = 640defaultWidth = 640;
enum int (constant) int app.defaultHeight = 480defaultHeight = 480;
enum int (constant) int app.wakeupHz = 10wakeupHz = 10; // producer posts 10×/s …
enum int (constant) int app.defaultWakeups = 300defaultWakeups = 300; // … for 30 s (override: WSI_F05_WAKEUPS)
enum long (constant) long app.timerTickNs = 142857142LtimerTickNs = 1_000_000_000L / 7; // the 7 Hz timerfd probe
enum int (constant) int app.pollTimeoutMs = 250pollTimeoutMs = 250; // backstop tick (poll is the real wait)
enum long (constant) long app.hardCapSlackUs = 10000000LhardCapSlackUs = 10_000_000; // grace beyond the nominal run length
enum int (constant) int app.maxSamples = 4096maxSamples = 4096;
// -------------------------------------------------------------------- state
/// One wl_shm-backed ARGB8888 buffer. `busy` is owned by the compositor
/// between wl_surface.commit and the wl_buffer.release event.
struct (struct) app.BufferOne wl_shm-backed ARGB8888 buffer. busy is owned by the compositor
between wl_surface.commit and the wl_buffer.release event.
Buffer
{
wl_buffer* (field) _error_ app.Buffer.handlehandle;
uint* (field) uint* app.Buffer.pixelspixels; // mmap'ed, shared with the compositor
(alias) object.size_t = ulongsize_t (field) ulong app.Buffer.byteSizebyteSize;
int (field) int app.Buffer.widthwidth, (field) int app.Buffer.heightheight;
bool (field) bool app.Buffer.busybusy;
}
__gshared
{
wl_display* _error_ app.g_displayg_display;
wl_registry* _error_ app.g_registryg_registry;
wl_compositor* _error_ app.g_compositorg_compositor;
wl_shm* _error_ app.g_shmg_shm;
wl_seat* _error_ app.g_seatg_seat;
xdg_wm_base* _error_ app.g_wmBaseg_wmBase;
wl_surface* _error_ app.g_surfaceg_surface;
xdg_surface* _error_ app.g_xdgSurfaceg_xdgSurface;
xdg_toplevel* _error_ app.g_toplevelg_toplevel;
wl_callback* _error_ app.g_frameCbg_frameCb; // at most one outstanding frame callback
(struct) app.BufferOne wl_shm-backed ARGB8888 buffer. busy is owned by the compositor
between wl_surface.commit and the wl_buffer.release event.
Buffer[2] _error_ app.g_buffersg_buffers;
int (__gshared global) int app.g_widthg_width = defaultWidth; // last *acked* size — buffers must match it
int (__gshared global) int app.g_heightg_height = defaultHeight;
int (__gshared global) int app.g_pendingWidthg_pendingWidth;
int (__gshared global) int app.g_pendingHeightg_pendingHeight;
bool (__gshared global) bool app.g_configuredg_configured;
bool (__gshared global) bool app.g_presentedg_presented;
bool (__gshared global) bool app.g_runningg_running = true;
int (__gshared global) int app.g_framesg_frames;
int (__gshared global) int app.g_commitsg_commits;
// The two external fds multiplexed with the wl_display fd.
int (__gshared global) int app.g_eventfdg_eventfd = -1; // cross-thread wakeup channel (mech=eventfd)
int (__gshared global) int app.g_timerfdg_timerfd = -1; // arbitrary-fd probe (mech=timerfd, 7 Hz)
// Producer → consumer timestamp ring. The eventfd's 8-byte counter is a
// SUM of the posted values, so coalesced posts (two writes before one
// read) would add two timestamps into garbage; the ring carries each
// timestamp intact and the counter only says how many to pop.
long[(constant) int app.maxSamples = 4096maxSamples] (__gshared global) long[4096] app.g_ringg_ring;
ulong (__gshared global) ulong app.g_ringWriteg_ringWrite; // producer-owned, release-published
ulong (__gshared global) ulong app.g_ringReadg_ringRead; // consumer-owned
bool (__gshared global) bool app.g_producerDoneg_producerDone; // release-published by the producer thread
int (__gshared global) int app.g_wakeupsWantedg_wakeupsWanted = defaultWakeups;
long[(constant) int app.maxSamples = 4096maxSamples] (__gshared global) long[4096] app.g_latenciesg_latencies; // consumed-wakeup latencies, µs
int (__gshared global) int app.g_latCountg_latCount;
int (__gshared global) int app.g_coalescedg_coalesced; // posts that arrived >1 per eventfd read
long (__gshared global) long app.g_fdTicksg_fdTicks; // timerfd expirations seen
}
// --------------------------------------------------------- producer thread
/// Second thread: every 100 ms, stamp "now", publish it in the ring, write
/// the eventfd. The write(2) is the only wakeup mechanism Wayland offers a
/// thread that is not the dispatcher — there is no protocol-level user event
/// to post. (libwayland itself is thread-safe, but events it queues are only
/// seen when the *dispatching* thread next wakes — which is this eventfd.)
extern (C) void* void* app.producerMain(void* __param_0) nothrow @nogcSecond thread: every 100 ms, stamp "now", publish it in the ring, write
the eventfd. The write(2) is the only wakeup mechanism Wayland offers a
thread that is not the dispatcher — there is no protocol-level user event
to post. (libwayland itself is thread-safe, but events it queues are only
seen when the dispatching thread next wakes — which is this eventfd.)
producerMain(void*) nothrow @nogc
{
foreach ((local variable) int ii; 0 .. (__gshared global) int app.g_wakeupsWantedg_wakeupsWanted)
{
timespec (local variable) _error_ tsts;
ts.tv_sec = 0;
ts.tv_nsec = 1_000_000_000 / wakeupHz;
nanosleep(&ts, null);
immutable (local variable) immutable(_error_) stampstamp = instrNowUs(); // same MonoTime epoch as the main loop
immutable (local variable) immutable(ulong) wiwi = ulong core.atomic.atomicLoad!(MemoryOrder.raw, ulong)(ref return scope shared(const(ulong)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad!((enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder.(enum value) core.atomic.MemoryOrder.raw = 0Not sequenced.
Corresponds to LLVM AtomicOrdering.Monotonic
and C++11/C11 memory_order_relaxed.
raw)(*cast(shared ulong*)&(__gshared global) ulong app.g_ringWriteg_ringWrite);
(__gshared global) long[4096] app.g_ringg_ring[cast((alias) object.size_t = ulongsize_t)((local variable) immutable(ulong) wiwi % (constant) int app.maxSamples = 4096maxSamples)] = stamp;
void core.atomic.atomicStore!(MemoryOrder.rel, ulong, ulong)(ref shared(ulong) val, ulong newval) pure nothrow @nogc @trustedWrites 'newval' into 'val'. The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
atomicStore!((enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder.(enum value) core.atomic.MemoryOrder.rel = 3Sink-load + sink-store barrier.
Corresponds to LLVM AtomicOrdering.Release
and C++11/C11 memory_order_release.
rel)(*cast(shared ulong*)&(__gshared global) ulong app.g_ringWriteg_ringWrite, (local variable) immutable(ulong) wiwi + 1);
ulong (local variable) ulong oneone = 1;
write(g_eventfd, &one, one.one.sizeofsizeof); // the doorbell: kicks poll()
}
void core.atomic.atomicStore!(MemoryOrder.rel, bool, bool)(ref shared(bool) val, bool newval) pure nothrow @nogc @trustedWrites 'newval' into 'val'. The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
atomicStore!((enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder.(enum value) core.atomic.MemoryOrder.rel = 3Sink-load + sink-store barrier.
Corresponds to LLVM AtomicOrdering.Release
and C++11/C11 memory_order_release.
rel)(*cast(shared bool*)&(__gshared global) bool app.g_producerDoneg_producerDone, true);
ulong (local variable) ulong oneone = 1;
write(g_eventfd, &one, one.one.sizeofsizeof); // final kick so the loop notices done
return null;
}
// ------------------------------------------------- external-fd drain hooks
/// eventfd readable: one read returns (and zeroes) the whole counter — the
/// number of posts since the last read. Latency is measured per post against
/// the timestamp it carried through the ring.
void void app.drainEventfd() nothrow @nogceventfd readable: one read returns (and zeroes) the whole counter — the
number of posts since the last read. Latency is measured per post against
the timestamp it carried through the ring.
drainEventfd() nothrow @nogc
{
ulong (local variable) ulong countcount;
if (read(g_eventfd, &count, count.count.sizeofsizeof) != (local variable) ulong countcount.sizeof)
return;
immutable (local variable) immutable(_error_) nownow = instrNowUs();
immutable (local variable) immutable(ulong) wiwi = ulong core.atomic.atomicLoad!(MemoryOrder.acq, ulong)(ref return scope shared(const(ulong)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad!((enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder.(enum value) core.atomic.MemoryOrder.acq = 2Hoist-load + hoist-store barrier.
Corresponds to LLVM AtomicOrdering.Acquire
and C++11/C11 memory_order_acquire.
acq)(*cast(shared ulong*)&(__gshared global) ulong app.g_ringWriteg_ringWrite);
int (local variable) int poppedpopped;
while ((__gshared global) ulong app.g_ringReadg_ringRead < (local variable) immutable(ulong) wiwi)
{
immutable (local variable) immutable(long) stampstamp = (__gshared global) long[4096] app.g_ringg_ring[cast((alias) object.size_t = ulongsize_t)((__gshared global) ulong app.g_ringReadg_ringRead % (constant) int app.maxSamples = 4096maxSamples)];
immutable (local variable) immutable(_error_) latlat = now - (local variable) immutable(long) stampstamp;
if ((__gshared global) int app.g_latCountg_latCount < (constant) int app.maxSamples = 4096maxSamples)
(__gshared global) long[4096] app.g_latenciesg_latencies[(__gshared global) int app.g_latCountg_latCount++] = lat;
instrEvent("wakeup", "latency_us=%lld mech=eventfd seq=%llu", lat, g_ringRead);
(__gshared global) ulong app.g_ringReadg_ringRead++;
(local variable) int poppedpopped++;
}
if ((local variable) int poppedpopped > 1)
{
(__gshared global) int app.g_coalescedg_coalesced += (local variable) int poppedpopped - 1;
instrEvent("wakeup_coalesced", "posts=%d eventfd_count=%llu", popped, count);
}
}
/// timerfd readable: the 8-byte read is the number of expirations since the
/// last read (>1 means the loop was late by a full period).
void void app.drainTimerfd() nothrow @nogctimerfd readable: the 8-byte read is the number of expirations since the
last read (>1 means the loop was late by a full period).
drainTimerfd() nothrow @nogc
{
ulong (local variable) ulong expirationsexpirations;
if (read(g_timerfd, &expirations, expirations.expirations.sizeofsizeof) != (local variable) ulong expirationsexpirations.sizeof)
return;
(__gshared global) long app.g_fdTicksg_fdTicks += (local variable) ulong expirationsexpirations;
instrEvent("fd_tick", "t=%lld mech=timerfd expirations=%llu n=%lld",
instrNowUs(), expirations, g_fdTicks);
}
// ------------------------------------------- the canonical multiplexing loop
/// THE deliverable: the wl_display fd multiplexed with arbitrary fds in one
/// poll(2), via libwayland's thread-safe read pattern.
///
/// wl_display_prepare_read's contract (wayland-client.h): it registers this
/// thread's intent to read the socket and FAILS (-1) while the default queue
/// still holds undispatched events — so step 1 loops dispatch_pending until
/// the queue is empty and the intent is registered. After it succeeds, this
/// thread must call exactly one of read_events (socket readable) or
/// cancel_read (woken for any other reason) — leaking the intent deadlocks
/// other readers. Requests are flushed *before* blocking so the compositor is
/// never left waiting on a half-sent message.
bool bool app.pumpOnce(int timeoutMs) nothrow @nogcTHE deliverable: the wl_display fd multiplexed with arbitrary fds in one
poll(2), via libwayland's thread-safe read pattern.
wl_display_prepare_read's contract (wayland-client.h): it registers this
thread's intent to read the socket and FAILS (-1) while the default queue
still holds undispatched events — so step 1 loops dispatch_pending until
the queue is empty and the intent is registered. After it succeeds, this
thread must call exactly one of read_events (socket readable) or
cancel_read (woken for any other reason) — leaking the intent deadlocks
other readers. Requests are flushed before blocking so the compositor is
never left waiting on a half-sent message.
pumpOnce(int (parameter) int timeoutMstimeoutMs) nothrow @nogc
{
// 1. Dispatch what is already queued; acquire the read intent.
while (wl_display_prepare_read(g_display) != 0)
if (wl_display_dispatch_pending(g_display) < 0)
return false;
// 2. Flush outgoing requests before sleeping.
wl_display_flush(g_display);
// 3. ONE poll over all event sources — the whole point of the pattern.
pollfd[3] (local variable) _error_ pfdspfds;
pfds[0].(__error)[0].fdfd = wl_display_get_fd(g_display);
pfds[0].events = POLLIN;
pfds[1].(__error)[1].fdfd = g_eventfd;
pfds[1].events = POLLIN;
pfds[2].(__error)[2].fdfd = g_timerfd;
pfds[2].events = POLLIN;
immutable (local variable) immutable(_error_) rr = poll(pfds.pfds.ptrptr, 3, timeoutMs);
if (r < 0)
{
wl_display_cancel_read(g_display);
return false;
}
// 4. Exactly one of read_events / cancel_read, per the contract.
if (pfds[0].revents & POLLIN)
{
if (wl_display_read_events(g_display) < 0)
return false; // read_events consumed the intent
if (wl_display_dispatch_pending(g_display) < 0)
return false;
}
else
wl_display_cancel_read(g_display);
// 5. The external fds, on the same wakeup.
if (pfds[1].revents & POLLIN)
void app.drainEventfd() nothrow @nogceventfd readable: one read returns (and zeroes) the whole counter — the
number of posts since the last read. Latency is measured per post against
the timestamp it carried through the ring.
drainEventfd();
if (pfds[2].revents & POLLIN)
void app.drainTimerfd() nothrow @nogctimerfd readable: the 8-byte read is the number of expirations since the
last read (>1 means the loop was late by a full period).
drainTimerfd();
return true;
}
// ---------------------------------------------------------- shm buffer pool
/// (Re)allocate `b` so it matches `w`×`h` (scaffold pattern).
bool app.ensureBuffer(Re)allocate b so it matches w×h (scaffold pattern).
ensureBuffer(ref (struct) app.BufferOne wl_shm-backed ARGB8888 buffer. busy is owned by the compositor
between wl_surface.commit and the wl_buffer.release event.
Buffer (parameter) Buffer bb, int (parameter) int ww, int (parameter) int hh) nothrow @nogc
{
if (b.b.handlehandle !is null && (b.b.widthwidth != w || b.b.heightheight != h))
{
wsi_buffer_destroy(b.b.handlehandle);
munmap(b.b.pixelspixels, b.b.byteSizebyteSize);
b = Buffer.init;
}
if (b.b.handlehandle !is null)
return true;
immutable _error_ stridestride = w * 4;
immutable _error_ sizesize = cast((unresolved type) size_tsize_t)(stride) * h;
immutable _error_ fdfd = memfd_create("wsi-f05", MFD_CLOEXEC);
if (fd < 0)
return false;
if (ftruncate(fd, cast(long) size) != 0)
{
close(fd);
return false;
}
void* _error_ memmem = mmap(null, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
if (mem is cast(void*)-1)
{
close(fd);
return false;
}
wl_shm_pool* _error_ poolpool = wsi_shm_create_pool(g_shm, fd, cast(int) size);
b.b.handlehandle = wsi_shm_pool_create_buffer(pool, 0, w, h, stride, WL_SHM_FORMAT_ARGB8888);
wsi_shm_pool_destroy(pool);
close(fd);
wsi_buffer_add_listener(b.b.handlehandle, &g_bufferListener, &b);
b.b.pixelspixels = cast(uint*) mem;
b.b.byteSizebyteSize = size;
b.b.widthwidth = w;
b.b.heightheight = h;
b.b.busybusy = false;
instrEvent("buffer_alloc", "size=%dx%d bytes=%zu", w, h, size);
return true;
}
/// Trivially cheap redraw (two alternating solid colors) — the window's only
/// job here is to keep real frame callbacks interleaving with the fd events.
void app.paintTrivially cheap redraw (two alternating solid colors) — the window's only
job here is to keep real frame callbacks interleaving with the fd events.
paint(ref (struct) app.BufferOne wl_shm-backed ARGB8888 buffer. busy is owned by the compositor
between wl_surface.commit and the wl_buffer.release event.
Buffer (parameter) Buffer bb, int frame) nothrow @nogc
{
immutable uint _error_ colorcolor = (frame & 1) ? 0xff20_6080 : 0xff80_4020;
immutable _error_ nn = cast((unresolved type) size_tsize_t) b.b.widthwidth * b.b.heightheight;
foreach ((parameter) ii; 0 .. n)
b.b.pixelspixels[i] = color;
}
/// Paint into a free buffer and commit it, requesting the next frame callback.
void void app.render() nothrow @nogcPaint into a free buffer and commit it, requesting the next frame callback.
render() nothrow @nogc
{
(struct) app.BufferOne wl_shm-backed ARGB8888 buffer. busy is owned by the compositor
between wl_surface.commit and the wl_buffer.release event.
Buffer* (local variable) _error_ bufbuf = null;
foreach (ref (parameter) bb; g_buffers)
if (!b.b.busybusy)
{
buf = &b;
break;
}
if (buf is null)
{
instrEvent("frame_skipped", "reason=all_buffers_busy");
return;
}
if (!ensureBuffer(*buf, g_width, g_height))
{
(__gshared global) bool app.g_runningg_running = false;
return;
}
paint(*buf, g_frames);
assert(buf.(field) _error_ buf.widthwidth == (__gshared global) int app.g_widthg_width && buf.(field) _error_ buf.heightheight == (__gshared global) int app.g_heightg_height,
"committed buffer size does not match the acked configure size");
wsi_surface_attach(g_surface, buf.buf.handlehandle, 0, 0);
wsi_surface_damage_buffer(g_surface, 0, 0, buf.buf.widthwidth, buf.buf.heightheight);
if (g_frameCb is null) // keep exactly one frame callback in flight
{
g_frameCb = wsi_surface_frame(g_surface);
wsi_callback_add_listener(g_frameCb, &g_frameListener, null);
}
wsi_surface_commit(g_surface);
buf.buf.busybusy = true;
(__gshared global) int app.g_commitsg_commits++;
if ((__gshared global) int app.g_commitsg_commits == 1)
instrEvent("first_commit", "size=%dx%d", buf.buf.widthwidth, buf.buf.heightheight);
}
// ---------------------------------------------------------------- listeners
// All callbacks are `extern (C) nothrow @nogc` to match the listener
// function-pointer types the ImportC `#pragma attribute` stamped in c.c.
extern (C) void app.onGlobalonGlobal(void* data, wl_registry* reg, uint name,
const(char)* iface, uint ver) nothrow @nogc
{
static uint uint capped(uint advertised, uint want) nothrow @nogccapped(uint (parameter) uint advertisedadvertised, uint (parameter) uint wantwant) nothrow @nogc
{
return advertised < want ? advertised : want;
}
if (strcmp(iface, wl_compositor_interface.name) == 0)
g_compositor = cast(wl_compositor*) wsi_registry_bind(reg, name,
&wl_compositor_interface, capped(ver, 4));
else if (strcmp(iface, wl_shm_interface.name) == 0)
g_shm = cast(wl_shm*) wsi_registry_bind(reg, name, &wl_shm_interface, 1);
else if (strcmp(iface, xdg_wm_base_interface.name) == 0)
g_wmBase = cast(xdg_wm_base*) wsi_registry_bind(reg, name, &xdg_wm_base_interface, 1);
else if (strcmp(iface, wl_seat_interface.name) == 0)
g_seat = cast(wl_seat*) wsi_registry_bind(reg, name, &wl_seat_interface,
capped(ver, 2));
}
extern (C) void app.onGlobalRemoveonGlobalRemove(void* data, wl_registry* reg, uint name) nothrow @nogc
{
}
extern (C) void app.onWmBasePingonWmBasePing(void* data, xdg_wm_base* (parameter) xdg_wm_base* bb, uint serial) nothrow @nogc
{
wsi_wm_base_pong(b, serial);
}
extern (C) void app.onSeatCapabilitiesonSeatCapabilities(void* data, wl_seat* s, uint caps) nothrow @nogc
{
}
extern (C) void app.onSeatNameonSeatName(void* data, wl_seat* s, const(char)* name) nothrow @nogc
{
}
extern (C) void app.onToplevelConfigureonToplevelConfigure(void* data, xdg_toplevel* t, int (parameter) int ww, int (parameter) int hh,
wl_array* states) nothrow @nogc
{
g_pendingWidth = w;
g_pendingHeight = h;
}
extern (C) void app.onXdgSurfaceConfigureonXdgSurfaceConfigure(void* data, xdg_surface* s, uint serial) nothrow @nogc
{
immutable _error_ ww = g_pendingWidth > 0 ? g_pendingWidth : defaultWidth;
immutable _error_ hh = g_pendingHeight > 0 ? g_pendingHeight : defaultHeight;
instrEvent("configure", "serial=%u size=%dx%d", serial, w, h);
wsi_xdg_surface_ack_configure(s, serial);
immutable _error_ resizedresized = w != g_width || h != g_height;
g_width = w;
g_height = h;
if (!g_configured)
{
g_configured = true;
instrFirstConfigure();
render();
}
else if (resized)
{
instrResize(w, h, 1);
render();
}
}
extern (C) void app.onToplevelCloseonToplevelClose(void* data, xdg_toplevel* t) nothrow @nogc
{
instrCloseRequested();
g_running = false;
}
extern (C) void app.onToplevelConfigureBoundsonToplevelConfigureBounds(void* data, xdg_toplevel* t, int (parameter) int ww, int (parameter) int hh) nothrow @nogc
{
}
extern (C) void app.onToplevelWmCapabilitiesonToplevelWmCapabilities(void* data, xdg_toplevel* t, wl_array* caps) nothrow @nogc
{
}
extern (C) void app.onBufferReleaseonBufferRelease(void* data, wl_buffer* (parameter) wl_buffer* bb) nothrow @nogc
{
auto _error_ bufbuf = cast((unresolved type) BufferBuffer*) data;
buf.buf.busybusy = false;
if (g_running && g_configured && g_frameCb is null)
render();
}
extern (C) void app.onFrameDoneonFrameDone(void* data, wl_callback* cb, uint timeMs) nothrow @nogc
{
wsi_callback_destroy(cb);
g_frameCb = null;
g_frames++;
// Log every 30th redraw so the fd_tick interleaving stays visible without
// 60 Hz noise drowning the trace.
if (g_frames % 30 == 0)
instrFrameCallback(timeMs);
if (!g_presented)
{
g_presented = true;
instrFirstPixelPresented();
}
render();
}
__gshared wl_registry_listener _error_ app.g_registryListenerg_registryListener = {&onGlobal, &onGlobalRemove};
__gshared xdg_wm_base_listener _error_ app.g_wmBaseListenerg_wmBaseListener = {&onWmBasePing};
__gshared wl_seat_listener _error_ app.g_seatListenerg_seatListener = {&onSeatCapabilities, &onSeatName};
__gshared xdg_surface_listener _error_ app.g_xdgSurfaceListenerg_xdgSurfaceListener = {&onXdgSurfaceConfigure};
__gshared xdg_toplevel_listener _error_ app.g_toplevelListenerg_toplevelListener = {
&onToplevelConfigure, &onToplevelClose,
&onToplevelConfigureBounds, &onToplevelWmCapabilities
};
__gshared wl_buffer_listener _error_ app.g_bufferListenerg_bufferListener = {&onBufferRelease};
__gshared wl_callback_listener _error_ app.g_frameListenerg_frameListener = {&onFrameDone};
// ------------------------------------------------------------------ stats
extern (C) int int app.cmpLong(const(void)* a, const(void)* b) nothrow @nogccmpLong(const(void)* (parameter) const(void)* aa, const(void)* (parameter) const(void)* bb) nothrow @nogc
{
immutable (local variable) immutable(long) xx = *cast(const(long)*) (parameter) const(void)* aa;
immutable (local variable) immutable(long) yy = *cast(const(long)*) (parameter) const(void)* bb;
return ((local variable) immutable(long) xx > (local variable) immutable(long) yy) - ((local variable) immutable(long) xx < (local variable) immutable(long) yy);
}
/// min/median/p99/max over the recorded wakeup latencies, to stdout.
void void app.printStats() nothrow @nogcmin/median/p99/max over the recorded wakeup latencies, to stdout.
printStats() nothrow @nogc
{
if ((__gshared global) int app.g_latCountg_latCount == 0)
{
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("stats: no wakeups recorded\n");
return;
}
void core.stdc.stdlib.qsort(void* base, ulong nmemb, ulong size, extern (C) int function(const(void*), const(void*)) compar) nothrow @nogcqsort((__gshared global) long[4096] app.g_latenciesg_latencies.(constant) long* long[4096].ptr = &g_latenciesptr, (__gshared global) int app.g_latCountg_latCount, long.(constant) ulong long.sizeof = 8LUsizeof, &int app.cmpLong(const(void)* a, const(void)* b) nothrow @nogccmpLong);
immutable (alias) object.size_t = ulongsize_t (local variable) immutable(ulong) p99Idxp99Idx = cast((alias) object.size_t = ulongsize_t)((__gshared global) int app.g_latCountg_latCount - 1) * 99 / 100;
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("stats mech=eventfd n=%d min_us=%lld median_us=%lld p99_us=%lld max_us=%lld coalesced=%d\n",
(__gshared global) int app.g_latCountg_latCount, (__gshared global) long[4096] app.g_latenciesg_latencies[0], (__gshared global) long[4096] app.g_latenciesg_latencies[(__gshared global) int app.g_latCountg_latCount / 2],
(__gshared global) long[4096] app.g_latenciesg_latencies[(local variable) immutable(ulong) p99Idxp99Idx], (__gshared global) long[4096] app.g_latenciesg_latencies[(__gshared global) int app.g_latCountg_latCount - 1], (__gshared global) int app.g_coalescedg_coalesced);
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("stats mech=timerfd ticks=%lld expected_hz=7\n", (__gshared global) long app.g_fdTicksg_fdTicks);
}
// ----------------------------------------------------------------- teardown
void void app.teardown() nothrow @nogcteardown() nothrow @nogc
{
foreach (ref (parameter) bb; g_buffers)
if (b.b.handlehandle !is null)
{
wsi_buffer_destroy(b.b.handlehandle);
munmap(b.b.pixelspixels, b.b.byteSizebyteSize);
b = Buffer.init;
}
if (g_frameCb !is null)
wsi_callback_destroy(g_frameCb);
if (g_toplevel !is null)
wsi_toplevel_destroy(g_toplevel);
if (g_xdgSurface !is null)
wsi_xdg_surface_destroy(g_xdgSurface);
if (g_surface !is null)
wsi_surface_destroy(g_surface);
if (g_wmBase !is null)
wsi_wm_base_destroy(g_wmBase);
if (g_seat !is null)
wl_proxy_destroy(cast(wl_proxy*) g_seat);
if (g_shm !is null)
wl_proxy_destroy(cast(wl_proxy*) g_shm);
if (g_compositor !is null)
wl_proxy_destroy(cast(wl_proxy*) g_compositor);
if (g_registry !is null)
wl_proxy_destroy(cast(wl_proxy*) g_registry);
wl_display_disconnect(g_display);
if ((__gshared global) int app.g_eventfdg_eventfd >= 0)
close(g_eventfd);
if ((__gshared global) int app.g_timerfdg_timerfd >= 0)
close(g_timerfd);
}
// --------------------------------------------------------------------- main
int int D main()main()
{
instrInit("f05-wayland");
cast(void) char* core.stdc.stdlib.getenv(scope const(char*) name) nothrow @nogcgetenv("WSI_AUTO_EXIT"); // accepted for runner symmetry; the run is bounded
if (const (local variable) const(char*) nn = char* core.stdc.stdlib.getenv(scope const(char*) name) nothrow @nogcgetenv("WSI_F05_WAKEUPS")) // shorter smoke runs
if (int core.stdc.stdlib.atoi(scope const(char*) nptr) nothrow @nogcatoi((local variable) const(char*) nn) > 0 && int core.stdc.stdlib.atoi(scope const(char*) nptr) nothrow @nogcatoi((local variable) const(char*) nn) <= (constant) int app.maxSamples = 4096maxSamples)
(__gshared global) int app.g_wakeupsWantedg_wakeupsWanted = int core.stdc.stdlib.atoi(scope const(char*) nptr) nothrow @nogcatoi((local variable) const(char*) nn);
// 1. Connect (SKIP cleanly on hosts without a compositor).
g_display = wl_display_connect(null);
if (g_display is null)
{
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("SKIP: no Wayland compositor (wl_display_connect returned null)\n");
return 0;
}
// 2. Registry: bind the globals a window needs.
g_registry = wsi_display_get_registry(g_display);
wsi_registry_add_listener(g_registry, &g_registryListener, null);
wl_display_roundtrip(g_display);
if (g_compositor is null || g_shm is null || g_wmBase is null)
{
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("SKIP: compositor lacks a required global (wl_compositor/wl_shm/xdg_wm_base)\n");
void app.teardown() nothrow @nogcteardown();
return 0;
}
wsi_wm_base_add_listener(g_wmBase, &g_wmBaseListener, null);
if (g_seat !is null)
wsi_seat_add_listener(g_seat, &g_seatListener, null);
// 3. Window object tree (scaffold handshake).
g_surface = wsi_compositor_create_surface(g_compositor);
g_xdgSurface = wsi_wm_base_get_xdg_surface(g_wmBase, g_surface);
wsi_xdg_surface_add_listener(g_xdgSurface, &g_xdgSurfaceListener, null);
g_toplevel = wsi_xdg_surface_get_toplevel(g_xdgSurface);
wsi_toplevel_add_listener(g_toplevel, &g_toplevelListener, null);
wsi_toplevel_set_title(g_toplevel, "wsi-f05-loop-wakeup");
wsi_toplevel_set_app_id(g_toplevel, "wsi-f05-loop-wakeup");
instrWindowCreated();
wsi_surface_commit(g_surface); // mandatory initial no-buffer commit
// 4. The two external fds. EFD_NONBLOCK/TFD_NONBLOCK: a spurious-looking
// drain must never block the loop.
(__gshared global) int app.g_eventfdg_eventfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
(__gshared global) int app.g_timerfdg_timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
if ((__gshared global) int app.g_eventfdg_eventfd < 0 || (__gshared global) int app.g_timerfdg_timerfd < 0)
{
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("SKIP: eventfd/timerfd unavailable\n");
void app.teardown() nothrow @nogcteardown();
return 0;
}
itimerspec (local variable) _error_ itsits;
its.it_value.tv_nsec = timerTickNs;
its.it_interval.tv_nsec = timerTickNs;
timerfd_settime(g_timerfd, 0, &its, null);
instrEvent("fd_setup", "eventfd=%d timerfd=%d timer_hz=7", g_eventfd, g_timerfd);
// 5. The producer thread — raw pthread_create (druntime's POSIX decls):
// no GC interaction to rule out, the demo is allocation-free after
// startup, and the thread only touches atomics + write(2).
pthread_t (local variable) ulong producerproducer;
if (int core.sys.posix.pthread.pthread_create(ulong*, scope const(core.sys.posix.sys.types.pthread_attr_t*), extern (C) void* function(void*), void*) nothrow @nogcpthread_create(&(local variable) ulong producerproducer, null, &void* app.producerMain(void* __param_0) nothrow @nogcSecond thread: every 100 ms, stamp "now", publish it in the ring, write
the eventfd. The write(2) is the only wakeup mechanism Wayland offers a
thread that is not the dispatcher — there is no protocol-level user event
to post. (libwayland itself is thread-safe, but events it queues are only
seen when the dispatching thread next wakes — which is this eventfd.)
producerMain, null) != 0)
{
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("SKIP: pthread_create failed\n");
void app.teardown() nothrow @nogcteardown();
return 0;
}
instrEvent("producer_started", "hz=%d posts=%d", wakeupHz, g_wakeupsWanted);
// 6. The multiplexing loop (see pumpOnce). Terminates when every posted
// wakeup has been consumed; hard wall-clock cap as a backstop.
immutable long (local variable) immutable(long) capUscapUs = (cast(long) (__gshared global) int app.g_wakeupsWantedg_wakeupsWanted * 1_000_000) / (constant) int app.wakeupHz = 10wakeupHz + (constant) long app.hardCapSlackUs = 10000000LhardCapSlackUs;
while ((__gshared global) bool app.g_runningg_running)
{
if (!bool app.pumpOnce(int timeoutMs) nothrow @nogcTHE deliverable: the wl_display fd multiplexed with arbitrary fds in one
poll(2), via libwayland's thread-safe read pattern.
wl_display_prepare_read's contract (wayland-client.h): it registers this
thread's intent to read the socket and FAILS (-1) while the default queue
still holds undispatched events — so step 1 loops dispatch_pending until
the queue is empty and the intent is registered. After it succeeds, this
thread must call exactly one of read_events (socket readable) or
cancel_read (woken for any other reason) — leaking the intent deadlocks
other readers. Requests are flushed before blocking so the compositor is
never left waiting on a half-sent message.
pumpOnce((constant) int app.pollTimeoutMs = 250pollTimeoutMs))
break;
if (bool core.atomic.atomicLoad!(MemoryOrder.acq, bool)(ref return scope shared(const(bool)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad!((enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder.(enum value) core.atomic.MemoryOrder.acq = 2Hoist-load + hoist-store barrier.
Corresponds to LLVM AtomicOrdering.Acquire
and C++11/C11 memory_order_acquire.
acq)(*cast(shared bool*)&(__gshared global) bool app.g_producerDoneg_producerDone)
&& (__gshared global) ulong app.g_ringReadg_ringRead >= ulong core.atomic.atomicLoad!(MemoryOrder.acq, ulong)(ref return scope shared(const(ulong)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad!((enum) core.atomic.MemoryOrderSpecifies the memory ordering semantics of an atomic operation.
MemoryOrder.(enum value) core.atomic.MemoryOrder.acq = 2Hoist-load + hoist-store barrier.
Corresponds to LLVM AtomicOrdering.Acquire
and C++11/C11 memory_order_acquire.
acq)(*cast(shared ulong*)&(__gshared global) ulong app.g_ringWriteg_ringWrite))
(__gshared global) bool app.g_runningg_running = false;
if (instrNowUs() > (local variable) immutable(long) capUscapUs)
{
instrEvent("hard_cap_hit");
(__gshared global) bool app.g_runningg_running = false;
}
}
int core.sys.posix.pthread.pthread_join(ulong, void**) nothrow @nogcpthread_join((local variable) ulong producerproducer, null);
// 7. Teardown + the numbers.
void app.teardown() nothrow @nogcteardown();
void app.printStats() nothrow @nogcmin/median/p99/max over the recorded wakeup latencies, to stdout.
printStats();
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf("ok: %d wakeups consumed, %lld fd_ticks, %d frame callbacks, %d commits\n",
(__gshared global) int app.g_latCountg_latCount, (__gshared global) long app.g_fdTicksg_fdTicks, (__gshared global) int app.g_framesg_frames, (__gshared global) int app.g_commitsg_commits);
return (__gshared global) int app.g_latCountg_latCount == (__gshared global) int app.g_wakeupsWantedg_wakeupsWanted ? 0 : 1;
}