SObjectizer：シンプルから複雑へ。 パートII

最初の記事では、SObjectizerについて説明しました。 2番目の記事では、エージェントがどのように見えるか、なぜ、どのように、どこで進化するかについて話し始めました。 今日もこの話を続け、デモエージェントの実装をさらに複雑にします。 同時に、非同期メッセージングの信頼性を確認します。



前回、電子メールでファイルの内容を読み取る操作は別のIOエージェントに任せる必要があるという事実に決着しました。 それをして、何が起こるか見てみましょう。



最初に、IOエージェントとemail_analyzerが相互に交換する一連のメッセージが必要です。

//     . struct load_email_request { //    . string email_file_; //    . mbox_t reply_to_; }; //    . struct load_email_succeed { //  . string content_; }; //    . struct load_email_failed { //   . string what_; };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、email_analyzerエージェントがリクエストメッセージload_email_requestを送信する正確な場所を決定する必要があります。 通常の方法で行うことができます：IOエージェントを登録するとき、そのdirect_mboxを保存し、このmboxをアナライザーエージェントコンストラクターにパラメーターとして渡し、次に各エージェントのemail_analyzerエージェントコンストラクターにパラメーターを渡します...基本的に、複数の異なるIOが必要な場合エージェント、それがどのように行われるべきかです。 しかし、私たちの問題では、1つのIOエージェントで十分です。 これにより、名前付きmboxのデモを行うことができます。



名前付きmboxは、so_5 :: environment_t :: create_mbox（name）を呼び出すことによって作成されます。 同じ名前でcreate_mboxを複数回呼び出すと、同じ名前のcreate_mboxが最初に呼び出されたときに作成される同じmboxが常に返されます。



IOエージェントは名前付きmboxを作成し、サブスクライブします。 Email_analyzerエージェントは、load_email_requestメッセージを送信する必要があるときに同じmboxを取得します。 したがって、analyzer_managerを介してmbox IOエージェントを「ドラッグ」する必要がなくなります。



IOエージェントとemail_manager間のインターフェースを決定したので、 email_analyzerエージェントの新しいバージョンを作成できます。

 //  .   IO-  IO-. class email_analyzer : public agent_t { public : email_analyzer( context_t ctx, string email_file, mbox_t reply_to ) : agent_t(ctx), email_file_(move(email_file)), reply_to_(move(reply_to)) {} //  ,      . //         //    . virtual void so_define_agent() override { //       IO-.  //       . so_subscribe_self() .event( &email_analyzer::on_load_succeed ) .event( &email_analyzer::on_load_failed ); } virtual void so_evt_start() override { //       IO-   //  email . send< load_email_request >( // mbox IO-    . so_environment().create_mbox( "io_agent" ), email_file_, //       mbox. so_direct_mbox() ); } private : const string email_file_; const mbox_t reply_to_; void on_load_succeed( const load_email_succeed & msg ) { try { //     . auto parsed_data = parse_email( msg.content_ ); auto status = check_headers( parsed_data->headers() ); if( check_status::safe == status ) status = check_body( parsed_data->body() ); if( check_status::safe == status ) status = check_attachments( parsed_data->attachments() ); send< check_result >( reply_to_, email_file_, status ); } catch( const exception & ) { //   -      //    email-   . send< check_result >( reply_to_, email_file_, check_status::check_failure ); } //    ,   , //   . so_deregister_agent_coop_normally(); } void on_load_failed( const load_email_failed & ) { //    .    //      . send< check_result >( reply_to_, email_file_, check_status::check_failure ); so_deregister_agent_coop_normally(); } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Email_analyzerエージェントは、これを効率的に行う方法を知っている別のエージェントにIO操作を委任するようになりました。 したがって、作業スレッドのemail_analyzerエージェントは、IOエージェントへの割り当てを処理するか、email_analyzer応答を処理します。 これにより、作成できるemail_analyzerエージェントの数と、必要なワークスレッドの数のビューを変更することができます。



