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

SObjectizerに関する次の記事では、開発が進むにつれてますます複雑になる、最初は単純なエージェントの進化を引き続き監視します。 関心がなくなった保留中のメッセージの処理方法を検討します。 そして、階層的な有限状態マシンのいくつかの機能を使用します。

前回の記事で 、エージェントemail_analyzerが存在するという事実に決着しました。このエージェントは、その問題を多かれ少なかれ確実に解決できると考えることができます。 ただし、彼自身は、電子メールをチェックする3つの段階を順番に実行します。最初にヘッダーをチェックし、次にコンテンツ、次に添付ファイルをチェックします。

ほとんどの場合、これらの各操作はCPUバウンドではありません。 チェックされたフラグメント（メッセージヘッダーなど）から一部の値を分離する可能性が非常に高いため、この値の有効性をチェックするためにどこかでリクエストを行う必要があります。 たとえば、送信者のホスト名がブラックリストに載っているかどうかを確認するためのデータベース内のクエリ。 このリクエストは実行されますが、たとえば、メッセージのテキストの内容を個別のキーフレーズに解析して、何らかの種類のスパムマーカーの辞書を使用してチェックできるようにするなど、他の操作を実行することも可能です。 または、添付ファイルにアーカイブがあるかどうかを確認し、ウイルス対策スキャンを開始します。 一般に、電子メール分析操作を並列化することは理にかなっています。

操作ごとに別々のエージェントを使用してみましょう。 つまり 次の形式のエージェントを作成できます。

