How to: Use Schedule Groups to Influence Order of Execution
In the Concurrency Runtime, the order in which tasks are scheduled is non-deterministic. However, you can use scheduling policies to influence the order in which tasks run. This topic shows how to use schedule groups together with the concurrency::SchedulingProtocol scheduler policy to influence the order in which tasks run.
The example runs a set of tasks two times, each with a different scheduling policy. Both policies limit the maximum number of processing resources to two. The first run uses the EnhanceScheduleGroupLocality policy, which is the default, and the second run uses the EnhanceForwardProgress policy. Under the EnhanceScheduleGroupLocality policy, the scheduler runs all tasks in one schedule group until each task finishes or yields. Under the EnhanceForwardProgress policy, the scheduler moves to the next schedule group in a round-robin manner after just one task finishes or yields.
When each schedule group contains related tasks, the EnhanceScheduleGroupLocality policy typically results in improved performance because cache locality is preserved between tasks. The EnhanceForwardProgress policy enables tasks to make forward progress and is useful when you require scheduling fairness across schedule groups.
Example
This example defines the work_yield_agent class, which derives from concurrency::agent. The work_yield_agent class performs a unit of work, yields the current context, and then performs another unit of work. The agent uses the concurrency::wait function to cooperatively yield the current context so that other contexts can run.
This example creates four work_yield_agent objects. To illustrate how to set scheduler policies to affect the order in which the agents run, the example associates the first two agents with one schedule group and the other two agents with another schedule group. The example uses the concurrency::CurrentScheduler::CreateScheduleGroup method to create the concurrency::ScheduleGroup objects. The example runs all four agents two times, each time with a different scheduling policy.
// 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();
}
This example produces the following output.
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...
Both policies produce the same sequence of events. However, the policy that uses EnhanceScheduleGroupLocality starts both agents that are part of the first schedule group before it starts the agents that are part of the second group. The policy that uses EnhanceForwardProgress starts one agent from the first group and then starts the first agent in the second group.
Compiling the Code
Copy the example code and paste it in a Visual Studio project, or paste it in a file that is named scheduling-protocol.cpp and then run the following command in a Visual Studio Command Prompt window.
cl.exe /EHsc scheduling-protocol.cpp