unique_lockとunique_ptrの順序によってメモリ解放漏れが発生する

Question

概要

unique_ptrを要素とするdequeを、スレッド間のデータ受け渡しに使おうとしています。
排他制御にはmutex（lock_guardおよびunique_lock）とcondition_variableを使用しています。
が、受け取りスレッド側におけるunique_lockとunique_ptrの位置関係によって、メモリが解放されていないように見える現象に悩まされています。
原因・解決策について知見をお持ちの方がいらっしゃいましたら、ご回答よろしくお願いいたします。

環境

• Visual Studio Professional 2017
• Visual C++ / Windows コンソール アプリケーション

再現コード

コード中、NGとコメントしてある箇所でunique_ptrの定義を行うと、メモリ使用量が増加し続けます（タスクマネージャーで確認、送信側でbad_alloc発生）。
本来 NG その1の箇所で定義・解放を行いたいのですが、NG その2の箇所で定義しても同様となります。
しかしOKとコメントしてある箇所で定義した場合は、そのような現象が発生しません。

#include "stdafx.h"
#include <cstdint>
#include <mutex>
#include <condition_variable>
#include <deque>
#include <thread>
#include <iostream>

using namespace std;

static constexpr auto LENGTH = 2 * 1024 * 1024;
static mutex mtx;
static condition_variable cond;
static deque<unique_ptr<uint8_t[]>> que;

void reader()
{
	while (true) {
		unique_ptr<uint8_t[]> values;				// NG その1

		// Critical section.
		{
			//unique_ptr<uint8_t[]> values;			// NG その2
			unique_lock<decltype(mtx)> lock(mtx);
			//unique_ptr<uint8_t[]> values;			// OK

			while (que.empty()) {
				cond.wait(lock);
			}

			values = move(que.front());
			que.pop_front();
		}

		// valuesに対して処理を行う

		this_thread::yield();
	}
}
void writer()
{
	for (auto i = 0; i <= 100000; ++i) {
		try {
			unique_ptr<uint8_t[]> values(new uint8_t[LENGTH]);

			// valuesに対して処理を行う

			// Critical section.
			{
				lock_guard<decltype(mtx)> lock(mtx);

				que.push_back(move(values));
				cond.notify_one();
			}
		}
		catch (exception e) {
			cout << e.what() << endl;
		}

		this_thread::yield();
	}
}

int main()
{
	unique_ptr<thread> reader_thread(new thread(reader));
	unique_ptr<thread> writer_thread(new thread(writer));

	writer_thread->join();
	reader_thread->join();	// infinity

	return 0;
}

余談

スレッド間で「1回あたり2MB以上」「秒間25回」程度のデータを送受信する、スマートな方法が思いつかずこのようなコードになっています。
送信前に受信側バッファが一定サイズを超えていない事の確認も必要で、何か良い方法がないものでしょうか……。

Answer 1

あげられているコードをそのまま実行してみたところ、bad_allocまではいきませんでしたが瞬間的にメモリの使用量が増えます。
単純にスレッドになかなかCPUが割り当てられないタイミングがあって、writerに割り当てられなかったらreaderはキューを全部処理しきって待つだけだけど、readerに割り当てられなかったらwriterはガンガンキューに突っ込んでいくのでメモリが足りなくなるんだと思います。
CPUが割り当てられないタイミングがあるのは別段不思議ではないです。

キューのデータ数が一定数を超えたら処理されるまでwriterを待機させるとか、キューの代わりに固定サイズのリングバッファを作って、writer側も空きバッファが無かったら(readerが処理しなかったら)空きができるまで待機するようにしたらどうでしょうか？

Comments

[HARQ]

ご回答ありがとうございます。見直してみると確かにwriterが書き込み続けてますね……。
定義位置の順序を変えただけで挙動が変わる（微妙なタイミングの違いかな？）ので混乱してしまいました。
複数台のPCを用意して確認したところ、それぞれ挙動が異なることから質問自体は勘違いの様です。