各email_analyzerエージェント自体が同期IO操作を実行した場合、並列IO操作を許可するのと同じ数のワークスレッドをプールに入れる必要がありました。 プール内の作業スレッドの数よりも多くのemail_analyzerエージェントを作成しても意味がありませんでした。 プールに16のスレッドがあり、32のエージェントが同時に存在できるようにすると、これらのエージェントの半分は、作業スレッドが解放されるのを単に待つことになります。



現在、IO操作が別の作業コンテキストに移動された後、まず、プール内の作業スレッドの数を減らすことができます。 イベント内のEmail_analyzerエージェントは、主にCPUロード操作を実行します。 したがって、利用可能な処理コアよりも多くのワークフローを作成しても意味がありません。 したがって、4コアプロセッサがある場合、プール内に16スレッドではなく、4スレッド以下である必要があります。



次に、IO操作が電子メールのコンテンツの処理よりも時間がかかる場合、プール内のスレッドよりも多くのemail_analyzerエージェントを作成する機会が得られます。 単純に、これらのエージェントのほとんどは、IO操作の結果を待ちます。 ただし、電子メールのダウンロード時間がコンテンツの分析時間に匹敵するか、それより短い場合、このアイテムは関連性を失い、プール内のスレッドの数よりも多く1-2-3 email_analyzerエージェントのみを作成できます。 これらの設定はすべて、analyzer_managerエージェントで1か所で簡単に行えます。 コード内のいくつかの定数を文字通り変更し、変更がソリューションのパフォーマンスにどのように影響するかを確認するだけで十分です。 ただし、パフォーマンスチューニングは別の大きなトピックであり、今すぐ掘り下げるのは時期尚早です...



したがって、以前のバージョンの問題を修正するemail_analyzerエージェントの次のバージョンがあります。 許容できると考えてもいいですか？



いや



問題は、結果の実装が信頼できると見なされないことです。



この実装は、送信するメッセージが失われずに常に宛先に届くという楽観的なシナリオ向けに設計されており、宛先に届くと常に処理されます。 その後、常に必要な答えが得られます。



しかし、人生の厳しい真実は、システムが個々のアクター/エージェント間の非同期メッセージ交換で構築されている場合、この同じ非同期交換は絶対に信頼できるものとは見なされないということです。 メッセージが失われる可能性があります。 そして、これは正常です。



メッセージの損失はさまざまな理由で発生する可能性があります。 たとえば、受信エージェントはまだメッセージをサブスクライブできていません。 または、現時点では受信者エージェントがまったくいない。 どちらかですが、過負荷保護メカニズムは機能しています（これについては後続の記事で詳しく説明します）。 エージェントがいて、メッセージが届きましたが、エージェントはこのメッセージが処理されていない状態です。 エージェントがいて、メッセージが彼に届き、処理を開始しましたが、処理中に何らかのアプリケーションエラーが発生し、失敗したエージェントは応答を送信しませんでした。



一般に、非同期メッセージを介したエージェント間の通信は、UDPプロトコルを介したホストの対話に似ています。 ほとんどの場合、データグラムは受信者に届きます。 ただし、処理中または処理中に失われることもあります。



上記は、load_email_requestがIOエージェントに到達しない可能性があることを意味します。 または、応答メッセージload_email_successed / load_email_failedがemail_analyzerエージェントに届かない場合があります。 そして、この場合はどうなりますか？



システムに存在するが何もしないemail_analyzerエージェントを取得します。 動作しません。 死ぬつもりはない。 また、他のemail_analyzerエージェントを開始しません。 幸運でなければ、analyzer_managerによって作成されたすべてのemail_analyzerエージェントが、何もしない半死体に変わる状況に遭遇する可能性があります。 その後、analyzer_managerは単純に順番にリクエストを蓄積し、タイムアウトが経過するとそれらを破棄します。 ただし、有用な作業は実行されません。



