I have need for a task scheduler determined by a directed graph. The tasks are held in a std::vector<task_type>
, while the dependency graph is held in a adjacency_list<vecS, vecS, bidirectionalS>
(bidirectionalS
so that I have access to the in_degree()
function). A single dispatch will start the tasks. Are there obvious improvements here?
To set up a task dependency graph as below, use the driver provided.
scheduler_driver.hpp:
#ifndef __SCHEDULER_DRIVER_HPP__
#define __SCHEDULER_DRIVER_HPP__
#include <iostream>
#include <ostream>
#include <iterator>
#include <vector>
#include <chrono>
#include "scheduler.h"
#endif
scheduler_driver.cpp:
#include "scheduler_driver.hpp"
enum task_nodes
{
task_0,
task_1,
task_2,
task_3,
task_4,
task_5,
task_6,
task_7,
task_8,
task_9,
N
};
int basic_task(int a, int d)
{
std::chrono::milliseconds sleepDuration(d);
std::this_thread::sleep_for(sleepDuration);
std::cout << "Result: " << a << "\n";
return a;
}
using namespace SCHEDULER;
int main(int argc, char **argv)
{
typedef int R;
typedef std::function<R()> F;
Graph deps(N);
boost::add_edge(task_0, task_1, deps);
boost::add_edge(task_0, task_2, deps);
boost::add_edge(task_0, task_3, deps);
boost::add_edge(task_1, task_4, deps);
boost::add_edge(task_1, task_5, deps);
boost::add_edge(task_1, task_6, deps);
boost::add_edge(task_2, task_7, deps);
boost::add_edge(task_2, task_8, deps);
boost::add_edge(task_2, task_9, deps);
std::vector<F> tasks =
{
std::bind(basic_task, 0, 1000),
std::bind(basic_task, 1, 1000),
std::bind(basic_task, 2, 1000),
std::bind(basic_task, 3, 1000),
std::bind(basic_task, 4, 1000),
std::bind(basic_task, 5, 1000),
std::bind(basic_task, 6, 1000),
std::bind(basic_task, 7, 1000),
std::bind(basic_task, 8, 1000),
std::bind(basic_task, 9, 1000)
};
scheduler<R> *s = new scheduler<R>(std::move(deps), std::move(tasks));
s->doit();
return 0;
}
scheduler.h:
#ifndef __SCHEDULER2_H__
#define __SCHEDULER2_H__
#include <iostream>
#include <vector>
#include <iterator>
#include <functional>
#include <algorithm>
#include <mutex>
#include <thread>
#include <future>
#include <boost/graph/graph_traits.hpp>
#include <boost/graph/adjacency_list.hpp>
#include <boost/graph/depth_first_search.hpp>
#include <boost/graph/visitors.hpp>
using namespace boost;
namespace SCHEDULER
{
using Graph = adjacency_list<vecS, vecS, bidirectionalS>;
using Edge = graph_traits<Graph>::edge_descriptor;
using Vertex = graph_traits<Graph>::vertex_descriptor;
using VectexCont = std::vector<Vertex>;
using outIt = graph_traits<Graph>::out_edge_iterator;
using inIt = graph_traits<Graph>::in_edge_iterator;
template<typename R>
class scheduler
{
public:
using ret_type = R;
using fun_type = std::function<R()>;
using prom_type = std::promise<ret_type>;
using fut_type = std::shared_future<ret_type>;
scheduler() = default;
scheduler(const Graph &deps_, const std::vector<fun_type> &tasks_) :
g(deps_),
tasks(tasks_) { init_();}
scheduler(Graph&& deps_, std::vector<fun_type>&& tasks_) :
g(std::move(deps_)),
tasks(std::move(tasks_)) { init_(); }
scheduler(const scheduler&) = delete;
scheduler& operator=(const scheduler&) = delete;
void doit();
private:
void init_();
std::list<Vertex> get_sources(const Vertex& v);
auto task_thread(fun_type&& f, int i);
Graph g;
std::vector<fun_type> tasks;
std::vector<prom_type> prom;
std::vector<fut_type> fut;
std::vector<std::thread> th;
std::vector<std::list<Vertex>> sources;
};
template<typename R>
void
scheduler<R>::init_()
{
int num_tasks = tasks.size();
prom.resize(num_tasks);
fut.resize(num_tasks);
// Get the futures
for(size_t i=0;
i<num_tasks;
++i)
{
fut[i] = prom[i].get_future();
}
// Predetermine in_edges for faster traversal
sources.resize(num_tasks);
for(size_t i=0;
i<num_tasks;
++i)
{
sources[i] = get_sources(i);
}
}
template<typename R>
std::list<Vertex>
scheduler<R>::get_sources(const Vertex& v)
{
std::list<Vertex> r;
Vertex v1;
inIt j, j_end;
boost::tie(j,j_end) = in_edges(v, g);
for(;j != j_end;++j)
{
v1 = source(*j, g);
r.push_back(v1);
}
return r;
}
template<typename R>
auto
scheduler<R>::task_thread(fun_type&& f, int i)
{
auto j_beg = sources[i].begin(),
j_end = sources[i].end();
for(;
j_beg != j_end;
++j_beg)
{
R val = fut[*j_beg].get();
}
return std::thread([this](fun_type f, int i)
{
prom[i].set_value(f());
},f,i);
}
template<typename R>
void
scheduler<R>::doit()
{
size_t num_tasks = tasks.size();
th.resize(num_tasks);
for(int i=0;
i<num_tasks;
++i)
{
th[i] = task_thread(std::move(tasks[i]), i);
}
for_each(th.begin(), th.end(), mem_fn(&std::thread::join));
}
} // namespace SCHEDULER
#endif