と、このままではお恥ずかしい限りなので、同期実行ビジュアライザーでスレッド切り替えの模様を確認してみました。やはり現状のコードではwriter側がreader側より頻繁に動作しており、 cond.wait()で待つことがほぼ無い = readerが間に合っていなかった模様です。

試しに「que.size() < 10 の場合のみ、writerがpush & notifyする」ように変更したところ、正常に動作しているようです。今度はwriter側が追いつかず、reader側がwaitしていましたが……。

[toki_td]

このスレッドのやり取りの前後にどういう入出力があるかわかりませんが、秒間25回(40ms間隔)で安定的に切り替えたいということでしょうか？それくらいの間隔ならタイマーのほうが良いように思いますが。
Windowsでの僕の経験則ですが、２つのスレッドを安定的に相互に動かしたい場合、yieldはOSが必要と判断しなかったら素通りして切り替わらない場合があるので、1msだけsleepして確実にCPUを放棄したほうがうまく切り替わるように思います。

[HARQ]

なるほど。yieldの実装は確かSleep(0)またはSwitchToThreadだったと思うので、Sleep(1)でなければ期待したように切り替わらない可能性は高いですね。
ちなみに実際のwriterはreaderに関係なく毎秒25回動作するスレッド（からのコールバック関数）となります。reader側の処理はかなり重たいので、こちらはこまめにyieldを入れる必要がありそうです。

Answer 2

こんにちは。

VC++ 2017でやってみました。結論としてはtoki_tdさんが書かれている通りのようです。

bad_allocも発生しました。途中でque.size()を表示してみたところ、徐々に増えていきました。そこで、writer()側でque.size()が10を超えたら、sleep_for()してみたところ、10個程度までしか増えなくなりました。

送信前に受信側バッファが一定サイズを超えていない事の確認も必要で、何か良い方法がないものでしょうか……。

基本的にはこの内容は妥当な気がします。
ただ、キューのたまり具合に応じて処理負荷を調整する必要があります。
ですので、固定長のFIFOを用いることが多いです。
C++の標準ライブラリのqueue（deque含む）はどこまでもメモリを獲得することがちょっと不思議です。キューサイズに応じた処理をプログラマが自由に作れるようにマルっと任せられているのかも知れません。

condition_variableを使えば簡単ですし。

Comments

[yumetodo]

>C++の標準ライブラリのqueue（deque含む）はどこまでもメモリを獲得することがちょっと不思議です。
std::dequeじゃないコンテナを渡せばいいのでは。もっともコンテナ書くのがだるいのはそうなんだけれども

[Chironian]

これってデキューより多くエンキューされる問題ですので、enqueに手心を加える必要があります。そのような時は、キューのサイズに上限を設けると比較的スマートに回避できます。
以前調べた時はsize()の上限を設定できるSTLコンテナはなかったと思います。

デキューより大幅にエンキューが遅いコンテナでも良いでしょうが、それは処理系や動作環境に強く依存するので宛にはできないと思います。

[yumetodo]

STLの範囲ではdequeとlistしか渡せない気がしますが、boost::static_vectorをいじってリングバッファにしたようなものを・・・(言ってて面倒だなって思う

[HARQ]

ご回答、コメントありがとうございます。普段組み込み系をメインとしている身としてお恥ずかしい限りです（言い訳としてはSTLを使える環境が久しぶりすぎて……）。

実際のコードではコールバック関数としてwriterが実装され、queueに空きがあればreaderへ送信、空きがなければ破棄という動作をします。そのため単純にwriterでqueueのサイズをチェックするだけで良さそうだと考えています（lockしてque.size()が重いので、atomic<int> que_size変数で別途カウント）。

queueに空きがない場合に送信側がblockする実装については、探したところ良さそうな例がありました。
<ミューテックスと複数の条件変数>
https://cpprefjp.github.io/article/lib/how_to_use_cv.html