C ++およびSObjectizerでのCSPチャネルを使用したマルチスレッド（ただし、アクターなし）...

以前は、C ++のアクターフレームワークとしてSObjectizerについて説明しましたが、実際にはこれは完全に真実ではありません。 たとえば、SObjectizerには長い間、 mchains （ CSPモデルからのチャネル）のようなクールなものがあります。 Mchain-sを使用すると、ワークフロー間でのデータ交換を簡単かつ自然に整理できます。 必ずしも必要ではないエージェントを作成しない。 先日、この機能を利用して、チャネル（SObjectizer mchains）を介してストリーム間でデータを転送することで、人生を簡素化する機会が再びありました。 Goだけでなく、CSPの使用を楽しむことができます。 C ++では、これも可能です。 猫の下で誰が何をどのように気にかけてください。

タスクは次のとおりです。同期要求を行う必要がある特定のサードパーティシステムがあります。 リクエストが1つのストリームではなく複数のストリームに送られた場合、このシステムの動作を確認する必要がありました。 これを行うには、既存のシングルスレッドマルチスレッドクライアントを作成する必要があります。このクライアントの作業スレッドは、独自のリクエストストリームをサードパーティシステムに発行します。

実行されるはずだったリクエストの完全なリストは、別のファイルにありました。 そのため、このファイルを順番に読み取り、別の要求を取得して、空き作業スレッドの1つに渡す必要がありました。 各スレッドは、完了したリクエストの数をカウントしました。 すべてのリクエストを校正して処理するのにかかる時間を決定し、完了したリクエストの数を計算する必要がありました。

明らかな解決策は単純な解決策でした。 要求ファイルを読み取るメインの作業スレッドがあります。 各リクエストは、リクエストの一般的なキューに入れられます。 作業スレッドによってリクエストが解析される場所。 すなわち 作業スレッドはキューから最初のリクエストを取得して実行し、次にキューから新しい最初のリクエストを取得します。 キューが空の場合、作業スレッドはキューに何かが現れるまで一時停止する必要があります。 いっぱいになっている場合は、キューに空き領域が現れるまでメインスレッドを中断する必要があります。

SObjectizerのMchainを使用すると、スレッドセーフキューを作成せずに実行できます。

この問題を解決するには、2つのmchainが必要でした。 最初のmchainは、読み取り要求を作業スレッドに送信するために使用されます。 メインスレッドはリクエストを書き込み、

ワーカースレッドはそこからリクエストを読み取ります。 要求ファイルが完全に読み取られると、メインスレッドはこのmchainを閉じます。 したがって、ワーカースレッドがmchainに何も存在せず、閉じられていることがわかると、すぐに作業を完了します。

2番目のmchainが必要だったのは、作業スレッドが、作業と処理したリクエストの数を保証した情報をメインスレッドに転送できるようにするためでした。 このmchainでは、ワーカースレッドは1つのメッセージのみを書き込みます。 そして、メインスレッドはこのmchainからのみ読み取ります。

さて、これですべてがコードでどのように見えるかを見ることができます。 コメントなしのコード 一度だけの排出プログラムでした。 したがって、必要な説明は、対応するコードの後に​​記載されます。

run_app関数から始めましょう。これは、プログラムがコマンドラインパラメーターを解析した直後にmain（）から呼び出されます。

