【ファイル読み込み高速化】セマフォを使って共有バッファのメモリを制限する

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

# ソースコード
[非同期エージェントライブラリのベストプラクティス](https://learn.microsoft.com/ja-jp/cpp/parallel/concrt/best-practices-in-the-asynchronous-agents-library?view=msvc-170)
ソースコードはこちらの”調整メカニズムを使用してデータ パイプライン内のメッセージ数を制限する”をそのまま使用しています。
``` cpp
// 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()を実行します

``` cpp
//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();
}
```
## 送信時
``` cpp
				_semapho->acquire();
				send( _target, buffers );
```
## 受信時
``` cpp
			//次のブロックを読み込む
			buffers = receive( _source );
			_semapho->release();
```

# 結果
４個にメッセージを制限した場合、Context::Block()でブレークポイントを置くとtargetのメッセージカウントが４つになっていることがわかります。
![acquire.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/257894/06b52e06-f601-ef06-a56a-258c869746dc.jpeg)
![semaphore.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/257894/3ae14333-2375-b319-03d9-41eb58a4b1e7.jpeg)


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

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