// F05 — loop wakeup & external handles, Win32 implementation
// (../../../features/f05-loop-wakeup.md). Extends the scaffold
// (../scaffold/app.d) in three directions:
//
// * Cross-thread wakeup, mechanism A: a worker thread (raw CreateThread —
// no druntime registration needed, the thread only touches user32 and
// QueryPerformanceCounter) posts WM_APP+1 to the window 10x/second for
// 30 s with PostMessageW; the wParam indexes a __gshared QPC-timestamp
// slot, so the WndProc can compute "wakeup latency_us=... mech=postmessage".
// * Cross-thread wakeup, mechanism B: the same worker also posts WM_APP+2
// with PostThreadMessageW to the UI thread id. Thread messages have
// msg.hwnd == null, so DispatchMessageW would drop them on the floor —
// they MUST be handled in the pump itself (and an hwnd-filtered
// GetMessage/PeekMessage never retrieves them: the filter probe below
// proves it). Latency is logged as mech=threadmessage.
// * External-handle waiting: the pump is not GetMessageW but
// MsgWaitForMultipleObjectsEx(1, &timer, INFINITE, QS_ALLINPUT,
// MWMO_INPUTAVAILABLE) over a CreateWaitableTimerW handle ticking at
// 7 Hz — Win32's answer to "add an arbitrary fd to the loop" is an ARRAY
// of kernel handles, capped at MAXIMUM_WAIT_OBJECTS-1 = 63. A start-up
// probe calls the wait with 64 handles to demonstrate the hard failure.
//
// Exit prints min/median/p99/max latency per mechanism plus the waitable
// timer's observed tick-interval distribution. WSI_AUTO_EXIT=1 destroys the
// window once the worker is done (bounded ~31 s run, exit 0).
//
// Only druntime's built-in core.sys.windows bindings — no third-party packages.
module (module) appapp;
import (package) corecore.(package) core.syssys.(package) core.sys.windowswindows.(module) core.sys.windows.windowsWindows API header module
Translated from MinGW API for MS-Windows 4.0
Source
core/sys/windows/windows.d
windows;
import instrument;
enum UINT (constant) _error_ app.WM_WAKEUP_POST = __error__WM_WAKEUP_POST = WM_APP + 1; // mech A: PostMessageW(hwnd, ...)
enum UINT (constant) _error_ app.WM_WAKEUP_THREAD = __error__WM_WAKEUP_THREAD = WM_APP + 2; // mech B: PostThreadMessageW(tid, ...)
enum UINT (constant) _error_ app.WM_WORKER_DONE = __error__WM_WORKER_DONE = WM_APP + 3;
enum (constant) int app.POSTS_PER_SECOND = 10POSTS_PER_SECOND = 10;
enum (constant) int app.RUN_SECONDS = 30RUN_SECONDS = 30;
enum (constant) int app.TOTAL_POSTS = 300TOTAL_POSTS = (constant) int app.POSTS_PER_SECOND = 10POSTS_PER_SECOND * (constant) int app.RUN_SECONDS = 30RUN_SECONDS; // 300 per mechanism
enum (constant) int app.TIMER_HZ = 7TIMER_HZ = 7; // waitable-timer tick rate
enum (constant) int app.TIMER_PERIOD_MS = 142TIMER_PERIOD_MS = 1000 / (constant) int app.TIMER_HZ = 7TIMER_HZ; // 142 ms (7.04 Hz nominal)
enum (constant) int app.FILTER_PROBE_SEQ = 150FILTER_PROBE_SEQ = 150; // mid-run hwnd-filter probe (see wndProc)
// ---------------------------------------------------------------------------
// Shared state. QPC timestamps cross the thread boundary through per-sequence
// slots indexed by the message's wParam, so a latency sample never races with
// the next post (the worker writes slot i strictly before posting seq i).
struct (struct) app.DemoDemo
{
HWND (field) _error_ app.Demo.hwndhwnd;
DWORD (field) _error_ app.Demo.uiThreadIduiThreadId;
HANDLE (field) _error_ app.Demo.timertimer; // auto-reset waitable timer, 7 Hz
HANDLE (field) _error_ app.Demo.workerworker; // CreateThread handle
long (field) long app.Demo.qpcFreqqpcFreq; // QueryPerformanceFrequency, counts/s
long (field) long app.Demo.lastTickQpclastTickQpc; // previous fd_tick, for interval stats
uint (field) uint app.Demo.tickCounttickCount;
bool (field) bool app.Demo.autoExitautoExit;
bool (field) bool app.Demo.workerDoneworkerDone;
bool (field) bool app.Demo.probeDoneprobeDone;
}
__gshared (struct) app.DemoDemo _error_ app.gg;
__gshared long[(constant) int app.TOTAL_POSTS = 300TOTAL_POSTS] (__gshared global) long[300] app.postStampQpcpostStampQpc; // written by worker, read by UI thread
__gshared long[(constant) int app.TOTAL_POSTS = 300TOTAL_POSTS] (__gshared global) long[300] app.threadStampQpcthreadStampQpc;
// Fixed-capacity sample sets (no allocation; the WndProc is nothrow).
struct (struct) app.SamplesSamples
{
long[(constant) int app.TOTAL_POSTS = 300TOTAL_POSTS] (field) long[300] app.Samples.bufbuf;
int (field) int app.Samples.nn;
void void app.Samples.add(long v) nothrow @nogcadd(long (parameter) long vv) nothrow @nogc
{
if ((field) int app.Samples.nn < (field) long[300] app.Samples.bufbuf.(constant) ulong long[300].length = 300LUlength)
(field) long[300] app.Samples.bufbuf[(field) int app.Samples.nn++] = (parameter) long vv;
}
}
__gshared (struct) app.SamplesSamples (__gshared global) app.Samples app.postLatpostLat, (__gshared global) app.Samples app.threadLatthreadLat, (__gshared global) app.Samples app.tickIntervalstickIntervals;
long long app.qpcNow() nothrow @nogcqpcNow() nothrow @nogc
{
LARGE_INTEGER (local variable) _error_ tt;
QueryPerformanceCounter(&t);
return t.QuadPart;
}
long long app.qpcToUs(long delta) nothrow @nogcqpcToUs(long (parameter) long deltadelta) nothrow @nogc
{
return (parameter) long deltadelta * 1_000_000 / g.(field) _error_ g.qpcFreqqpcFreq;
}
// ---------------------------------------------------------------------------
// Worker thread: a raw kernel thread (CreateThread, extern(Windows) entry).
// Every ~100 ms it stamps QPC and fires both mechanisms back to back; both
// are documented as callable from any thread targeting another thread's queue.
extern (Windows) DWORD app.workerMainworkerMain(LPVOID) nothrow
{
foreach ((parameter) ii; 0 .. TOTAL_POSTS)
{
Sleep(1000 / POSTS_PER_SECOND);
postStampQpc[i] = qpcNow();
if (!PostMessageW(g.g.hwndhwnd, WM_WAKEUP_POST, cast(WPARAM) i, 0))
logEvent("error what=PostMessageW seq=%d code=%lu", cast(int) i, GetLastError());
threadStampQpc[i] = qpcNow();
if (!PostThreadMessageW(g.g.uiThreadIduiThreadId, WM_WAKEUP_THREAD, cast(WPARAM) i, 0))
logEvent("error what=PostThreadMessageW seq=%d code=%lu", cast(int) i, GetLastError());
}
PostMessageW(g.g.hwndhwnd, WM_WORKER_DONE, 0, 0);
return 0;
}
// ---------------------------------------------------------------------------
// The 63-handle ceiling, demonstrated: MsgWaitForMultipleObjectsEx accepts at
// most MAXIMUM_WAIT_OBJECTS-1 = 63 handles (the message queue itself occupies
// the 64th slot). 64 handles fail hard with ERROR_INVALID_PARAMETER.
void void app.probeHandleLimit() nothrowprobeHandleLimit() nothrow
{
HANDLE[64] (local variable) _error_ evev;
foreach ((local variable) int ii; 0 .. 64)
ev[i] = CreateEventW(null, FALSE, FALSE, null);
SetLastError(0);
const (local variable) const(_error_) r64r64 = MsgWaitForMultipleObjectsEx(64, ev.ptr, 0, QS_ALLINPUT, 0);
logEvent("handle_limit_probe n=64 result=0x%08lx err=%lu", r64, GetLastError());
SetLastError(0);
const (local variable) const(_error_) r63r63 = MsgWaitForMultipleObjectsEx(63, ev.ptr, 0, QS_ALLINPUT, 0);
logEvent("handle_limit_probe n=63 result=0x%08lx err=%lu", r63, GetLastError());
foreach ((local variable) int ii; 0 .. 64)
CloseHandle(ev[i]);
}
// ---------------------------------------------------------------------------
// Stats: insertion sort (300 elements, exit path only) + percentile report.
void void app.sortSamples(ref app.Samples s) nothrow @nogcsortSamples(ref (struct) app.SamplesSamples (parameter) app.Samples ss) nothrow @nogc
{
foreach ((local variable) int ii; 1 .. (parameter) app.Samples ss.(field) int app.Samples.nn)
{
const (local variable) const(long) vv = (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(local variable) int ii];
int (local variable) int jj = (local variable) int ii - 1;
while ((local variable) int jj >= 0 && (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(local variable) int jj] > (local variable) const(long) vv)
{
(parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(local variable) int jj + 1] = (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(local variable) int jj];
(local variable) int jj--;
}
(parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(local variable) int jj + 1] = (local variable) const(long) vv;
}
}
void void app.reportStats(ref app.Samples s, const(char)* name) nothrowreportStats(ref (struct) app.SamplesSamples (parameter) app.Samples ss, const(char)* (parameter) const(char)* namename) nothrow
{
if ((parameter) app.Samples ss.(field) int app.Samples.nn == 0)
{
logEvent("latency_stats mech=%s n=0", name);
return;
}
void app.sortSamples(ref app.Samples s) nothrow @nogcsortSamples((parameter) app.Samples ss);
const (local variable) const(long) minmin = (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[0];
const (local variable) const(long) medianmedian = (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(parameter) app.Samples ss.(field) int app.Samples.nn / 2];
const (local variable) const(long) p99p99 = (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[((parameter) app.Samples ss.(field) int app.Samples.nn * 99) / 100];
const (local variable) const(long) maxmax = (parameter) app.Samples ss.(field) long[300] app.Samples.bufbuf[(parameter) app.Samples ss.(field) int app.Samples.nn - 1];
logEvent("latency_stats mech=%s n=%d min_us=%lld median_us=%lld p99_us=%lld max_us=%lld",
name, s.s.nn, min, median, p99, max);
}
// ---------------------------------------------------------------------------
// WndProc: WM_WAKEUP_POST arrives here through DispatchMessageW like any
// window message. WM_WAKEUP_THREAD never does — see the pump.
void void app.recordWakeup(ref app.Samples s, long stamp, const(char)* mech, ulong seq) nothrowrecordWakeup(ref (struct) app.SamplesSamples (parameter) app.Samples ss, long (parameter) long stampstamp, const(char)* (parameter) const(char)* mechmech, (alias) object.size_t = ulongsize_t (parameter) ulong seqseq) nothrow
{
const (local variable) const(long) latlat = long app.qpcToUs(long delta) nothrow @nogcqpcToUs(long app.qpcNow() nothrow @nogcqpcNow() - (parameter) long stampstamp);
(parameter) app.Samples ss.void app.Samples.add(long v) nothrow @nogcadd((local variable) const(long) latlat);
logEvent("wakeup latency_us=%lld mech=%s seq=%d", lat, mech, cast(int) seq);
}
extern (Windows) LRESULT app.wndProcwndProc(HWND (parameter) HWND hwndhwnd, UINT (parameter) UINT msgmsg, WPARAM wParam, LPARAM lParam) nothrow
{
switch (msg)
{
case WM_WAKEUP_POST:
recordWakeup(postLat, postStampQpc[wParam], "postmessage", wParam);
if (wParam == FILTER_PROBE_SEQ && !g.g.probeDoneprobeDone)
{
// The matching WM_WAKEUP_THREAD was posted right behind this
// message, so it is (almost certainly) sitting in the queue now:
// an hwnd-filtered peek cannot see it, a null-filtered one can.
// This is why thread messages are lost inside any modal loop that
// pumps with an hwnd filter (dialogs, menus, DefWindowProc's
// move/size loop): no hwnd matches a message that has none.
g.g.probeDoneprobeDone = true;
MSG _error_ probeprobe;
const _error_ filteredfiltered = PeekMessageW(&probe, hwnd,
WM_WAKEUP_THREAD, WM_WAKEUP_THREAD, PM_NOREMOVE);
const _error_ openopen = PeekMessageW(&probe, null,
WM_WAKEUP_THREAD, WM_WAKEUP_THREAD, PM_NOREMOVE);
logEvent("thread_msg_filter_probe hwnd_filtered=%d null_filtered=%d",
cast(int) filtered, cast(int) open);
}
return 0;
case WM_WORKER_DONE:
logEvent("worker_done posts=%d", TOTAL_POSTS);
g.g.workerDoneworkerDone = true;
if (g.g.autoExitautoExit)
DestroyWindow(hwnd);
return 0;
case WM_PAINT:
PAINTSTRUCT _error_ psps;
HDC _error_ hdchdc = BeginPaint(hwnd, &ps);
FillRect(hdc, &ps.rcPaint, cast(HBRUSH)(COLOR_WINDOW + 1));
EndPaint(hwnd, &ps);
return 0;
case WM_CLOSE:
logEvent("close_requested");
goto default;
case WM_DESTROY:
logEvent("msg name=WM_DESTROY");
PostQuitMessage(0);
return 0;
default:
return DefWindowProcW(hwnd, msg, wParam, lParam);
}
}
// ---------------------------------------------------------------------------
bool bool app.wantAutoExit() nothrowwantAutoExit() nothrow
{
WCHAR[8] (local variable) _error_ bufbuf;
const (local variable) const(_error_) nn = GetEnvironmentVariableW("WSI_AUTO_EXIT"w.ptr, buf.ptr, buf.buf.lengthlength);
return n >= 1 && n < buf.(field) _error_ buf.lengthlength && buf[0] == '1';
}
int int D main()main()
{
instrumentInit("f05_loop_wakeup_win32");
logEvent("init_start");
g.g.autoExitautoExit = wantAutoExit();
logEvent("mode auto_exit=%d", g.g.autoExitautoExit ? 1 : 0);
LARGE_INTEGER (local variable) _error_ freqfreq;
QueryPerformanceFrequency(&freq);
g.g.qpcFreqqpcFreq = freq.QuadPart;
logEvent("qpc_freq hz=%lld", g.g.qpcFreqqpcFreq);
g.g.uiThreadIduiThreadId = GetCurrentThreadId();
HINSTANCE (local variable) _error_ hInsthInst = GetModuleHandleW(null);
auto (local variable) wstring clsNameclsName = "wsi-f05-class"w;
WNDCLASSEXW (local variable) _error_ wcwc;
wc.cbSize = WNDCLASSEXW.sizeof;
wc.lpfnWndProc = &wndProc;
wc.hInstance = hInst;
wc.lpszClassName = clsName.ptr;
wc.hCursor = LoadCursorW(null, IDC_ARROW);
if (!RegisterClassExW(&wc))
{
logEvent("error what=RegisterClassExW code=%lu", GetLastError());
return 1;
}
g.g.hwndhwnd = CreateWindowExW(0, clsName.ptr, "wsi-f05-loop-wakeup"w.ptr,
WS_OVERLAPPEDWINDOW, CW_USEDEFAULT, CW_USEDEFAULT, 480, 320,
null, null, hInst, null);
if (g.(field) _error_ g.hwndhwnd is null)
{
logEvent("error what=CreateWindowExW code=%lu", GetLastError());
return 1;
}
logEvent("window_created");
ShowWindow(g.g.hwndhwnd, SW_SHOW);
UpdateWindow(g.g.hwndhwnd);
void app.probeHandleLimit() nothrowprobeHandleLimit();
// The "arbitrary fd": an auto-reset waitable timer at ~7 Hz. Auto-reset
// means a satisfied wait consumes the signal — no manual ResetEvent dance.
g.g.timertimer = CreateWaitableTimerW(null, FALSE, null);
LARGE_INTEGER (local variable) _error_ duedue;
due.QuadPart = -10_000L * TIMER_PERIOD_MS; // relative, 100 ns units
if (g.(field) _error_ g.timertimer is null || !SetWaitableTimer(g.g.timertimer, &due, TIMER_PERIOD_MS, null, null, FALSE))
{
logEvent("error what=SetWaitableTimer code=%lu", GetLastError());
return 1;
}
logEvent("step name=SetWaitableTimer period_ms=%d", TIMER_PERIOD_MS);
g.g.workerworker = CreateThread(null, 0, &workerMain, null, 0, null);
if (g.(field) _error_ g.workerworker is null)
{
logEvent("error what=CreateThread code=%lu", GetLastError());
return 1;
}
logEvent("step name=CreateThread rate_hz=%d duration_s=%d", POSTS_PER_SECOND, RUN_SECONDS);
// The pump: wait on {timer} + the message queue in one call. QS_ALLINPUT
// wakes for any queued message; MWMO_INPUTAVAILABLE closes the race where
// a message arrived between the drain below and re-entering the wait
// (without it, already-queued-but-already-seen input would not satisfy
// the wait and a wakeup could stall until the next timer tick).
int (local variable) int exitCodeexitCode = 0;
pump: while (true)
{
const (local variable) const(_error_) rr = MsgWaitForMultipleObjectsEx(1, &g.g.timertimer, INFINITE,
QS_ALLINPUT, MWMO_INPUTAVAILABLE);
if (r == WAIT_OBJECT_0) // the timer handle, not the queue
{
const (local variable) const(long) tt = long app.qpcNow() nothrow @nogcqpcNow();
++g.(field) _error_ g.tickCounttickCount;
logEvent("fd_tick t=%lld n=%u", nowUs(), g.g.tickCounttickCount);
if (g.(field) _error_ g.lastTickQpclastTickQpc != 0)
(__gshared global) app.Samples app.tickIntervalstickIntervals.tickIntervals.addadd(long app.qpcToUs(long delta) nothrow @nogcqpcToUs((local variable) const(long) tt - g.(field) _error_ g.lastTickQpclastTickQpc));
g.g.lastTickQpclastTickQpc = t;
}
else if (r == WAIT_FAILED)
{
logEvent("error what=MsgWaitForMultipleObjectsEx code=%lu", GetLastError());
(local variable) int exitCodeexitCode = 1;
break;
}
// r == WAIT_OBJECT_0 + 1: queue input. Drain it fully either way —
// a timer wake may coincide with pending messages.
MSG (local variable) _error_ msgmsg;
while (PeekMessageW(&msg, null, 0, 0, PM_REMOVE))
{
if (msg.message == WM_QUIT)
{
(local variable) int exitCodeexitCode = cast(int) msg.wParam;
break pump;
}
if (msg.(field) _error_ msg.hwndhwnd is null)
{
// A thread message: DispatchMessageW would silently drop it
// (no hwnd -> no WndProc). Handle it here, in the pump.
if (msg.message == WM_WAKEUP_THREAD)
void app.recordWakeup(ref app.Samples s, long stamp, const(char)* mech, ulong seq) nothrowrecordWakeup((__gshared global) app.Samples app.threadLatthreadLat, (__gshared global) long[300] app.threadStampQpcthreadStampQpc[msg.wParam],
"threadmessage", msg.wParam);
continue;
}
TranslateMessage(&msg);
DispatchMessageW(&msg);
}
}
WaitForSingleObject(g.g.workerworker, 5000);
CloseHandle(g.g.workerworker);
CancelWaitableTimer(g.g.timertimer);
CloseHandle(g.g.timertimer);
void app.reportStats(ref app.Samples s, const(char)* name) nothrowreportStats((__gshared global) app.Samples app.postLatpostLat, "postmessage");
void app.reportStats(ref app.Samples s, const(char)* name) nothrowreportStats((__gshared global) app.Samples app.threadLatthreadLat, "threadmessage");
void app.reportStats(ref app.Samples s, const(char)* name) nothrowreportStats((__gshared global) app.Samples app.tickIntervalstickIntervals, "handle_tick_interval");
logEvent("tick_total n=%u nominal_period_ms=%d", g.g.tickCounttickCount, TIMER_PERIOD_MS);
logEvent("exit code=%d", exitCode);
return (local variable) int exitCodeexitCode;
}