std.concurrency 応用編

std.concurrencyを使うことでメッセージパッシングスタイルの平行処理が出来るようになることは11日目の @repeatedly さんの記事によって示されていますが、今回はその応用。なので11日目を読んでいなければ先にそちらを読むことをおすすめします。
今回はstd.concurrencyの応用として、spawnの引数に渡せるものや、sendの引数に渡せるもの==receive/receiveOnlyで受け取れるものに関して、詳しく語っていきます。
今回の記事は、以下の進行に従って説明します。

スレッドを超えて安全に取り扱われる型
spawnの引数に指定することのできるもの
send/receive できるもの

スレッドを超えて安全に取り扱われる型

スレッドを超えて安全に取り扱われる型とは、つまり、

!hasUnsharedAliasing!T

な、 T のことです。
dlang.org:hasUnsharedAliasing

template hasUnsharedAliasing(T...)

各Tが以下のいずれかの条件をみたす場合にtrueを返します：

生のポインタU*を含み、Uはimmutableでもsharedでもない。

配列U[]を含み、Uはimmutableでもsharedでもない。

classの参照型Cを含み、Cはimmutableでもsharedでもない。(訳注: たぶんinterfaceもこれと同じ)

連想配列型を含み、それがimmutableでもsharedでもない。

delegateを含み、それがsharedではない。(訳注:たぶんimmutableが抜けてる)

(原文):
Returns true if and only if T's representation includes at least one of the following:

a raw pointer U* and U is not immutable or shared;

an array U[] and U is not immutable or shared;

a reference to a class type C and C is not immutable or shared.

an associative array that is not immutable or shared.

a delegate that is not shared.

実はこのテンプレートはまさにこの目的のため、すなわち、ある型Tによって形成されるデータが、スレッドのコンテキスト間を渡って取り扱うことができるかどうかを判断することを主な目的としたテンプレートです。
裏返せば !hasUnsharedAliasing!T であるということは、Tという型に拠るデータからたどることのできるすべての参照・ポインタ・デリゲートが shared/immutable/const shared であること、要するにスレッドを超えて安全に取り扱われる型であることを保証します。

具体的に !hasUnsharedAliasing!T を満たすのはどのような型か、ほんの一部ですが以下の例を御覧ください。

Phobosのstd.traits.hasUnsharedAliasingのunittestより一部抜粋

unittest
{
    struct S2 { string a; }
    struct S3 { int a; immutable Object b; }
    static assert(!hasUnsharedAliasing!S2);
    static assert(!hasUnsharedAliasing!S3);

    struct S4 { int a; shared Object b; }
    struct S6 { shared char[] b; }
    struct S7 { float[3] vals; }
    static assert(!hasUnsharedAliasing!S4);
    static assert(!hasUnsharedAliasing!S6);
    static assert(!hasUnsharedAliasing!S7);

    struct S8 { int a; Rebindable!(immutable Object) b; }
    static assert(!hasUnsharedAliasing!S8);

    static assert(!hasUnsharedAliasing!(void delegate() immutable));
    static assert(!hasUnsharedAliasing!(void delegate() shared));
    static assert(!hasUnsharedAliasing!(void delegate() shared const));
    static assert(!hasUnsharedAliasing!(const(void delegate() immutable)));
    static assert(!hasUnsharedAliasing!(immutable(void delegate())));
    static assert(!hasUnsharedAliasing!(shared(void delegate())));
    static assert(!hasUnsharedAliasing!(shared(const(void delegate()))));
    static assert(!hasUnsharedAliasing!(void function()));

    static assert(!hasUnsharedAliasing!(Rebindable!(immutable Object)));
    static assert(!hasUnsharedAliasing!(Rebindable!(shared Object)));

    static assert(!hasUnsharedAliasing!(int, shared(int)*));
    static assert(!hasUnsharedAliasing!(shared(int)*, Rebindable!(shared Object)));
    static assert(!hasUnsharedAliasing!());

    class S13
    {
        void delegate() shared a;
        void delegate() immutable b;
        shared(const(void delegate() shared const)) o;
    }
    static assert(!hasUnsharedAliasing!(immutable(S13)));
    static assert(!hasUnsharedAliasing!(shared(S13)));

    // などなど
}

ここでは!hasUnsharedAliasing!Tなものに限りましたが、Phobosのソースコードを読むとhasUnsharedAliasing!Tな例も見ることができます。

`spawn`の引数に指定することのできるもの

spawn 関数によって新しいスレッドのコンテキストとなれるものは、以下の条件をすべて満たすものとなります。

spawnの第一引数のFはisCallable!F。呼び出し可能な型であること。以下のいずれかを満たす。
- function(関数ポインタ)
- delegate (デリゲート)
- opCallを実装したクラスオブジェクト/構造体(関数オブジェクト)
戻り値がvoid
spawn に指定した第二引数以降の引数T...はすべて、暗黙にFの引数に指定可能
Fは関数ポインタか、!hasUnsharedAliasing!F
引数T...は!hasUnsharedAliasing!TまたはTid。スレッドを超えて安全に取り扱われる型であること。以下のいずれかを満たす。
- immutable/sharedなターゲットを指定するポインタ
- immutable/sharedな要素を持つ配列
- immutable/sharedなclass/interface
- immutable/sharedな連想配列
- immutable/sharedなdelegate
- 参照を含まないデータ
- Tid