この状況から抜け出す方法は？



たとえば、タイムアウトを制御することにより。 email_analyzerエージェントによるIO操作の実行時間の制御を導入できます（つまり、応答が長すぎる場合、IO操作が失敗したと想定します）。 または、analyzer_managerエージェントで電子メール分析操作全体の実行時間を制御できます。 または両方を行います。



簡単にするために、email_analyzerエージェントで IO操作のタイムアウトをカウントするように制限しています：

 //  .   -   IO-. class email_analyzer : public agent_t { //     ,    //   IO-    . struct io_agent_response_timeout : public signal_t {}; public : email_analyzer( context_t ctx, string email_file, mbox_t reply_to ) : agent_t(ctx), email_file_(move(email_file)), reply_to_(move(reply_to)) {} virtual void so_define_agent() override { so_subscribe_self() .event( &email_analyzer::on_load_succeed ) .event( &email_analyzer::on_load_failed ) //    -   IO-. .event< io_agent_response_timeout >( &email_analyzer::on_io_timeout ); } virtual void so_evt_start() override { //       IO-   //  email . send< load_email_request >( so_environment().create_mbox( "io_agent" ), email_file_, so_direct_mbox() ); //      -    IO-. send_delayed< io_agent_response_timeout >( *this, 1500ms ); } private : const string email_file_; const mbox_t reply_to_; void on_load_succeed( const load_email_succeed & msg ) { try { auto parsed_data = parse_email( msg.content_ ); auto status = check_headers( parsed_data->headers() ); if( check_status::safe == status ) status = check_body( parsed_data->body() ); if( check_status::safe == status ) status = check_attachments( parsed_data->attachments() ); send< check_result >( reply_to_, email_file_, status ); } catch( const exception & ) { send< check_result >( reply_to_, email_file_, check_status::check_failure ); } so_deregister_agent_coop_normally(); } void on_load_failed( const load_email_failed & ) { send< check_result >( reply_to_, email_file_, check_status::check_failure ); so_deregister_agent_coop_normally(); } void on_io_timeout() { //     ,     -. send< check_result >( reply_to_, email_file_, check_status::check_failure ); so_deregister_agent_coop_normally(); } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このオプションemail_analyzerは、すでにかなり受け入れられると考えることができます。 彼のコードでは、リファクタリングは、いくつかの操作（sendおよびso_deregister_agent_coop_normally）を別個のヘルパーメソッドに組み込むことを提案しています。 ただし、これは意図的に行われたものではないため、email_analyzerエージェントの後続の各バージョンのコードは、前のバージョンのコードと最小限の違いしかありません。



上記のemail_analyzerエージェントの2つのバージョンを比較すると、日常業務でSObjectizerを長年使用してきたプログラマーに高く評価されている1つの機能がわかります。それは、エージェント拡張手順の単純さとわかりやすさです。 エージェントは別のイベントに応答する必要がありましたか？ そのため、別のサブスクリプションと別のイベントハンドラーを追加する必要があります。 通常、サブスクリプションは同じ場所で行われるため、どこに行って何を編集するかがすぐにわかります。



SObjectizerは、エージェントがイベントを書き込む場所と方法に制限を課しませんが、単純な合意に従います-サブスクリプションはso_define_agent（）で行われ、非常に単純な場合、エージェントの最終クラス、コンストラクターで-生活を大幅に簡素化します。 あなたは他の誰かのエージェントのコードまたはあなたのエージェントのコードさえ見ますが、数年前に書かれており、エージェントの振る舞いを理解するために何を見る必要があるかすぐにわかります。 便利ですが、この便利さを理解するには、おそらく複数の実際のエージェント、または2つ以上のエージェントを作成してデバッグする必要があります。



ただし、上記で挙げられ、email_analyzerエージェントの次の6番目のバージョンが登場したため、エージェントの信頼性のトピックに戻りましょう。エージェント間の非同期メッセージングのメカニズムは信頼できません 。



