方法: スケジュールグループを使用して実行順序に影響を与える

[アーティクル]
06/16/2023

コンカレンシーランタイムでは、タスクをスケジュールする順序は非確定的です。ただし、スケジューリングポリシーを使用して、タスクの実行順序に影響を与えることができます。ここでは、concurrency::SchedulingProtocol スケジューラポリシーと共にスケジュールグループを使用して、タスクの実行順序に影響を与える方法について説明します。

例では、一連のタスクを 2 回、それぞれ異なるスケジューリングポリシーを使用して実行します。どちらのポリシーでも、処理リソースの最大数は 2 つに制限されます。最初の実行では EnhanceScheduleGroupLocality ポリシー (既定) を使用し、2 回目の実行では EnhanceForwardProgress ポリシーを使用します。 EnhanceScheduleGroupLocality ポリシーでは、1 つのスケジュールグループ内の各タスクが終了または生成するまですべてのタスクを実行します。 EnhanceForwardProgress ポリシーでは、1 つのタスクが終了または生成すると、ラウンドロビン方式で次のスケジュールグループに移動します。

各スケジュールグループに関連するタスクが含まれている場合、EnhanceScheduleGroupLocality ポリシーを使用すると、通常、タスク間でキャッシュの局所性が保持されるため、パフォーマンスが向上します。 EnhanceForwardProgress ポリシーを使用すると、タスクを次に進めることができるため、スケジュールグループ間で公平にスケジュールを設定する必要がある場合に便利です。

例

この例では、concurrency::agent から派生する work_yield_agent クラスを定義します。 work_yield_agent クラスでは、1 つの作業単位を実行し、現在のコンテキストを生成した後、別の作業単位を実行します。エージェントは、他のコンテキストが実行できるように、concurrency::wait 関数を使用して、現在のコンテキストを協調的に譲渡します。

この例では、work_yield_agent オブジェクトを 4 つ作成します。スケジューラポリシーを設定してエージェントの実行順序に影響を与える方法がわかるように、最初の 2 つのエージェントと残りの 2 つのエージェントを別々のスケジュールグループに関連付けます。また、concurrency::CurrentScheduler::CreateScheduleGroup メソッドを使用して、concurrency::ScheduleGroup オブジェクトを作成します。例では、4 つすべてのエージェントを 2 回実行します。それぞれの実行で、異なるスケジューリングポリシーを使用します。

// scheduling-protocol.cpp
// compile with: /EHsc
#include <agents.h>
#include <vector>
#include <algorithm>
#include <iostream>
#include <sstream>

using namespace concurrency;
using namespace std;

#pragma optimize( "", off )
// Simulates work by performing a long spin loop.
void spin_loop()
{
   for (int i = 0; i < 500000000; ++i)
   {
   }
}
#pragma optimize( "", on )

// Agent that performs some work and then yields the current context.
class work_yield_agent : public agent
{
public:
   explicit work_yield_agent(
      unsigned int group_number, unsigned int task_number)
      : _group_number(group_number)
      , _task_number(task_number)
   {
   }

   explicit work_yield_agent(Scheduler& scheduler,
      unsigned int group_number, unsigned int task_number)
      : agent(scheduler)
      , _group_number(group_number)
      , _task_number(task_number)
   {
   }

   explicit work_yield_agent(ScheduleGroup& group,
      unsigned int group_number, unsigned int task_number)
      : agent(group)       
      , _group_number(group_number)
      , _task_number(task_number)
   {
   }
   
protected:
   // Performs the work of the agent.   
   void run()
   {
      wstringstream header, ss;

      // Create a string that is prepended to each message.
      header << L"group " << _group_number 
             << L",task " << _task_number << L": ";

      // Perform work.
      ss << header.str() << L"first loop..." << endl;
      wcout << ss.str();
      spin_loop();

      // Cooperatively yield the current context. 
      // The task scheduler will then run all blocked contexts.
      ss = wstringstream();
      ss << header.str() << L"waiting..." << endl;
      wcout << ss.str();
      concurrency::wait(0);

      // Perform more work.
      ss = wstringstream();
      ss << header.str() << L"second loop..." << endl;
      wcout << ss.str();
      spin_loop();

      // Print a final message and then set the agent to the 
      // finished state.
      ss = wstringstream();
      ss << header.str() << L"finished..." << endl;
      wcout << ss.str();

      done();
   }  

private:
   // The group number that the agent belongs to.
   unsigned int _group_number;
   // A task number that is associated with the agent.
   unsigned int _task_number;
};