ここでも出てくるのが !hasUnsharedAliasing!F です。要するに、スレッドを超えて安全に取り扱われる関数呼び出し可能な型、例えばvoid delegate() sharedみたいな型である必要があります。
void delegate() sharedの作り方の詳細については 10日目とか御覧ください。
さらに、コンテキストに指定するFの引数T...についても、!hasUnsharedAliasing!Tでなければならないという条件があります。

要するに、コンテキストはspawnを呼び出したスレッドとは異なるスレッドで実行されるのだから、コンテキストそのものも、コンテキストに指定した関数の引数も、異なるスレッドで扱って問題ないデータでなければなりませんよ、ということです。

以下、例をご覧ください。

例1 11日目 @repeatedly さんの記事より抜粋

import std.stdio, std.concurrency;

void func(string message)
{
	writeln(message);
}

void main()
{
	spawn(&func, "Hello!");
}

上記コードでは、spawnに関数ポインタを指定しており第1と第4の条件を満たします。また、funcの戻り値はvoidであり、第2の条件を満たします。さらに、引数の型はfuncの仮引数、spawnの第2実引数ともにstringで一致しており、第3の条件も問題なく満たしています。なお、引数の型stringは、immutable(char)[]と等価、つまりhasUnsharedAliasingの第2条件に一致しないため、 !hasUnsharedAliasing!(string) == true となり、第5条件を満たします。
以下の表にまとめました。

条件番号	条件	判定対象	可否
1	`isCallable!F`を満たす	&func	○ function
2	戻り値が`void`	funcの戻り値型	○ void
3	`T...`は`F`の引数に暗黙変換可能	stringとstring	○ 完全一致
4	Fは関数ポインタか`!hasUnsharedAliasing!F`	&func	○ function
5	`!hasUnsharedAliasing!T`を満たすかTid	string	○`immutable`な要素を持つ配列

例2

import std.stdio, std.concurrency;

class Callable
{
	int a;
	this(int x) shared
	{
		a = x;
	}
	void opCall(shared Callable callable, const shared int* x) shared
	{
		writefln("this.a=%s, callable.a=%s, arg x=%s", a, callable.a, *x);
	}
}

class DerivedCallable: Callable
{
	this(int x) shared
	{
		super(x*x);
	}
}

void main()
{
	auto callable1 = new shared Callable(1);
	auto callable2 = new shared DerivedCallable(2);
	shared int* a = new shared int;
	*a = 100;
	spawn(callable1, callable2, a);
}

条件を満たせば、上記コードのようなことも可能。
各条件を満たす理由については以下の表にまとめました。

条件番号	条件	判定対象	可否
1	`isCallable!F`を満たす	callable1	○ opCallを実装したクラスオブジェクト
2	戻り値が`void`	Callable.opCall	○ void
3-1	`T...`は`F`の引数に暗黙変換可能	shared Callableとshared Callable	○ 完全一致
3-2	`T...`は`F`の引数に暗黙変換可能	shared Callableとshared DerivedCallable	○ 暗黙変換(アップキャスト)可能
3-3	`T...`は`F`の引数に暗黙変換可能	const shared int* とshared int*	○ 暗黙変換(shared型からconst shared型への変換)可能
4	Fは関数ポインタか`!hasUnsharedAliasing!F`	callable1	○ `shared`な`class`なので`!hasUnsharedAliasing!F`を満たす
5-1	`!hasUnsharedAliasing!T`を満たすかTid	shared Callable	○ `shared`な`class`なので`!hasUnsharedAliasing!T`を満たす
5-2	`!hasUnsharedAliasing!T`を満たすかTid	shared Callable	○ `shared`な`class`なので`!hasUnsharedAliasing!T`を満たす
5-3	`!hasUnsharedAliasing!T`を満たすかTid	const shared int*	○ `shared`なターゲットを指定するポインタなので`!hasUnsharedAliasing!T`を満たす

`send`/`receive`できるもの

spawnの引数が !hasUnsharedAliasing!T でなければならないのと同様に、コンテキストが異なるスレッドで実行されている以上、そこに送付するデータについても、!hasUnsharedAliasing!T でなければなりません。

ここまでならspawnと同じなので簡単なのですが、実はもう一つ大きめの制限があります。

それは、実引数に指定するデータが、creal・文字列(char[])・デリゲートのうちビット長が最大の型のビット長を超えてはならない、というものです。
要するに、基本型、配列、クラスオブジェクト、デリゲート、ポインタくらいしか指定できず、構造体変数はたかだかポインタと同等サイズ程度のものしか指定できないと思ったほうがいいです。