ここで重要な発言をする必要があります。SObjectizerのメッセージ配信メカニズムは本当に「漏れやすい」と言って間違っており、必要に応じてメッセージを失うことができます。



SObjectizerのメッセージは単に失われるのではなく、すべての損失には独自の理由があります。 エージェントが自身にメッセージを送信し、送信機能が成功した場合、メッセージはエージェントに届きます。 開発者自身が特定のケースでSObjectizerにこのメッセージをスローするように明示的に何らかのアクションをとらない限り（たとえば、開発者はいずれかの状態でエージェントをメッセージにサブスクライブしないか、limit_then_dropを使用して過負荷から保護します）。



そのため、開発者自身が特定の状況でSObjectizerが特定のメッセージをスローすることを許可しない場合、エージェントが自分に送信したメッセージはエージェントに届くはずです。 したがって、上記のコードでは、保留中のメッセージが途中で失われることを恐れずに、完全に落ち着いてメッセージを送信しました。



ただし、メッセージが別のエージェントに送信されると、状況は多少変わります。 配達の成功に自信がある場合があります。 たとえば、私たち自身が受信者エージェントを実装し、送信者エージェントが住んでいるのと同じ協力にそれを含めた場合です。



しかし、受信者エージェントが私たちによって書かれていない場合、外国の協力の一環として作成および破棄され、その動作を制御しない場合、エージェントが過負荷からどのように保護されるか、特定の状況でどのように動作するかがわからない場合、自信がありますUDPプロトコルを介してデータグラムを送信する場合と同じです。すべてが正常であれば、データグラムが送信者に到達し、応答が返される可能性が高いです。 すべてがうまくいけば。 しかし、そうでない場合は？



興味深い点に来ました。非同期メッセージングの相対的な信頼性が低いため、アクター/エージェントでのソフトウェア開発は、プログラム内のオブジェクトの同期相互作用に基づくアプローチを使用する場合よりも時間がかかるように見えます。



時々そうです。 しかし、最終的に、私たちの意見では、ソフトウェアはより信頼性が高くなります。 コードでは、メッセージの損失と、メッセージの配信および処理時間の変動の両方に関連する多くの緊急事態を処理する必要があります。



email_analyzersが非同期メッセージではなく同期要求を介してio_agentにアクセスし、io_agentが例外をスローすることでIO操作中の障害について通知するとします。 長い間、すべてが正常に機能します。email_analyzerはio_agentを同期的に要求し、それに応じて電子メールの内容または例外を受信します。 しかし、ある時点で、io_agent内のどこかに隠れたバグが現れ、同期呼び出しがハングします。 答えも例外もなく、ただ凍結するだけです。 したがって、1つのemail_analyzerが最初にハングし、次に別のメール、次に別のメールなどがハングします。 その結果、アプリケーション全体が中断されます。



一方、非同期メッセージングでは、io_agentエージェントはそのGibletsのどこかにハングする可能性があります。 ただし、これはemail_analyzerエージェントには影響しません。email_analyzerエージェントは、リクエストタイムアウトの有効期限を簡単に追跡し、否定的な結果を送信できます。 つまり、アプリケーションの一部で障害が発生した場合でも、アプリケーションの他の部分は作業を続行できます。この作業では、否定的な応答のストリームを生成することになります。 結局のところ、このフローの実際は重要な症状になり、アプリケーションで何かがうまくいかなかったことをオブザーバーに伝えることができます。



ところで、エージェントで書かれたアプリケーションの作業を監視するというトピックについて。



SObjectizerとの長年の作業を通じて、アクター/エージェント上に構築されたアプリケーションの内部で何が起こっているかを見る能力が非常に重要であるという信念を発展させてきました。 原則として、これはこの記事でも示されています。 タイムアウト制御なしでemail_analyzerの5番目のバージョンを使用して実行しようとすると、リクエスト処理が停止するまで速度が低下する様子を見ることができます。 しかし、問題が何であるかを正確に理解する方法は？