// Creates and runs several groups of agents. Each group of agents is associated 
// with a different schedule group.
void run_agents()
{
   // The number of schedule groups to create.
   const unsigned int group_count = 2;
   // The number of agent to create per schedule group.
   const unsigned int tasks_per_group = 2;

   // A collection of schedule groups.
   vector<ScheduleGroup*> groups;
   // A collection of agents.
   vector<agent*> agents;

   // Create a series of schedule groups. 
   for (unsigned int group = 0; group < group_count; ++group)
   {
      groups.push_back(CurrentScheduler::CreateScheduleGroup());

      // For each schedule group, create a series of agents.
      for (unsigned int task = 0; task < tasks_per_group; ++task)
      {
         // Add an agent to the collection. Pass the current schedule 
         // group to the work_yield_agent constructor to schedule the agent
         // in this group.
         agents.push_back(new work_yield_agent(*groups.back(), group, task));
      }
   }

   // Start each agent.
   for_each(begin(agents), end(agents), [](agent* a) {
      a->start();
   });

   // Wait for all agents to finsih.
   agent::wait_for_all(agents.size(), &agents[0]);

   // Free the memory that was allocated for each agent.
   for_each(begin(agents), end(agents), [](agent* a) {
      delete a;
   });

   // Release each schedule group.
   for_each(begin(groups), end(groups), [](ScheduleGroup* group) {
      group->Release();
   });
}

int wmain()
{
   // Run the agents two times. Each run uses a scheduler
   // policy that limits the maximum number of processing resources to two.

   // The first run uses the EnhanceScheduleGroupLocality 
   // scheduling protocol. 
   wcout << L"Using EnhanceScheduleGroupLocality..." << endl;
   CurrentScheduler::Create(SchedulerPolicy(3, 
      MinConcurrency, 1,
      MaxConcurrency, 2,
      SchedulingProtocol, EnhanceScheduleGroupLocality));

   run_agents();
   CurrentScheduler::Detach();

   wcout << endl << endl;

   // The second run uses the EnhanceForwardProgress 
   // scheduling protocol. 
   wcout << L"Using EnhanceForwardProgress..." << endl;
   CurrentScheduler::Create(SchedulerPolicy(3, 
      MinConcurrency, 1,
      MaxConcurrency, 2,
      SchedulingProtocol, EnhanceForwardProgress));

   run_agents();
   CurrentScheduler::Detach();
}

この例を実行すると、次の出力が生成されます。

Using EnhanceScheduleGroupLocality...
group 0,
    task 0: first loop...
group 0,
    task 1: first loop...
group 0,
    task 0: waiting...
group 1,
    task 0: first loop...
group 0,
    task 1: waiting...
group 1,
    task 1: first loop...
group 1,
    task 0: waiting...
group 0,
    task 0: second loop...
group 1,
    task 1: waiting...
group 0,
    task 1: second loop...
group 0,
    task 0: finished...
group 1,
    task 0: second loop...
group 0,
    task 1: finished...
group 1,
    task 1: second loop...
group 1,
    task 0: finished...
group 1,
    task 1: finished...

Using EnhanceForwardProgress...
group 0,
    task 0: first loop...
group 1,
    task 0: first loop...
group 0,
    task 0: waiting...
group 0,
    task 1: first loop...
group 1,
    task 0: waiting...
group 1,
    task 1: first loop...
group 0,
    task 1: waiting...
group 0,
    task 0: second loop...
group 1,
    task 1: waiting...
group 1,
    task 0: second loop...
group 0,
    task 0: finished...
group 0,
    task 1: second loop...
group 1,
    task 0: finished...
group 1,
    task 1: second loop...
group 0,
    task 1: finished...
group 1,
    task 1: finished...

どちらのポリシーでも、イベントの順序は同じになります。ただし、EnhanceScheduleGroupLocality を使用するポリシーでは、最初のスケジュールグループに属するエージェントを両方開始した後に、2 番目のグループに属するエージェントを開始します。 EnhanceForwardProgress を使用するポリシーでは、最初のスケジュールグループのエージェントを 1 つ開始した後、2 番目のグループの最初のエージェントを開始します。

コードのコンパイル

コード例をコピーし、Visual Studio プロジェクトに貼り付けるか、scheduling-protocol.cpp という名前のファイルに貼り付けてから、Visual Studio のコマンドプロンプトウィンドウで次のコマンドを実行します。

cl.exe /EHsc scheduling-protocol.cpp

次の方法で共有

方法: スケジュールグループを使用して実行順序に影響を与える

例

コードのコンパイル

関連項目

フィードバック

その他のリソース

次の方法で共有

方法: スケジュール グループを使用して実行順序に影響を与える

例

コードのコンパイル

関連項目

フィードバック

その他のリソース

方法: スケジュールグループを使用して実行順序に影響を与える