forked from taskflow/taskflow
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdynamic_traversal.cpp
More file actions
148 lines (123 loc) · 3.41 KB
/
dynamic_traversal.cpp
File metadata and controls
148 lines (123 loc) · 3.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
// 2019/02/15 - modified by Tsung-Wei Huang
// - refactored the code
//
// 2019/01/03 - created by Chun-Xun Lin
// - use dynamic tasking to implement graph traversal
#include <taskflow/taskflow.hpp>
#include <random>
#include <chrono>
#include <cstring>
struct Node {
size_t level {0};
bool visited {false};
std::atomic<size_t> dependents {0};
std::vector<Node*> successors;
void precede(Node& n) {
successors.emplace_back(&n);
n.dependents ++;
}
};
void traverse(Node* n, tf::SubflowBuilder& subflow) {
assert(!n->visited);
n->visited = true;
for(size_t i=0; i<n->successors.size(); i++) {
if(--(n->successors[i]->dependents) == 0) {
n->successors[i]->level = n->level + 1;
subflow.emplace([s=n->successors[i]](tf::SubflowBuilder &subflow){
traverse(s, subflow);
});
}
}
}
void sequential_traversal(std::vector<Node*>& src) {
auto start = std::chrono::system_clock::now();
while(!src.empty()) {
auto n = src.back();
assert(!n->visited);
n->visited = true;
src.pop_back();
for(auto& s: n->successors) {
if(--s->dependents == 0) {
s->level = n->level + 1;
src.emplace_back(s);
}
}
}
auto end = std::chrono::system_clock::now();
std::cout << "Seq runtime: "
<< std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count()
<< '\n';
}
void tf_traversal(std::vector<Node*>& src) {
auto start = std::chrono::system_clock::now();
tf::Taskflow tf;
for(size_t i=0; i<src.size(); i++) {
tf.emplace([i=i, &src](auto& subflow){ traverse(src[i], subflow); });
}
tf::Executor().run(tf); // block until finished
auto end = std::chrono::system_clock::now();
std::cout << "Tf runtime: "
<< std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count()
<< '\n';
}
int main(int argc, char* argv[]){
enum class Mode {TF, SEQ};
Mode mode {Mode::TF};
bool fully_connected {false};
for(int i=0; i<argc; i++) {
if(::strcmp(argv[i], "full") == 0) {
fully_connected = true;
}
if(::strcmp(argv[i], "tf") == 0) {
mode = Mode::TF;
}
if(::strcmp(argv[i], "seq") == 0) {
mode = Mode::SEQ;
}
}
size_t max_degree {4};
size_t num_nodes {1000000};
Node* nodes = new Node[num_nodes];
// A lambda to verify all nodes are visited
auto validate = [&nodes, &num_nodes](){
for(size_t i=0; i<num_nodes; i++) {
assert(nodes[i].visited);
assert(nodes[i].dependents == 0);
}
};
// Make sure nodes are in clean state
for(size_t i=0; i<num_nodes; i++) {
assert(!nodes[i].visited);
assert(nodes[i].successors.empty());
assert(nodes[i].dependents == 0);
}
// Create a DAG
for(size_t i=0; i<num_nodes; i++) {
size_t degree {0};
for(size_t j=i+1; j<num_nodes && degree < max_degree; j++) {
if(fully_connected || rand()%2 == 1) {
nodes[i].precede(nodes[j]);
if(!fully_connected) {
degree ++;
}
}
}
}
// Find source nodes
std::vector<Node*> src;
for(size_t i=0; i<num_nodes; i++) {
if(!fully_connected) {
assert(nodes[i].successors.size() <= max_degree);
}
if(nodes[i].dependents == 0) {
src.emplace_back(&nodes[i]);
}
}
switch(mode) {
case Mode::TF: tf_traversal(src); break;
case Mode::SEQ: sequential_traversal(src); break;
};
validate();
delete[] nodes;
return 0;
}