現在作成されているemail_analyzerエージェントの数と各エージェントが何をしているのかについての情報は、良い手がかりを与えることができます。 これには、アプリケーション内で何が起こっているかを監視する機能が必要です。 これはまさに、Erlangとそのプラットフォームが非常に高く評価されているものです。そこで、稼働中のErlang VMに接続し、Erlangプロセスのリスト、それらのパラメーターなどを見ることができます。 しかし、Erlangでは、ErlangアプリケーションがErlang VMを実行しているため、これが可能です。



ネイティブC ++アプリケーションの場合、状況はより複雑です。 SObjectizer Environment内で何が起こっているかを監視するためのツールがSObjectizerに追加されました（これらのツールはまだ最も基本的な機能のみを提供しますが）。 したがって、彼らの助けを借りて、 デモアプリケーションの作業中に、次の情報を取得できます。

  mbox_repository / named_mbox.count-> 1
 coop_repository / coop.reg.count-> 20
 coop_repository / coop.dereg.count-> 0
 coop_repository / agent.count-> 20
 coop_repository / coop.final.dereg.count-> 0
 timer_thread / single_shot.count-> 0
 timer_thread / periodic.count-> 1
 disp / ot / DEFAULT / agent.count-> 3
 disp / ot / DEFAULT / wt-0 / demands.count-> 8
 disp / tp /アナライザー/ threads.count-> 4
 disp / tp / analysers / agent.count-> 16
 disp / tp /アナライザー/ cq / __ so5_au ... 109 __ / agent.count-> 1
 disp / tp /アナライザー/ cq / __ so5_au ... 109 __ / demands.count-> 0
 disp / tp /アナライザー/ cq / __ so5_au ... 124 __ / agent.count-> 1
 disp / tp /アナライザー/ cq / __ so5_au ... 124 __ / demands.count-> 0
 ...
 disp / tp /アナライザー/ cq / __ so5_au ..._ 94 __ / agent.count-> 1
 disp / tp /アナライザー/ cq / __ so5_au ..._ 94 __ / demands.count-> 0
 disp / ot / req_initiator / agent.count-> 1
 disp / ot / req_initiator / wt-0 / demands.count-> 0 

この監視情報の枯渇により、4つのワークフローが機能する「アナライザー」と呼ばれるスレッドプールを持つディスパッチャーがあることを理解できます。 このディスパッチャでは、email_analyzerエージェントがこの例で機能します。 16のエージェントがディスパッチャに接続されており、それぞれが個別の協力を構成しています。 そして、これらのエージェントにはアプリケーションがありません。 つまり、エージェントは存在しますが、彼らのための仕事はありません。 そして、これはなぜこれが起こったのかを理解する機会です。



明らかに、SObjectizer Environmentで利用可能な低レベルの情報は、アプリケーションプログラマーにとって常に役立つものではありません。 議論中の例では、email_analyzerエージェントの数のカウンターと、analyzer_managerエージェント内のリクエストのリストの長さのサイズは、開発者により多くの利益を与えることができます。 しかし、これは適用されたデータであり、SObjectizerはそれについて何も知りません。 したがって、エージェントでアプリケーションを開発する場合、プログラマーは、アプリケーションの正常性と実行可能性を評価するのに最も役立つ、アプリケーション外部の情報が利用可能であることを確認する必要があります。 これはすでに別の議論の大きなトピックですが。



おそらく、これはあなたが終えることができる次の記事です。 次の記事では、電子メール分析操作をさらに並列化しようとした場合に何ができるかを示します。 たとえば、ヘッダー、レターの本文、添付ファイルの並行分析を実行する場合。 そして、email_analyzerエージェントのコードは何になりますか。



記事に示されている例のソースコードは、 このリポジトリにあります 。