Windowsデスクトップアプリとstd.concurrencyの連携

以前からやってみたいことが有りました。

Windowsのウィンドウプロシージャ回すメッセージループで、std.concurrencyのメッセージをウィンドウメッセージ同様に受信できないのか？

PeekMessageでポーリングする

std.concurrency.Schedulerが導入されてこれがスマートに実装できるようになったわけですが、導入以前のやり方は強引なものでした。
ウィンドウメッセージとstd.concurrencyのメッセージを両方ともポーリングしながら監視する方法です。
ウィンドウメッセージの受信方法は、大雑把に分けて2種類ほど有ります。
GetMessageでブロッキングで受信する方法とPeekMessageでノンブロッキングで受信する方法です。
そこらで紹介されている野良のWindowsAPI解説サイトなんかでよく見るのはGetMessageを使用した実装ですね。一般的にはWindowsのGUIソフトはウィンドウメッセージの処理以外にメインスレッドを使用することは少なく、解説文のサンプル程度ではこれで十分という理由からこれが利用されることが多いのではないかと思います。
一方、PeekMessageを使用した実装は、DirectXなんかの解説サイトをみるとよく載っています。
これは、ゲームなどで60FPS(毎秒60回の描写)を維持するために、ウィンドウメッセージ(WM_PAINT等)ではない方法で描写処理を行うためです。
std.concurrencyによるスレッドへのメッセージをウィンドウメッセージと同じように受信するには、PeekMessageを使用した実装方法をつかい、ウィンドウメッセージがない場合に std.concurrency.receiveTimeout(..., 0.seconds) など、タイムアウト0秒でスレッドへのメッセージの着信を確認するようにします。

while (1)
{
	if (PeekMessageW(&msg, null, 0, 0, PM_REMOVE))
	{
		// ウィンドウメッセージあり
		if (msg.message == WM_QUIT)
			return msg.wParam;
		TranslateMessage(&msg);
		DispatchMessageW(&msg);
	}
	else if (!receiveTimeout(0.seconds,
	(shared ConcurrencyMessage cmsg)
	{
		// スレッドへのメッセージあり
		// (独自定義のConcurrencyMessage型メッセージ)
	}
	(Variant var)
	{
		// スレッドへのメッセージあり
		// (その他の型)
	}))
	{
		// ウィンドウメッセージも
		// スレッドへのメッセージもなし
	}
}

動作確認とかはしていませんが、たぶんこんな感じで動くのではないかと思います。

PeekMessageの問題点

しかし、残念ながら先のコードではあまり上手いこと動いてくれません。見ての通りの休止なしの無限ループとなっていて、非常にCPU負荷率の高いプログラムになってしまうのです。
そのため、ウィンドウメッセージもスレッドへのメッセージもない場合に Thread.sleep(0.seconds); などして負荷100%の状態を避けたりしますが、あまり効率的な方法ではありません。
理想としては、ウィンドウメッセージか、スレッドにメッセージが届いた時だけ動いて、それ以外は休止(Sleep)している状態が良いですね。

メッセージを待つ方法

さて、次のウィンドウメッセージが来るまで休止する方法はいくつかあります。
やはりここでも一番有名なのは、GetMessageでしょう。この関数はあたらしいウィンドウメッセージを受信するまで制御を返さなくなり、その間スレッドは休止します(リファレンスにはスレッドの休止云々のことは書かれてないけど、たぶん休止します)。
GetMessageのように、次のメッセージを待つ方法としては、PeekMessageした後に、WaitMessageする方法も考えられます。これによってもGetMessage同様、つぎのウィンドウメッセージ受信まで制御を返さなくなります。
しかし、それらでは当初の目的を果たせません。なぜなら、ウィンドウメッセージは待ってくれますが、スレッドへのメッセージ(std.concurrencyのメッセージ受信)は残念ながら感知しないからです。
スレッドへのメッセージを待つ方法としては、WaitForSingleObjectなどが有名です。
この関数は、イベントやミューテックス、プロセスの変化を検知して制御を返すことが出来ますが、残念ながら先ほどとは逆で、ウィンドウメッセージには応答してくれません。
ウィンドウメッセージもスレッドのメッセージも検知してくれる関数は無いのでしょうか。あります。 MsgWaitForMultipleObjectsまたはMsgWaitForMultipleObjectsExという関数がそれです。
ちょっと遠回りしましたが、この関数によって、ウィンドウメッセージとスレッドへのメッセージを同時に監視し、片方でも受信したら先のメッセージループを動かすようにすればよいというわけです。

