#はじめに

俺は去年の秋頃に「さいきょうの関数型プログラミング言語は C++ かもしれない」とかなんとかそんな記事を投稿した者だ。憶えている人はいるだろうか…。

あの時いろいろ調べて分かったのは、new だの malloc だのといった動的なメモリ確保には結構な時間がかかる、ことにマルチスレッド下ではロックが発生するらしいということだ。まあそうだわな。それぞれのスレッドが勝手にメモリを確保したらぐちゃぐちゃになっちゃうからな。

マルチスレッド下での動的なメモリ確保において発生するロックを回避する方法としては、malloc を入れ替えるというのがあるらしい。FreeBSD の jemalloc とか Google 先生の TCMalloc とかだ。これは小さなメモリを確保する場合にそれぞれのスレッド専用に確保された領域を使うことでロックを回避しているそうだが良くは知らない。

しかしこういうなんちゃら malloc は環境に依存するみたいだし、俺はできるだけなんでもお仕着せのやつを使いたい人なんだ。

そこでうだうだ考えていたところ、いいことを思いついたのだった。一度確保したメモリを解放せず thread_local にキャッシュし再利用するというアイデアだ。必要なメモリの量がテスト段階で明らかになれば、プログラムの初期化時にキャッシュを作っておくことで以後は高速にメモリ確保を行える。だから「半」動的というわけ。

これはガチで使えるかもしれないから前回の記事でがっかりさせてしまった人にもざっくり目を通してもらいたいと思う。まあひょっとしたら別の人がとっくに思いついてすでに実装済みかもしれないが。

#cached_ptr

実装としてはスマートポインタの形式を取っている。いや、スマートポインタじゃないのかなこれ？まあ普通に配列のように使えるんだよね。

struct number { int num; }
cached_ptr <number, 100> nums = {1, 2, 3};
for (const auto & h : numbers)
	printf ("%d\n", h.num);
for (int i=0; i < nums.size(); i++)
	printf ("%d\n", nums [i]);

テンプレート引数は最初のが型名、次のが最大の要素数だ。上の例ではにょろ括弧で初期化指定している要素が３つだから、nums の要素数は３で、そこから最大100までの範囲で増減できることになる。std::vector と同じように push_back とか resize といったメソッドを用意している。なお２番目のテンプレート引数を省略した場合は要素１になり単独の変数のように扱える。

型と最大数をテンプレートで指定することによって必要なメモリ量は静的に決定される。cached_ptr 内部においてスレッドローカルにキャッシュされたメモリを取り出す部分のコードは以下のようになる。コードの断片だからまじめに読む必要はないからね。

template <size_t C>
memory_chain<C> & get_memory_chain ()
{
	thread_local memory_chain<C> chain;
	return chain;
}

スレッドローカルで宣言しているのは memory_chain というクラスのオブジェクトだな。こいつの pop() というメソッドを呼び出すと、すでにキャッシュされているメモリがあればそれを返すし、なければ新たに動的な確保を行う。

memory_node<C> * pop ()
{
	if (reserved)
		unreserve();
	auto ret = chain;
	if (ret)
		chain = ret->next;
	else
		ret = new memory_node<C>;
	ret->chain = this;
	return ret;
}

ここで reserved というのがでてくるが、これは別のスレッドからメモリが戻された場合につなぐリストだ。同一スレッド内でメモリが戻された場合には chain というリストにつなぐ。メモリを戻す push() というメソッドは以下のようになっている。

void push (memory_node<C> * node)
{
	if (thread_id != std::this_thread::get_id()) {
		reserve (node);
		return;
	}
	node->next = chain;
	chain = node;
}

reserved につなぐ時にはロックがかかるが、chain につなぐ時にはロックがかからない。また取り出す時にも reserved にメモリがつながれている場合はロックをかけて chain につなぎ変えるということをするが、reserved が空ならロックがかからない。つまり同一スレッド内でメモリを取り出したり戻したりするのであればロックなしで高速に行えるわけだ。

テンプレートを使ってメモリサイズ毎に memory_chain を作成し、どの memory_chain からメモリを取り出すのかをコンパイル時に決定しているのもいいところだ。メモリの断片化が起こらないのもいい。

欠点としてはスレッドプールでスレッドをいろいろな目的で使い回した場合に同じようなメモリのキャッシュがスレッド毎にできてしまい無駄なことだ。また上述のようにあるスレッドで発生したデータを別のスレッドで破棄するとロックが発生するから、データの破棄は元のスレッドに戻した上で行うのが望ましい。そこから言ってもスレッドを目的別に管理するということが必要になってくると思う。あとスレッドを破棄するのは全ての cached_ptr が破棄された後にしなければならない。スレッドがなくなったらメモリを戻せなくなっちゃうからね。

なおメモリ自体はただの char 型の配列で、コンストラクタやデストラクタは cached_ptr の中で呼び出している。配置 new というやつだ。

また memory_chain はデストラクタでいくつのメモリを解放したかを標準出力に出力する。

lm2::make_memory_cache<800>(9);
lm2::make_memory_cache<1600>(2);

この例では800バイトのメモリを９個、1600バイトのメモリを２個ということになる。こいつをソースコードに張り付けて初期化時にスレッドに実行させれば事前にキャッシュを作成できるというわけだ。

#linear_move_2

前回の linear_move は std::vector ベースだったが、これをわりと単純に cached_ptr に置き換えたのが linear_move_2 だ。おざなりな通り一遍のテストコードは一応通った。

前回の記事でlinear_move の紹介に用いた無意味なコードをlinear_move_2 に移植すると以下のようになる。

auto hoge =
	fold (Hoge (1),
		[] (Hoge && x, Fuga && y) {
			int xn = x.get_num();
			int yn = y.get_num();
			x.set_num (xn + yn);
			return std::move (x);
		},
	filter (
		[] (const Fuga & fuga) {
			int num = fuga.get_num();
			return
				!(num % 2) ?
					false:
				!(num % 3) ?
					false:
				true;
		},
	map (
		[] (Hoge && hoge) {
			return Fuga (hoge.get_num() * 5);
		},
	progress<1000> (1000, Hoge(1),
		[] (const Hoge & hoge) {
			return Hoge (hoge.get_num() + 1);
		}
	))));

違うのは progress() って関数でテンプレート引数に最大要素数をしているところだけで、他は同じだな。

#おわりに

static とか thread_local とかで静的に確保された配列と cached_ptr を比べてみると、cached_ptr のいいところは再帰呼び出しのなかで使えたり戻り値として外に出せたりすることかな。上述のようにスレッドの外に出すことも可能だ。

なので実時間制御の人工知能とか、同じループを高速に延々とぶんまわすプログラムの中で複雑なアルゴリズムを実装するのがやりやすくなるんではないかと思う。まあ俺は実時間制御の人工知能がどうやって実装されているかなんて全然知らないんだけどね。

あと cached_ptr のソースではにょろ括弧による初期化とか範囲 for 文への対応とか配置 new とか頑張ってみたんだけど、全部インターネットから仕入れた付け焼き刃の知識で実装してるから何かおかしなことをやっていたら教えてもらえるとありがたい。

いじょう