この条件は、send/receiveの内部で、Variant型が使用されていることに由来します。Variantに詰め込めるデータの条件が、まさに上記で述べた内容そのままなのです。
sendに関して、受け渡しの際にVariantが使用される関係上、Variantが許容するサイズ以下のデータしか引き渡すことができません。
(正直これはA先生も意図していない挙動な気がしますが、現状そのように実装されているのでそれが仕様なのでしょう…。)
(私はVariantに制限がある事自体なんかダメなんじゃないかと思います)

もしも受渡したいデータのサイズが大い場合は、newしたうえで、ポインタによる受け渡しを行うと良いでしょう。
なお、ポインタや配列などを受け渡す際には、型名は正確に記述する必要があります。特に型を修飾する場合は要注意です。
receive((shared int* i) {} );
などという書き方はNGです。
receive((shared(int)* i) {} );
上記のように、型名を正確に記述します。

以下、例を示します。

例3 11日目 @repeatedly さんの記事より抜粋

import std.stdio, std.concurrency;

void func()
{
	bool running = true;
	
	while (running) {
		receive((int i) { writeln("Received an int: ", i); },
		        (float f) { writeln("Received a float: ", f); },
		        (Variant v) {
		            writeln("Received unknown type. Terminated...");
		            running = false;
		        });
    }
}

void main()
{
	auto tid = spawn(&func);
	send(tid, 42);
	send(tid, 42.0f);
	send(tid, "Hello!");
}

1つ目のsendは42 つまり、int型です。int型は基本型で参照型を含まないので!hasUnsharedAliasing!Tを満たし、かつVariantに指定可能なので条件を満たします。
2つ目のsendは42.0f つまり、float型です。float型は基本型で参照型を含まないので!hasUnsharedAliasing!Tを満たし、かつVariantに指定可能なので条件を満たします。
3つ目のsendは"Hello!" つまり、immutable(char)[]型です。配列ですが、要素がimmutableなので!hasUnsharedAliasing!Tを満たし、かつVariantに指定可能なので条件を満たします。

##例4

main.d

import std.stdio, std.concurrency;

// クラス
class A { }
// 大きいサイズの構造体
struct S { int[32] dat; }

void func()
{
	bool running = true;
	
	while (running) {
		receive(// 以下2つの関数は、どちらが呼ばれるか確認して下さい。
		        (shared int* i) { writeln("Received an int ptr 1:", *i); },
		        (shared(int)* i) { writeln("Received an int ptr 2:", *i); },
		        // 以下のaは排他制御なしに無理やりキャストしていますが、これは
		        // main関数内で変更されないことに確証を得ているから行える行為です。
		        // 確証なしに排他制御なしでアクセスするのはNGです。
		        (shared A a) { writeln("Received a class object A:", (cast()a).toString()); },
		        // 大きいサイズの構造体は以下のようにポインタで受け渡しを行うのが吉。
		        (shared(S)* s) { writeln("Received a struct S: ", s.dat[]); },
		        (Variant v) {
		            writeln("Received unknown type. Terminated...");
		            running = false;
		        });
	}
}

void main()
{
	auto tid = spawn(&func);
	auto i = new shared int;
	auto a = new shared A;
	auto s = new shared S;
	*i = 123;
	s.dat[] = 456;
	// 以下、i, a, s の型がそれぞれ何かを確認して下さい。
	pragma(msg, typeof(i));
	pragma(msg, typeof(a));
	pragma(msg, typeof(s));
	
	send(tid, i);
	send(tid, a);
	send(tid, s);
	send(tid, "Hello!");
}

まとめ

std.concurrencyの導入について、各関数の引数に何を指定すればよいのかについてお話しました。

スレッドを超えて安全に取り扱われる型とは、!hasUnsharedAliasing!Tを満たすTのことで、shared/immutableのついていない参照データを含みません。
spawnの引数に指定することのできるものとは、!hasUnsharedAliasing!Fを満たすFと、Fの引数に暗黙変換可能な!hasUnsharedAliasing!Tを満たすT...のことです。Fには関数ポインタ以外にも、void delegate() sharedなどが含まれます。
send/receive できるものとは、!hasUnsharedAliasing!Tを満たし、かつVariantに指定可能なTのことです。また、receiveの際には型名を正確に記述しましょう。

この記事を読むとわかると思いますが、並行プログラミングを行う上で、shared属性は!hasUnsharedAliasing!Tを満たす条件にとって、immutableと同じくらい非常に高い重要性を持ちます。しかしながら、sharedの使いこなしの技術についてはまだまだ未発達な分野です。これらをうまく使おうと思うと、必ずsharedの使い方について幾つもの壁が出てくることでしょう。もし仮に、この記事を読み、sharedを使ってみて、使い方についていい方法や定石などが見えてくるようでしたら、是非記事にしてください。sharedを使う人と情報が増えることを心から願っております。
なお、私の書いたsharedの使い方についての記事は以下です。ぜひご参考にしてください。
sharedの話
 shraedのcastについて