void run_app( const app_args_t & args ) { so_5::wrapped_env_t sobj( []( so_5::environment_t & ) {}, []( so_5::environment_params_t & params ) { params.infrastructure_factory( so_5::env_infrastructures::simple_mtsafe::factory() ); } ); auto tasks_ch = create_mchain( sobj, std::chrono::seconds(5), 50, so_5::mchain_props::memory_usage_t::preallocated, so_5::mchain_props::overflow_reaction_t::abort_app ); auto finish_ack_ch = create_mchain( sobj ); std::vector< std::thread > workers; const auto cleanup = cpp_util_3::at_scope_exit( [&] { so_5::close_drop_content( finish_ack_ch ); so_5::close_drop_content( tasks_ch ); for( auto & t : workers ) t.join(); } ); cpp_util_3::n_times( args.m_threads_count, [&] { workers.emplace_back( [&] { worker_thread( args, tasks_ch, finish_ack_ch ); } ); } ); do_main_work( args, tasks_ch, finish_ack_ch ); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここでは、まず、mObjectが属するSObjectizer Environmentのインスタンスが作成されます。 SOEnvironmentがないと、mchainを作成できないため、SOEnvironmentを作成する必要があります。

ただし、SOEnvironmentが複数の独自の補助スレッドを作成することを強制される効果的な管理のために、アプリケーション内にエージェントのクラウドを作成するように設計された本格的なSOEnvironmentは必要ありません。 したがって、SOEnvironmentのパラメーターでは、 特別なシングルスレッドバージョンのSObjectizerを使用するように要求します。 この場合、wrapped_env_tは1つの補助スレッドを作成し、その上でso_5 :: launch（）が呼び出されます。 それ以上のSObjectizerは何もしません。 はい。この補助スレッドは、run_appから戻るまでso_5 :: launch（）でスリープします。

次に、リクエストを作業スレッドに分散するためにmchainが必要です。 これがtasks_chです。 しかし、これは単純なmchainではありません。 まず、容量が制限されたmchainです。 いっぱいになったmchainに別のメッセージを追加しようとすると、現在のスレッドがブロックされます。 ただし、永久にブロックするのではなく、5秒間だけブロックします。 5秒後でもmchainに空き領域がない場合、std :: abort（）を呼び出すことでアプリケーション全体が中断されます。 この場合、正当化されます。 通常の状態では、5秒は言うまでもなく、動作中のスレッドの1つが数ミリ秒以上スリープ状態になることはありません。 したがって、5秒以内にtasks_chに空き領域がない場合は、間違いが発生しているので、std :: abort（）を呼び出す必要があります。 さらに、tasks_chのサイズはあらかじめ決められているため、mchainのメッセージキュー全体に必要なメモリをすぐに割り当てるように指示します。

ワーカースレッドがfinish_ackメッセージを送信する2番目のmchainでは、すべてがはるかに単純です。 したがって、finish_ack_chは、デフォルトのパラメーター（送信操作をブロックしない無次元mchain）を使用してcreate_mchainを呼び出すだけで作成されます。

次に、N個の作業スレッドを開始し、それらをワーカーベクトルに保存する必要があります。 しかし、ここではそれほど単純ではありません。 次の作業スレッドを作成するときに例外を取得できます。 この場合、すでに作成されているスレッドを通常完了すると便利です。

以前に実行した操作をロールバックして生活を簡素化するために、Dスコープsc_exitのアナログが使用されます（BOOST_SCOPE_EXITのアナログか、Gセクションの延期のいずれか、ここでは誰にも近い）。 クリーンアップ変数は、基本的に内部にラムダを持つオブジェクトです。 このラムダは、クリーンアップ変数が呼び出されたときに呼び出されます。 小さなcpp_utilヘルパーライブラリを使用したクリーンアップによって作成されます。 クリーンアップに関する別の説明：クリーニング時に最初に行う必要があるのは、mchainsを閉じることです。 作業スレッドのいずれかがすでに開始され、tasks_chからの受信呼び出しでスリープ状態になった場合、クリーンアップでtasks_chを閉じると、すぐにこのスレッドが起動し、作業を完了することができます。

それでは、作業スレッドを作成してdo_main_workを呼び出します。 do_main_workの内部では、アプリケーションのメインスレッドのメイン作業が実行されます。リクエストを含むファイルの読み取り、作業スレッドへのリクエストの送信、結果の収集です。 次に、do_main_workの簡易バージョンがどのように見えるかを示します。このバージョンからマイナーな詳細が削除されました。

 void do_main_work( const app_args_t & args, so_5::mchain_t tasks_ch, so_5::mchain_t finish_ack_ch ) { data_file_handler_t file{ args.m_data_file, args.m_force_keep_alive }; const auto started_at = hires_clock::now(); while( !file.is_eof() ) { auto request = file.get_next_request(); if( !request ) break; so_5::send< std::string >( tasks_ch, *request ); } so_5::close_retain_content( tasks_ch ); unsigned long long total_requests{}; so_5::receive( from(finish_ack_ch).handle_n( args.m_threads_count ), [&]( const finish_ack_t & what ) { total_requests += what.m_requests; } ); const auto total_time = hires_clock::now() - started_at; if( total_requests ) { ... // Print the results... } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここで最も興味深いものはすべて2つの場所に集められています。

まず、しばらくの間。 そこでは、ファイルからのリクエストが順番に読み込まれ、sendを呼び出すことでワークスレッドに渡されます。 tasks_chがいっぱいのときにsendが呼び出されると、メインスレッドは中断されます（ただし、5秒以内）。

第二に、リクエストファイル全体が読み込まれたら、すべての作業スレッドからの応答を待つ必要があります。 これを行うには、最初にtasks_chを閉じて、作業中のスレッドが作業を完了する時が来たことを理解します。 ただし、既にキューにあるがまだ処理されていない要求が失われるように、それを閉じる必要があります。 したがって、close_retain_contentが呼び出されます（ただし、run_appのクリーンアップアクションでは、close_drop_contentが使用されました。これは、閉じられているチャネルに何も保存する必要がないためです）。

tasks_chを閉じた後、N個の作業スレッドからの応答を待つ必要があります。 正確にN個の回答が期待されることは、1つの魔法の行に記録されます。

 so_5::receive( from(finish_ack_ch).handle_n( args.m_threads_count ),
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

彼女は、文字通り次のように言っています。threads_count個のメッセージが読み取られて処理されるまで、finish_ack_chチャネルから読み取ります。

さて、図を完成させるには、作業スレッドのコードがどのように見えるかを示す必要があります。 それは非常に簡単です：

 void worker_thread( const app_args_t & args, so_5::mchain_t tasks_ch, so_5::mchain_t finish_ack_ch ) { io_performer_t io_performer{ args.m_srv, args.m_port }; unsigned long long total_requests{}; so_5::receive( from(tasks_ch), [&]( const std::string & request ) { io_performer.request_response( request ); ++total_requests; } ); so_5::send< finish_ack_t >( finish_ack_ch, total_requests ); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

スレッドは、tasks_chチャネルからreceive内でハングします。 受信からの戻りは、tasks_chが閉じられたときに発生します。 tasks_chが空の場合、receiveはまでスリープします

何かがチャネルに入るまで（またはチャネルが閉じられるまで）。 そして、受信からの戻りが発生すると、作業スレッドは単にfinish_ackメッセージをfinish_ack_chに送信して終了します。

実際、それがすべてです。

フロー間のマルチスレッド化と情報交換に問題はなかったと言わざるを得ません。 文字通り、最初に起動して動作しました。 io_performer_t :: request_response実装内で問題が発生しました。クライアントとサーバー間の対話の実装のエラーが原因で、現在のスレッドが中断されました。 そして、tasks_ch全体への書き込みを待機する時間の5秒の制限が役立ちました。スレッドがハングし始めたとき、タイムアウトが機能し、マルチスレッドクライアントがクラッシュしました。 バグ、さらにrequest_responseにバグがあることがすぐに明らかになりました。 tasks_chからの通常の読み取りを停止できるのは、そこにぶら下がっていただけです。

結論として、アクターのモデルとインタラクティブシーケンシャルプロセスのモデル（別名CSP）の両方が素晴らしいものであると言いたいと思います。 1つはうまく機能し、2つ目はどこかで機能します。 SObjectizerでは、両方を使用できます。 そして、一度にすべて、時にはこれが時々必要です。