概要

C++でファイル読み込みを高速化する
こちらの記事の続きとなります。
ファイル読み込み＜文字列の変換　のように片方のタスクが早く終る場合、メッセージブロックがどんどん溜まり、メモリ使用量が必要以上に増えてしまいます。
そこで共有バッファに送信するメッセージブロックの数を制限することでメモリ使用量を制限します。

ソースコード

非同期エージェントライブラリのベストプラクティス
ソースコードはこちらの”調整メカニズムを使用してデータパイプライン内のメッセージ数を制限する”をそのまま使用しています。

// A Semaphore type that uses cooperative blocking semantics.
class Semaphore
{
public:
	explicit Semaphore(long long capacity)
		: _semaphore_count(capacity)
	{
	}

	// Acquires access to the Semaphore.
	void acquire()
	{
		// The capacity of the Semaphore is exceeded when the Semaphore count 
		// falls below zero. When this happens, add the current context to the 
		// back of the wait queue and block the current context.
		if (--_semaphore_count < 0)
		{
			_waiting_contexts.push(Context::CurrentContext());
			Context::Block();
		}
	}

	// Releases access to the Semaphore.
	void release()
	{
		// If the Semaphore count is negative, unblock the first waiting context.
		if (++_semaphore_count <= 0)
		{
			// A call to acquire might have decremented the counter, but has not
			// yet finished adding the context to the queue. 
			// Create a spin loop that waits for the context to become available.
			Context* waiting = NULL;
			while (!_waiting_contexts.try_pop(waiting))
			{
				(Context::Yield)(); // <windows.h> defines Yield as a macro. The parenthesis around Yield prevent the macro expansion so that Context::Yield() is called.  
			}

			// Unblock the context.
			waiting->Unblock();
		}
	}

private:
	// The Semaphore count.
	atomic<long long> _semaphore_count;

	// A concurrency-safe queue of contexts that must wait to 
	// acquire the Semaphore.
	concurrent_queue<Context*> _waiting_contexts;
};

Semaphoreクラスを作成する場合にはコンストラクタにメッセージの許容数を指定します。
その後メッセージを送受信するAgentに同じSemaphoreのインスタンスを渡し、sendの前にacquire(),receive()の後にrelease()を実行します

//Agentを用いてファイルの読み込みと変換を並列実行する
concurrency::concurrent_vector<lineInfo>& ReadWithAgent() {
	shared_ptr<Semaphore> semaphore = make_shared<Semaphore>(4);//メッセージブロックを４個に制限する
	unbounded_buffer<shared_ptr<vector<string>>> buffer;
	FileRead_agent reader( buffer, ReadFileName,1,gSentinel,semaphore );//同じsemaphoreを渡す
	Converter_Agent converter( buffer,false,gSentinel,semaphore );//同じsemaphoreを渡す

	reader.start();
	converter.start();

	agent::wait( &reader );
	agent::wait( &converter );

	return converter.GetInfos();
}

送信時

				_semapho->acquire();
				send( _target, buffers );

受信時

			//次のブロックを読み込む
			buffers = receive( _source );
			_semapho->release();

結果

４個にメッセージを制限した場合、Context::Block()でブレークポイントを置くとtargetのメッセージカウントが４つになっていることがわかります。

処理の流れ

構造自体はとてもシンプルで、acquire()を実行すると_semaphore_countが-1され、0未満となればそのコンテキストがブロックされます。
_semaphore_countには制限したい個数が入っているので、4が入っていたならば4回のacquire()まで許容する、といった形です。
逆にrelease()が実行されると_semaphore_countが+1され、またメッセージ送信ができるようになります。

あとがき

並列処理を知ることでatomic変数やshared_ptrなどの使い方が少しづつわかってきたように感じます。
今回はあっさりした記事でしたが、いいねや補足訂正コメントなどいただけるととっても嬉しいです。
コンテキスト周りの理解が浅いのでそのあたりを追記したい・・・