class email_headers_checker : public agent_t { public : struct result { check_status status_ }; /*    */ email_headers_checker( context_t ctx, ... /* -  */ ) {...} virtual void so_evt_start() override { ... /*      */ } ... /* -   */ }; class email_body_checker : public agent_t {...}; class email_attachment_checker : public agent_t {...};
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そのような各エージェントは、その操作に固有のアクションを実行し、email_analyzerの結果をメッセージとして送信します。 email_analyzerは、これらのエージェントのインスタンスを自宅で作成し、分析結果を含むそれらからのメッセージを待つ必要があります。

 void on_load_succeed( const load_email_succeed & msg ) { try { auto parsed_data = parse_email( msg.content_ ); introduce_child_coop( *this, // -checker-      // thread-pool-,     //   . disp::thread_pool::create_disp_binder( "checkers", disp::thread_pool::bind_params_t{} ), [&]( coop_t & coop ) { coop.make_agent< email_headers_checker >( so_direct_mbox(), parsed_data->headers() ); coop.make_agent< email_body_checker >( so_direct_mbox(), parsed_data->body() ); coop.make_agent< email_attach_checker >( so_direct_mbox(), parsed_data->attachments() ); } ); } catch( const exception & ) {...} }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

以前の記事を注意深く読んだ人は、「彼らからのメッセージを待つ」というフレーズに注意する必要があります。 時間制限なしで待機するのは良くありません;それはシステムにぶら下がって何もしないエージェントを得る直接的な方法です。 したがって、チェッカーからの回答を待つとき、IO操作の結果を待つときと同じことを行うのは理にかなっています：何らかの種類の保留中の信号を自分自身に送信し、それを受信すると、それ以上待機する意味がないことがわかります。 つまり 次のようなものを書く必要があります。

 //    email_analyzer    . class email_analyzer : public agent_t { //     ,    //   IO-    . struct io_agent_response_timeout : public signal_t {}; //     ,    //     email-. struct checkers_responses_timeout : public signal_t {}; ... virtual void so_evt_start() override { ... /*   IO- */ //      -    IO-. send_delayed< io_agent_response_timeout >( *this, 1500ms ); } ... void on_load_succeed( const load_succeed & msg ) { ... /*     checker- */ //     -    -checker-. send_delayed< checkers_responses_timeout >( *this, 750ms ); } ... void on_checkers_responses_timeout() { ... /*   . */ } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ただし、このパスに従って、レーキを踏むことになります。チェッカーからの応答を待って、保留中のシグナルio_agent_response_timeoutを簡単に受信できます。 結局、誰もそれをキャンセルしていません。 そして、この信号が到着すると、おそらく存在しないI / Oタイムアウトが存在するため、否定的な応答を生成します。 このレーキを回避してみましょう。

多くの場合、非同期メッセージングに慣れていない開発者は、保留中の信号をキャンセルしようとします。 これは、send_periodicを参照するときにタイマー識別子を保存する場合に実行できます。

 //    email_analyzer    //  io_agent_response_timeout. class email_analyzer : public agent_t { struct io_agent_response_timeout : public signal_t {}; ... virtual void so_evt_start() override { ... /*   IO- */ //  ,      // send_periodic  send_delayed,   period //   0,     , //   . io_response_timer_ = send_periodic< io_agent_response_timeout >( *this, 1500ms, 0ms ); } ... void on_load_succeed( const load_succeed & msg ) { //   . io_response_timer_.reset(); ... /*     checker- */ //     -    -checker-. send_delayed< checkers_responses_timeout >( *this, 750ms ); } ... //       -  IO-. timer_id_t io_response_timer_; };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

残念ながら、この単純な方法は常に機能するとは限りません。 問題は、email_analyzerエージェントがこの遅延信号のタイマーをリセットする少し前に、遅延信号をemail_analyzerエージェントに文字通り送信できることです。 何もする必要はありません-マルチスレッドの驚異、そうです。

email_analyzerエージェントは、その作業スレッドのコンテキストでon_load_succeedにアクセスでき、タイマーのreset（）呼び出しを入力することさえできます...しかし、その後、強制的に削除され、制御スレッドはSObjectizerタイマースレッドを受信し、遅延信号が送信されます。 その後、email_analyzer（）エージェントの作業スレッドが再び制御を受け取り、タイマーのreset（）メソッドが既に送信されたシグナルをキャンセルします。 ただし、信号はすでにエージェントのメッセージキューにあり、誰からも送信されません。メッセージはエージェントへのキューに到着したため、そこから削除することはできません。

この状況で最悪なのは、同様のエラーがときどき発生することです。 何のために、何が正確に起こっており、何が正確にエラーであるかを理解することは困難です。 したがって、保留中のメッセージをキャンセルしても、送信されないという保証はありません。

したがって、保留中のメッセージを誤ってキャンセルした場合、どうすればよいですか？

たとえば、エージェントの状態を使用できます。 email_analyzerがIOエージェントからの応答を待っているとき、それは1つの状態にあります。 IOエージェントからの応答が到着すると、email_analyzerエージェントは異なる状態になり、チェッカーからの応答を待ちます。 なぜなら 2番目の状態では、email_analyzerはio_agent_response_timeoutシグナルに署名されていないため、このシグナルは単に無視されます。

email_analyzerエージェントに状態を導入すると、次のようになります。

 //    email_analyzer   //  . class email_analyzer : public agent_t { struct io_agent_response_timeout : public signal_t {}; struct checkers_responses_timeout : public signal_t {}; // ,       IO-. state_t st_wait_io{ this }; // ,        checker-. state_t st_wait_checkers{ this }; ... virtual void so_define_agent() override { //        . //  ,    ,    //   –    state_t. st_wait_io .event( &email_analyzer::on_load_succeed ) .event( &email_analyzer::on_load_failed ) .event< io_agent_response_timeout >( &email_analyzer::on_io_timeout ); st_wait_checkers .event( &email_analyzer::on_header_check_result ) .event( &email_analyzer::on_body_check_result ) .event( &email_analyzer::on_attach_check_result ) .event< checkers_responses_timeout >( &email_analyzer::on_checkers_timeout ); } ... };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ただし、SObjectizerを使用するとさらに簡単に実行できます。エージェントが特定の状態に留まるように時間制限を設定できます。 この制限が切れると、エージェントは強制的に別の状態になります。 つまり 次のように書くことができます。

 //    email_analyzer     //      . class email_analyzer : public agent_t { state_t st_wait_io{ this }; state_t st_io_timeout{ this }; state_t st_wait_checkers{ this }; state_t st_checkers_timeout{ this }; ... virtual void so_define_agent() override { st_wait_io .event( &email_analyzer::on_load_succeed ) .event( &email_analyzer::on_load_failed ) //   . .time_limit( 1500ms, st_io_timeout ); st_wait_checkers .event( &email_analyzer::on_header_check_result ) .event( &email_analyzer::on_body_check_result ) .event( &email_analyzer::on_attach_check_result ) .time_limit( 750ms, st_checkers_timeout ); } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

しかし、ある状態で費やす時間を制限するだけでは十分ではありません。 この時間が経過したら、まだ何らかのアクションをとる必要があります。 どうやってやるの？

状態エントリハンドラのようなものを使用します。 エージェントが特定の状態に入ると、ユーザーがそのような機能を割り当てた場合、SObjectizerはこの状態に入るハンドラー関数を呼び出します。 これは、ハンドラーを切断してst_io_timeoutを入力できることを意味します。これにより、負の結果のcheck_resultが送信され、エージェントが終了します。

 st_io_timeout.on_enter( [this]{ send< check_result >( reply_to_, email_file_, check_status::check_failure ); so_deregister_agent_coop_normally(); } );
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

st_checkers_timeoutへの入力でまったく同じハンドラーを切断します。 そして以来 これらのハンドラー内のアクションは同じであるため、別のemail_analyzerエージェントメソッドに入れて、このメソッドをst_io_timeout状態とst_checkers_timeout状態の両方の入力ハンドラーとして指定できます。

 class email_analyzer : public agent_t { state_t st_wait_io{ this }; state_t st_io_timeout{ this }; state_t st_wait_checkers{ this }; state_t st_checkers_timeout{ this }; ... virtual void so_define_agent() override { ... st_io_timeout .on_enter( &email_analyzer::on_enter_timeout_state ); ... st_checkers_timeout .on_enter( &email_analyzer::on_enter_timeout_state ); }; ... void on_enter_timeout_state() { send< check_result >( reply_to_, email_file_, check_status::check_failure ); so_deregister_agent_coop_normally(); } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

しかし、それだけではありません。 エージェントの状態とその機能のトピックに触れたので、さらに開発してemail_analyzerコードをリファクタリングできます。

コードでは、check_resultメッセージの送信とエージェントの協力の登録解除という2つのアクションがしばしば重複していることに気付くのは簡単です。 そのような複製は良くありません;それを取り除くべきです。

実際、email_analyzerエージェントの作業は、エージェントが次の2つの状態のいずれかになることを保証するために削減されます：すべてが正常に終了し、肯定的な結果が送信された後、作業を​​完了するか、またはすべてがエラーで終了した場合、否定的な結果を送信する必要があります同じように、エージェントをシャットダウンします。 そのため、2つのエージェント状態st_successおよびst_failureを使用して、コードでこれを直接表現しましょう。

 //    email_analyzer    //  st_success  st_failure. class email_analyzer : public agent_t { state_t st_wait_io{ this }; state_t st_wait_checkers{ this }; state_t st_failure{ this }; state_t st_success{ this }; ... virtual void so_define_agent() override { st_wait_io .event( &email_analyzer::on_load_succeed ) .event( &email_analyzer::on_load_failed ) //   . .time_limit( 1500ms, st_failure ); st_wait_checkers .event( &email_analyzer::on_header_check_result ) .event( &email_analyzer::on_body_check_result ) .event( &email_analyzer::on_attach_check_result ) .time_limit( 750ms, st_failure ); st_failure .on_enter( [this]{ send< check_result >( reply_to_, email_file_, status_ ); so_deregister_agent_coop_normally(); } ); st_success .on_enter( [this]{ send< check_result >( reply_to_, email_file_, check_status::safe ); so_deregister_agent_coop_normally(); } ); }; ... //        . check_status status_{ check_status::check_failure }; };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これにより、エージェントコードの状態を変更するだけで、何らかの方法でエージェントをシャットダウンできます。

 void on_load_failed( const load_email_failed & ) { st_failure.activate(); } void on_checker_result( check_status status ) { //        . if( check_status::safe != status ) { status_ = status; st_failure.activate(); } else { ++checks_passed_; if( 3 == checks_passed_ ) //   .     //  . st_success.activate(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

しかし、さらに先へ進むことができます。 st_failureおよびst_success状態の場合、これらの状態のいずれかを入力するときに実行する必要がある共通アクションが1つあります-so_deregister_agent_coop_normally（）を呼び出します。 また、これらの状態の両方がエージェントをシャットダウンする責任があるため、これは偶然ではありません。 その場合、ネストされた状態を使用できます。 つまり st_finiure状態を導入します。st_failureおよびst_successはサブ状態になります。 st_finishingを入力すると、so_deregister_agent_coop_normally（）が呼び出されます。 また、st_failureとst_successを入力すると、対応するメッセージのみが送信されます。

なぜなら st_failure状態とst_success状態はst_finishingにネストされているため、これらのいずれかを入力すると、st_finishingのエントリハンドラーが最初に呼び出され、その後でのみst_failureまたはst_successのエントリハンドラーが呼び出されます。 st_finishingを入力するとエージェントの登録が解除され、st_failureまたはst_successを入力するとcheck_resultメッセージが送信されます。

ネストされた状態、状態エントリハンドラ、状態で費やされる時間の制限について言及するときに読者の1人が不安を感じる場合、階層型有限状態マシンのトピックに関する基本的な記事の1つを読むのが理にかなっています： David Harel、Statecharts：A visual formatism複雑なシステム用。 コンピュータプログラミングの科学 。 SObjectizerのエージェント状態は、そこに記述されているかなりの量の機能を実装しています。

これらすべての変換の結果として、email_analyzerエージェントは以下に示す形式を取ります。

 //    email_analyzer,      //  email-    . class email_analyzer : public agent_t { state_t st_wait_io{ this }; state_t st_wait_checkers{ this }; state_t st_finishing{ this }; state_t st_failure{ initial_substate_of{ st_finishing } }; state_t st_success{ substate_of{ st_finishing } }; 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 { st_wait_io .event( &email_analyzer::on_load_succeed ) .event( &email_analyzer::on_load_failed ) //  -   . .time_limit( 1500ms, st_failure ); st_wait_checkers .event( [this]( const email_headers_checker::result & msg ) { on_checker_result( msg.status_ ); } ) .event( [this]( const email_body_checker::result & msg ) { on_checker_result( msg.status_ ); } ) .event( [this]( const email_attach_checker::result & msg ) { on_checker_result( msg.status_ ); } ) //   -  . .time_limit( 750ms, st_failure ); //  ,     , //     . st_finishing.on_enter( [this]{ so_deregister_agent_coop_normally(); } ); st_failure.on_enter( [this]{ send< check_result >( reply_to_, email_file_, status_ ); } ); st_success.on_enter( [this]{ send< check_result >( reply_to_, email_file_, check_status::safe ); } ); } virtual void so_evt_start() override { //      ,  //      . st_wait_io.activate(); //       IO-   //  email . send< load_email_request >( so_environment().create_mbox( "io_agent" ), email_file_, so_direct_mbox() ); } private : const string email_file_; const mbox_t reply_to_; //      ,   //      st_failure. check_status status_{ check_status::check_failure }; int checks_passed_{}; void on_load_succeed( const load_email_succeed & msg ) { //   ..    . st_wait_checkers.activate(); try { auto parsed_data = parse_email( msg.content_ ); introduce_child_coop( *this, // -checker-      // thread-pool-,     //   . disp::thread_pool::create_disp_binder( "checkers", disp::thread_pool::bind_params_t{} ), [&]( coop_t & coop ) { coop.make_agent< email_headers_checker >( so_direct_mbox(), parsed_data->headers() ); coop.make_agent< email_body_checker >( so_direct_mbox(), parsed_data->body() ); coop.make_agent< email_attach_checker >( so_direct_mbox(), parsed_data->attachments() ); } ); } catch( const exception & ) { st_failure.activate(); } } void on_load_failed( const load_email_failed & ) { st_failure.activate(); } void on_checker_result( check_status status ) { //        . if( check_status::safe != status ) { status_ = status; st_failure.activate(); } else { ++checks_passed_; if( 3 == checks_passed_ ) //   .     //  . st_success.activate(); } } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

さて、結果のemail_analyzerエージェントのコードを見て、単純だが重要な質問を自問するのは理にかなっています。

明らかに、この質問に対する答えはそれほど単純ではありません。 しかし、これについては次の記事で説明します。 ここでは、ソフトウェアシステムの開発にSObjectizerを10年以上使用した後に学んだ教訓のトピックを取り上げます。

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