std.concurrencyのメッセージ検知

ところで、ここまでぼかしてきましたが、スレッドへのメッセージの受信を検知といっても、標準のstd.concurrencyからは、イベントやミューテックスなどのハンドラは(相当強引なことをしないと)得られません。つまり、そのままではMsgWaitForMultipleObjectsで監視する対象になれないということです。
ではどのようにしてスレッドへのメッセージを検出するのか。
ここに、std.concurrency.Schedulerを利用します。
std.concurrency.Schedulerは、これを継承した独自のスケジューラーを作成することで、スレッドへの通知をデコレーションすることが出来るようになります。
メッセージ受信通知にはイベントを利用するじゃろ？
これを

class EventScheduler: ThreadScheduler
{
private:
	import core.sync.mutex, core.sync.condition;
	HAHDLE _hEvent;
public:
	this()
	{
		_hEvent = CreateEventW(null, 0, 1, null);
	}
	~this() nothrow @nogc @trusted
	{
		CloseHandle(_hEvent);
	}
	
	override Condition newCondition( Mutex m ) nothrow
	{
		return new class Condition
		{
			this() nothrow { super(m); }
			override void notify()
			{
				super.notify();
				SetEvent(_hEvent);
			}
			override void notifyAll()
			{
				super.notifyAll();
				SetEvent(_hEvent);
			}
		};
	}
	
	HANDLE eventHandle() pure nothrow @safe @nogc @property
	{
		return _hEvent;
	}
}

こうして

// std.concurrencyで定義されている変数 scheduler へと代入
scheduler = new EventScheduler;

こうじゃ。

while (1)
{
	if (PeekMessageW(&msg, null, 0, 0, PM_REMOVE))
	{
		// ウィンドウメッセージあり
		if (msg.message == WM_QUIT)
			return msg.wParam;
		TranslateMessage(&msg);
		DispatchMessageW(&msg);
	}
	else if (!receiveTimeout(0.seconds,
	(shared ConcurrencyMessage cmsg)
	{
		// スレッドへのメッセージあり
		// (独自定義のConcurrencyMessage型メッセージ)
	}
	(Variant var)
	{
		// スレッドへのメッセージあり
		// (その他の型)
	}))
	{
		// ウィンドウメッセージも
		// スレッドへのメッセージもなし
		if (auto sch = cast(EventScheduler)scheduler)
		{
			auto hEv = sch.eventHandler;
			// とりあえずタイムアウト1秒とかで監視
			MsgWaitForMultipleObjects(1, &hEv, 0, 1000, QS_ALLINPUT);
			ResetEvent(hEv);
			continue;
		}
		else
		{
			// 別のスケジューラーに切り替わってしまっていたら
			// しょうがないのでスリープ挟んでタスクスイッチ。
			Thread.sleep(1.msecs);
		}
	}
}

これで、MsgWaitForMultipleObjectsで監視可能なハンドルを扱うことが出来、これを監視することでメッセージループ中にstd.concurrencyのメッセージ受信を検知することが出来るようになります。

効果

ウィンドウメッセージに加えて、std.concurrencyのGUIスレッドへのメッセージを受信できるようになるので、ウィンドウメッセージ以外の自由なメッセージをスレッド間でやり取りすることが出来るようになります。
たとえば別スレッドで読み込んだ画像ファイルをGUIスレッドに投げて、GUIスレッド側で受信して表示したり、別スレッドでCurlでダウンロードしたTwitterのデータを解析し、解析結果をGUIスレッドに投げて表示処理したり。
使い方は様々だと思います。

まとめ

• Windowsのメッセージループでstd.concurrencyのメッセージを検出するには、PeekMessage+MsgWaitForMultipleObjectsを使う
• MsgWaitForMultipleObjectsでstd.concurrencyのメッセージを扱うにはstd.concurrency.Schedulerを継承したクラスを作る
• なお、このページのサンプルコードは自動的に消滅する書いただけでコンパイルされるかどうかすら確認していないので、動かなくても泣かないこと(Windowsの生のプログラムはちょっと書いてみたってのが面倒すぎるんよね)