florian/mesytec-mnode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
mesytec-mnode/external/taskflow-3.8.0/unittests/test_scalable_pipelines.cpp
Florian Lüke e426fb7628 add taskflow-3.8.0
2025-01-04 01:25:05 +01:00

1388 lines
38 KiB
C++
Raw Permalink Blame History

#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
#include <doctest.h>
#include <taskflow/taskflow.hpp>
#include <taskflow/algorithm/pipeline.hpp>
// ----------------------------------------------------------------------------
// Constructors and Assignments
// ----------------------------------------------------------------------------
TEST_CASE("ScalablePipeline.Basics" * doctest::timeout(300)) {
size_t N = 10;
std::vector< tf::Pipe<std::function<void(tf::Pipeflow&)>> > pipes;
for(size_t i=0; i<N; i++) {
pipes.emplace_back(tf::PipeType::SERIAL, [&](tf::Pipeflow&) {});
}
using iterator_type = decltype(pipes)::iterator;
tf::ScalablePipeline<iterator_type> rhs;
REQUIRE(rhs.num_lines() == 0);
REQUIRE(rhs.num_pipes() == 0);
REQUIRE(rhs.num_tokens() == 0);
rhs.reset(1, pipes.begin(), pipes.end());
REQUIRE(rhs.num_lines() == 1);
REQUIRE(rhs.num_pipes() == N);
REQUIRE(rhs.num_tokens() == 0);
tf::ScalablePipeline<iterator_type> lhs(std::move(rhs));
REQUIRE(rhs.num_lines() == 0);
REQUIRE(rhs.num_pipes() == 0);
REQUIRE(rhs.num_tokens() == 0);
REQUIRE(lhs.num_lines() == 1);
REQUIRE(lhs.num_pipes() == N);
REQUIRE(lhs.num_tokens() == 0);
rhs = std::move(lhs);
REQUIRE(lhs.num_lines() == 0);
REQUIRE(lhs.num_pipes() == 0);
REQUIRE(lhs.num_tokens() == 0);
REQUIRE(rhs.num_lines() == 1);
REQUIRE(rhs.num_pipes() == N);
REQUIRE(rhs.num_tokens() == 0);
}
// ----------------------------------------------------------------------------
// Scalable Pipeline
// ----------------------------------------------------------------------------
void scalable_pipeline(size_t num_lines, size_t num_pipes) {
tf::Executor executor;
tf::Taskflow taskflow;
size_t N = 0;
std::vector< tf::Pipe<std::function<void(tf::Pipeflow&)>> > pipes;
std::vector< int > data(num_lines, -1);
for(size_t i=0; i<num_pipes; i++) {
pipes.emplace_back(tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) mutable {
switch(pf.pipe()) {
case 0:
if(pf.token() == 1111) {
pf.stop();
return;
}
data[pf.line()] = num_pipes * pf.token();
break;
default: {
++data[pf.line()];
}
break;
}
//printf("data[%zu]=%d\n", pf.line(), data[pf.line()]);
REQUIRE(data[pf.line()] == (pf.token() * num_pipes + pf.pipe()));
if(pf.pipe() == num_pipes - 1) {
N++;
}
});
}
tf::ScalablePipeline spl(num_lines, pipes.begin(), pipes.end());
taskflow.composed_of(spl);
executor.run(taskflow).wait();
REQUIRE(N == 1111);
}
TEST_CASE("ScalablePipeline" * doctest::timeout(300)) {
for(size_t L=1; L<=10; L++) {
for(size_t P=1; P<=10; P++) {
scalable_pipeline(L, P);
}
}
}
// ----------------------------------------------------------------------------
// Scalable Pipeline using Reset
// ----------------------------------------------------------------------------
void scalable_pipeline_reset(size_t num_lines, size_t num_pipes) {
tf::Executor executor;
tf::Taskflow taskflow;
size_t N = 0;
std::vector< tf::Pipe<std::function<void(tf::Pipeflow&)>> > pipes;
std::vector< int > data(num_lines, -1);
tf::ScalablePipeline<typename decltype(pipes)::iterator> spl(num_lines);
auto init = taskflow.emplace([&](){
for(size_t i=0; i<num_pipes; i++) {
pipes.emplace_back(tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) mutable {
switch(pf.pipe()) {
case 0:
if(pf.token() == 1111) {
pf.stop();
return;
}
data[pf.line()] = num_pipes * pf.token();
break;
default: {
++data[pf.line()];
}
break;
}
//printf("data[%zu]=%d\n", pf.line(), data[pf.line()]);
REQUIRE(data[pf.line()] == (pf.token() * num_pipes + pf.pipe()));
if(pf.pipe() == num_pipes - 1) {
N++;
}
});
}
spl.reset(pipes.begin(), pipes.end());
});
auto pipeline = taskflow.composed_of(spl);
pipeline.succeed(init);
executor.run(taskflow).wait();
REQUIRE(N == 1111);
}
TEST_CASE("ScalablePipeline.Reset" * doctest::timeout(300)) {
for(size_t L=1; L<=10; L++) {
for(size_t P=1; P<=10; P++) {
scalable_pipeline_reset(L, P);
}
}
}
// ----------------------------------------------------------------------------
// Scalable Pipeline using Iterative Reset
// ----------------------------------------------------------------------------
void scalable_pipeline_iterative_reset(size_t num_lines, size_t num_pipes) {
tf::Executor executor;
tf::Taskflow taskflow;
size_t N = 0;
std::vector< tf::Pipe<std::function<void(tf::Pipeflow&)>> > pipes;
std::vector< int > data(num_lines, -1);
tf::ScalablePipeline<typename decltype(pipes)::iterator> spl(num_lines);
auto init = taskflow.emplace([&](){
for(size_t i=0; i<num_pipes; i++) {
pipes.emplace_back(tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) mutable {
switch(pf.pipe()) {
case 0:
if(pf.token() == 1111) {
pf.stop();
return;
}
data[pf.line()] = num_pipes * pf.token();
break;
default: {
++data[pf.line()];
}
break;
}
//printf("data[%zu]=%d\n", pf.line(), data[pf.line()]);
REQUIRE(data[pf.line()] == (pf.token() * num_pipes + pf.pipe()));
if(pf.pipe() == num_pipes - 1) {
N++;
}
});
}
spl.reset(pipes.begin(), pipes.end());
});
auto cond = taskflow.emplace([&, i=0]()mutable{
REQUIRE(N == 1111*(i+1));
spl.reset();
return (i++ < 3) ? 0 : -1;
});
auto pipeline = taskflow.composed_of(spl);
pipeline.succeed(init)
.precede(cond);
cond.precede(pipeline);
executor.run(taskflow).wait();
}
TEST_CASE("ScalablePipeline.IterativeReset" * doctest::timeout(300)) {
for(size_t L=1; L<=10; L++) {
for(size_t P=1; P<=10; P++) {
scalable_pipeline_iterative_reset(L, P);
}
}
}
// ----------------------------------------------------------------------------
// Scalable Pipeline Reset
//
// reset(num_lines, pipes.begin(), pipes.end())
// ----------------------------------------------------------------------------
void scalable_pipeline_lines_reset(size_t num_lines, size_t num_pipes) {
tf::Executor executor;
size_t N = 0;
std::vector<tf::Pipe<>> pipes;
tf::ScalablePipeline<typename decltype(pipes)::iterator> spl;
for(size_t l = 1; l <= num_lines; ++l) {
tf::Taskflow taskflow;
std::vector<int> data(l, -1);
auto init = taskflow.emplace([&](){
for(size_t i=0; i<num_pipes; i++) {
pipes.emplace_back(tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) mutable {
switch(pf.pipe()) {
case 0:
if(pf.token() == 1111) {
pf.stop();
return;
}
data[pf.line()] = num_pipes * pf.token();
break;
default: {
++data[pf.line()];
}
break;
}
//printf("data[%zu]=%d\n", pf.line(), data[pf.line()]);
REQUIRE(data[pf.line()] == (pf.token() * num_pipes + pf.pipe()));
if(pf.pipe() == num_pipes - 1) {
N++;
}
});
}
spl.reset(l, pipes.begin(), pipes.end());
});
auto check = taskflow.emplace([&]()mutable{
REQUIRE(N == 1111 * l);
pipes.clear();
});
auto pipeline = taskflow.composed_of(spl);
pipeline.succeed(init)
.precede(check);
executor.run(taskflow).wait();
}
}
TEST_CASE("ScalablePipeline.LinesReset" * doctest::timeout(300)) {
for(size_t P=1; P<=10; P++) {
scalable_pipeline_lines_reset(10, P);
}
}
// ----------------------------------------------------------------------------
//
// ifelse ScalablePipeline has three pipes, L lines, w workers
//
// SPS
// ----------------------------------------------------------------------------
int ifelse_spipe_ans(int a) {
// pipe 1
if(a / 2 != 0) {
a += 8;
}
// pipe 2
if(a > 4897) {
a -= 1834;
}
else {
a += 3;
}
// pipe 3
if((a + 9) / 4 < 50) {
a += 1;
}
else {
a += 17;
}
return a;
}
void ifelse_spipeline(size_t L, unsigned w) {
srand(time(NULL));
tf::Executor executor(w);
size_t maxN = 200;
std::vector<int> source(maxN);
for(auto&& s: source) {
s = rand() % 9962;
}
std::vector<std::array<int, 4>> buffer(L);
std::vector<tf::Pipe<>> pipes;
tf::ScalablePipeline<typename decltype(pipes)::iterator> pl;
for(size_t N = 1; N < maxN; ++N) {
tf::Taskflow taskflow;
std::vector<int> collection;
collection.reserve(N);
// pipe 1
pipes.emplace_back(tf::PipeType::SERIAL, [&, N](auto& pf){
if(pf.token() == N) {
pf.stop();
return;
}
if(source[pf.token()] / 2 == 0) {
buffer[pf.line()][pf.pipe()] = source[pf.token()];
}
else {
buffer[pf.line()][pf.pipe()] = source[pf.token()] + 8;
}
});
// pipe 2
pipes.emplace_back(tf::PipeType::PARALLEL, [&](auto& pf){
if(buffer[pf.line()][pf.pipe() - 1] > 4897) {
buffer[pf.line()][pf.pipe()] = buffer[pf.line()][pf.pipe() - 1] - 1834;
}
else {
buffer[pf.line()][pf.pipe()] = buffer[pf.line()][pf.pipe() - 1] + 3;
}
});
// pipe 3
pipes.emplace_back(tf::PipeType::SERIAL, [&](auto& pf){
if((buffer[pf.line()][pf.pipe() - 1] + 9) / 4 < 50) {
buffer[pf.line()][pf.pipe()] = buffer[pf.line()][pf.pipe() - 1] + 1;
}
else {
buffer[pf.line()][pf.pipe()] = buffer[pf.line()][pf.pipe() - 1] + 17;
}
collection.push_back(buffer[pf.line()][pf.pipe()]);
});
pl.reset(L, pipes.begin(), pipes.end());
auto pl_t = taskflow.composed_of(pl).name("pipeline");
auto check_t = taskflow.emplace([&](){
for(size_t n = 0; n < N; ++n) {
REQUIRE(collection[n] == ifelse_spipe_ans(source[n]));
}
}).name("check");
pl_t.precede(check_t);
executor.run(taskflow).wait();
pipes.clear();
}
}
TEST_CASE("ScalablePipeline.Ifelse.1L.1W" * doctest::timeout(300)) {
ifelse_spipeline(1, 1);
}
TEST_CASE("ScalablePipeline.Ifelse.1L.2W" * doctest::timeout(300)) {
ifelse_spipeline(1, 2);
}
TEST_CASE("ScalablePipeline.Ifelse.1L.3W" * doctest::timeout(300)) {
ifelse_spipeline(1, 3);
}
TEST_CASE("ScalablePipeline.Ifelse.1L.4W" * doctest::timeout(300)) {
ifelse_spipeline(1, 4);
}
TEST_CASE("ScalablePipeline.Ifelse.3L.1W" * doctest::timeout(300)) {
ifelse_spipeline(3, 1);
}
TEST_CASE("ScalablePipeline.Ifelse.3L.2W" * doctest::timeout(300)) {
ifelse_spipeline(3, 2);
}
TEST_CASE("ScalablePipeline.Ifelse.3L.3W" * doctest::timeout(300)) {
ifelse_spipeline(3, 3);
}
TEST_CASE("ScalablePipeline.Ifelse.3L.4W" * doctest::timeout(300)) {
ifelse_spipeline(3, 4);
}
TEST_CASE("ScalablePipeline.Ifelse.5L.1W" * doctest::timeout(300)) {
ifelse_spipeline(5, 1);
}
TEST_CASE("ScalablePipeline.Ifelse.5L.2W" * doctest::timeout(300)) {
ifelse_spipeline(5, 2);
}
TEST_CASE("ScalablePipeline.Ifelse.5L.3W" * doctest::timeout(300)) {
ifelse_spipeline(5, 3);
}
TEST_CASE("ScalablePipeline.Ifelse.5L.4W" * doctest::timeout(300)) {
ifelse_spipeline(5, 4);
}
TEST_CASE("ScalablePipeline.Ifelse.7L.1W" * doctest::timeout(300)) {
ifelse_spipeline(7, 1);
}
TEST_CASE("ScalablePipeline.Ifelse.7L.2W" * doctest::timeout(300)) {
ifelse_spipeline(7, 2);
}
TEST_CASE("ScalablePipeline.Ifelse.7L.3W" * doctest::timeout(300)) {
ifelse_spipeline(7, 3);
}
TEST_CASE("ScalablePipeline.Ifelse.7L.4W" * doctest::timeout(300)) {
ifelse_spipeline(7, 4);
}
// ----------------------------------------------------------------------------
// ScalablePipeline in ScalablePipeline
// pipeline has 4 pipes, L lines, W workers
// each subpipeline has 3 pipes, subL lines
//
// pipeline = SPPS
// each subpipeline = SPS
//
// ----------------------------------------------------------------------------
void spipeline_in_spipeline(size_t L, unsigned w, unsigned subL) {
tf::Executor executor(w);
const size_t maxN = 7;
const size_t maxsubN = 7;
std::vector<std::vector<int>> source(maxN);
for(auto&& each: source) {
each.resize(maxsubN);
std::iota(each.begin(), each.end(), 0);
}
std::vector<std::array<int, 4>> buffer(L);
std::vector<tf::Pipe<>> pipes;
tf::ScalablePipeline<typename decltype(pipes)::iterator> pl;
// each pipe contains one subpipeline
// each subpipeline has three pipes, subL lines
//
// subbuffers[0][1][2][2] means
// first line, second pipe, third subline, third subpipe
std::vector<std::vector<std::vector<std::array<int, 3>>>> subbuffers(L);
for(auto&& b: subbuffers) {
b.resize(4);
for(auto&& each: b) {
each.resize(subL);
}
}
for (size_t N = 1; N < maxN; ++N) {
for(size_t subN = 1; subN < maxsubN; ++subN) {
size_t j1 = 0, j4 = 0;
std::atomic<size_t> j2 = 0;
std::atomic<size_t> j3 = 0;
// begin of pipeline ---------------------------
// begin of pipe 1 -----------------------------
pipes.emplace_back(tf::PipeType::SERIAL, [&, N, subN, subL](auto& pf) mutable {
if(j1 == N) {
pf.stop();
return;
}
size_t subj1 = 0, subj3 = 0;
std::atomic<size_t> subj2 = 0;
std::vector<int> subcollection;
subcollection.reserve(subN);
std::vector<tf::Pipe<>> subpipes;
tf::ScalablePipeline<typename decltype(subpipes)::iterator> subpl;
// subpipe 1
subpipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& subpf) mutable {
if(subj1 == subN) {
subpf.stop();
return;
}
REQUIRE(subpf.token() % subL == subpf.line());
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]
= source[pf.token()][subj1] + 1;
++subj1;
});
// subpipe 2
subpipes.emplace_back(tf::PipeType::PARALLEL, [&, subN](auto& subpf) mutable {
REQUIRE(subj2++ < subN);
REQUIRE(subpf.token() % subL == subpf.line());
REQUIRE(
source[pf.token()][subpf.token()] + 1 ==
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe() - 1]
);
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]
= source[pf.token()][subpf.token()] + 1;
});
// subpipe 3
subpipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& subpf) mutable {
REQUIRE(subj3 < subN);
REQUIRE(subpf.token() % subL == subpf.line());
REQUIRE(
source[pf.token()][subj3] + 1 ==
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe() - 1]
);
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]
= source[pf.token()][subj3] + 3;
subcollection.push_back(subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]);
++subj3;
});
tf::Taskflow taskflow;
// test task
auto test_t = taskflow.emplace([&, subN](){
REQUIRE(subj1 == subN);
REQUIRE(subj2 == subN);
REQUIRE(subj3 == subN);
REQUIRE(subpl.num_tokens() == subN);
REQUIRE(subcollection.size() == subN);
}).name("test");
// subpipeline
subpl.reset(subL, subpipes.begin(), subpipes.end());
auto subpl_t = taskflow.composed_of(subpl).name("module_of_subpipeline");
subpl_t.precede(test_t);
executor.corun(taskflow);
buffer[pf.line()][pf.pipe()] = std::accumulate(
subcollection.begin(),
subcollection.end(),
0
);
j1++;
});
// end of pipe 1 -----------------------------
//begin of pipe 2 ---------------------------
pipes.emplace_back(tf::PipeType::PARALLEL, [&, subN, subL](auto& pf) mutable {
REQUIRE(j2++ < N);
int res = std::accumulate(
source[pf.token()].begin(),
source[pf.token()].begin() + subN,
0
);
REQUIRE(buffer[pf.line()][pf.pipe() - 1] == res + 3 * subN);
size_t subj1 = 0, subj3 = 0;
std::atomic<size_t> subj2 = 0;
std::vector<int> subcollection;
subcollection.reserve(subN);
std::vector<tf::Pipe<>> subpipes;
tf::ScalablePipeline<typename decltype(subpipes)::iterator> subpl;
// subpipe 1
subpipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& subpf) mutable {
if(subj1 == subN) {
subpf.stop();
return;
}
REQUIRE(subpf.token() % subL == subpf.line());
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()] =
source[pf.token()][subj1] + 1;
++subj1;
});
// subpipe 2
subpipes.emplace_back(tf::PipeType::PARALLEL, [&, subN](auto& subpf) mutable {
REQUIRE(subj2++ < subN);
REQUIRE(subpf.token() % subL == subpf.line());
REQUIRE(
source[pf.token()][subpf.token()] + 1 ==
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe() - 1]
);
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]
= source[pf.token()][subpf.token()] + 1;
});
// subpipe 3
subpipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& subpf) mutable {
REQUIRE(subj3 < subN);
REQUIRE(subpf.token() % subL == subpf.line());
REQUIRE(
source[pf.token()][subj3] + 1 ==
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe() - 1]
);
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]
= source[pf.token()][subj3] + 13;
subcollection.push_back(subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]);
++subj3;
});
tf::Taskflow taskflow;
// test task
auto test_t = taskflow.emplace([&, subN](){
REQUIRE(subj1 == subN);
REQUIRE(subj2 == subN);
REQUIRE(subj3 == subN);
REQUIRE(subpl.num_tokens() == subN);
REQUIRE(subcollection.size() == subN);
}).name("test");
// subpipeline
subpl.reset(subL, subpipes.begin(), subpipes.end());
auto subpl_t = taskflow.composed_of(subpl).name("module_of_subpipeline");
subpl_t.precede(test_t);
executor.corun(taskflow);
buffer[pf.line()][pf.pipe()] = std::accumulate(
subcollection.begin(),
subcollection.end(),
0
);
});
// end of pipe 2 -----------------------------
// begin of pipe 3 ---------------------------
pipes.emplace_back(tf::PipeType::SERIAL, [&, N, subN, subL](auto& pf) mutable {
REQUIRE(j3++ < N);
int res = std::accumulate(
source[pf.token()].begin(),
source[pf.token()].begin() + subN,
0
);
REQUIRE(buffer[pf.line()][pf.pipe() - 1] == res + 13 * subN);
size_t subj1 = 0, subj3 = 0;
std::atomic<size_t> subj2 = 0;
std::vector<int> subcollection;
subcollection.reserve(subN);
std::vector<tf::Pipe<>> subpipes;
tf::ScalablePipeline<typename decltype(subpipes)::iterator> subpl;
// subpipe 1
subpipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& subpf) mutable {
if(subj1 == subN) {
subpf.stop();
return;
}
REQUIRE(subpf.token() % subL == subpf.line());
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]
= source[pf.token()][subj1] + 1;
++subj1;
});
// subpipe 2
subpipes.emplace_back(tf::PipeType::PARALLEL, [&, subN](auto& subpf) mutable {
REQUIRE(subj2++ < subN);
REQUIRE(subpf.token() % subL == subpf.line());
REQUIRE(
source[pf.token()][subpf.token()] + 1 ==
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe() - 1]
);
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()] =
source[pf.token()][subpf.token()] + 1;
});
// subpipe 3
subpipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& subpf) mutable {
REQUIRE(subj3 < subN);
REQUIRE(subpf.token() % subL == subpf.line());
REQUIRE(
source[pf.token()][subj3] + 1 ==
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe() - 1]
);
subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()] =
source[pf.token()][subj3] + 7;
subcollection.push_back(subbuffers[pf.line()][pf.pipe()][subpf.line()][subpf.pipe()]);
++subj3;
});
tf::Taskflow taskflow;
// test task
auto test_t = taskflow.emplace([&, subN](){
REQUIRE(subj1 == subN);
REQUIRE(subj2 == subN);
REQUIRE(subj3 == subN);
REQUIRE(subpl.num_tokens() == subN);
REQUIRE(subcollection.size() == subN);
}).name("test");
// subpipeline
subpl.reset(subL, subpipes.begin(), subpipes.end());
auto subpl_t = taskflow.composed_of(subpl).name("module_of_subpipeline");
subpl_t.precede(test_t);
executor.corun(taskflow);
buffer[pf.line()][pf.pipe()] = std::accumulate(
subcollection.begin(),
subcollection.end(),
0
);
});
// end of pipe 3 -----------------------------
// begin of pipe 4 ---------------------------
pipes.emplace_back(tf::PipeType::SERIAL, [&, subN](auto& pf) mutable {
int res = std::accumulate(
source[j4].begin(),
source[j4].begin() + subN,
0
);
REQUIRE(buffer[pf.line()][pf.pipe() - 1] == res + 7 * subN);
j4++;
});
// end of pipe 4 -----------------------------
pl.reset(L, pipes.begin(), pipes.end());
tf::Taskflow taskflow;
taskflow.composed_of(pl).name("module_of_pipeline");
executor.run(taskflow).wait();
pipes.clear();
}
}
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.1L.1W.1subL" * doctest::timeout(300)) {
spipeline_in_spipeline(1, 1, 1);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.1L.1W.3subL" * doctest::timeout(300)) {
spipeline_in_spipeline(1, 1, 3);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.1L.1W.4subL" * doctest::timeout(300)) {
spipeline_in_spipeline(1, 1, 4);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.1L.2W.1subL" * doctest::timeout(300)) {
spipeline_in_spipeline(1, 2, 1);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.1L.2W.3subL" * doctest::timeout(300)) {
spipeline_in_spipeline(1, 2, 3);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.1L.2W.4subL" * doctest::timeout(300)) {
spipeline_in_spipeline(1, 2, 4);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.3L.1W.1subL" * doctest::timeout(300)) {
spipeline_in_spipeline(3, 1, 1);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.3L.1W.3subL" * doctest::timeout(300)) {
spipeline_in_spipeline(3, 1, 3);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.3L.1W.4subL" * doctest::timeout(300)) {
spipeline_in_spipeline(3, 1, 4);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.3L.2W.1subL" * doctest::timeout(300)) {
spipeline_in_spipeline(3, 2, 1);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.3L.2W.3subL" * doctest::timeout(300)) {
spipeline_in_spipeline(3, 2, 3);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.3L.2W.4subL" * doctest::timeout(300)) {
spipeline_in_spipeline(3, 2, 4);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.5L.1W.1subL" * doctest::timeout(300)) {
spipeline_in_spipeline(5, 1, 1);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.5L.1W.3subL" * doctest::timeout(300)) {
spipeline_in_spipeline(5, 1, 3);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.5L.1W.4subL" * doctest::timeout(300)) {
spipeline_in_spipeline(5, 1, 4);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.5L.2W.1subL" * doctest::timeout(300)) {
spipeline_in_spipeline(5, 2, 1);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.5L.2W.3subL" * doctest::timeout(300)) {
spipeline_in_spipeline(5, 2, 3);
}
TEST_CASE("ScalablePipeline.PipelineinPipeline.5L.2W.4subL" * doctest::timeout(300)) {
spipeline_in_spipeline(5, 2, 4);
}
// ----------------------------------------------------------------------------
/* SNIG task graph
// o: normal task
// c: condition task
// p: pipeline
//
// four devices example:
// o
// / | | \
// c c c c -----
// | | | | |
// -----> p p p p |
// | | | | | |
// ----- c c c c |
// | | | | |
// o o o o |
// \ \ | / |
// \||/ |
// o <-----------
//
// each pipeline has five pipes, L lines, W workers
// each pipeline = SPSPS
*/
// ----------------------------------------------------------------------------
void snig_spipeline(size_t L, unsigned w) {
size_t NUM_SOURCE = 70000;
size_t BATCH_SIZE = 100;
std::array<size_t, 7> NUM_DEVICES = {1, 2, 4, 6, 9, 13, 17};
std::atomic<size_t> finished{0};
std::vector<int> source(NUM_SOURCE);
std::iota(source.begin(), source.end(), 0);
for(auto&& NUM_DEVICE: NUM_DEVICES) {
std::vector<std::vector<std::array<int, 5>>> buffers(NUM_DEVICE);
for(auto&& buffer: buffers) {
buffer.resize(L);
}
tf::Taskflow taskflow;
tf::Executor executor(w);
auto start_t = taskflow.emplace([](){}).name("start");
auto end_t = taskflow.emplace([](){}).name("end");
std::vector<tf::Task> dev_ends(NUM_DEVICE);
for(auto&& dev_end: dev_ends) {
dev_end = taskflow.emplace([](){}).name("dev_end");
}
std::vector<tf::Task> first_fetches(NUM_DEVICE);
std::vector<tf::Task> fetches(NUM_DEVICE);
std::vector<std::vector<tf::Pipe<>>> pipes(NUM_DEVICE);
// for type
using pipeline_it = std::vector<tf::Pipe<>>::iterator;
std::vector<tf::Task> module_of_pipelines(NUM_DEVICE);
std::vector<tf::ScalablePipeline<pipeline_it>> pipelines(NUM_DEVICE);
std::vector<size_t> dev_begins(NUM_DEVICE);
std::vector<size_t> j1s(NUM_DEVICE, 0);
std::vector<size_t> j3s(NUM_DEVICE, 0);
std::vector<size_t> j5s(NUM_DEVICE, 0);
std::vector<std::unique_ptr<std::atomic<size_t>>> j2s(NUM_DEVICE);
std::vector<std::unique_ptr<std::atomic<size_t>>> j4s(NUM_DEVICE);
for(size_t dev = 0; dev < NUM_DEVICE; ++dev) {
j2s[dev] = std::make_unique<std::atomic<size_t>>(0);
j4s[dev] = std::make_unique<std::atomic<size_t>>(0);
}
std::vector<std::vector<int>> collections(NUM_DEVICE);
for(auto&& collection: collections) {
collection.reserve(BATCH_SIZE);
}
for(size_t dev = 0; dev < NUM_DEVICE; ++dev) {
first_fetches[dev] = taskflow.emplace([&, dev, BATCH_SIZE](){
size_t num = finished.fetch_add(BATCH_SIZE);
dev_begins[dev] = num;
return num >= NUM_SOURCE;
}).name("first_fetch");
// pipe 1
pipes[dev].emplace_back(
tf::PipeType::SERIAL,
[&, dev, BATCH_SIZE](auto& pf) mutable {
if(j1s[dev] == BATCH_SIZE) {
pf.stop();
return;
}
REQUIRE(pf.token() % L == pf.line());
buffers[dev][pf.line()][pf.pipe()] = source[dev_begins[dev] + j1s[dev]] + 1;
++j1s[dev];
}
);
// pipe 2
pipes[dev].emplace_back(
tf::PipeType::PARALLEL, [&, dev, BATCH_SIZE](auto& pf) mutable {
REQUIRE((*j2s[dev])++ < BATCH_SIZE);
REQUIRE(pf.token() % L == pf.line());
REQUIRE(source[dev_begins[dev] + pf.token()] + 1 == buffers[dev][pf.line()][pf.pipe() - 1]);
buffers[dev][pf.line()][pf.pipe()] = source[dev_begins[dev] + pf.token()] + 3;
}
);
// pipe 3
pipes[dev].emplace_back(
tf::PipeType::SERIAL, [&, dev, BATCH_SIZE](auto& pf) mutable {
REQUIRE(j3s[dev] < BATCH_SIZE);
REQUIRE(pf.token() % L == pf.line());
REQUIRE(source[dev_begins[dev] + j3s[dev]] + 3 == buffers[dev][pf.line()][pf.pipe() - 1]);
buffers[dev][pf.line()][pf.pipe()] = source[dev_begins[dev] + j3s[dev]] + 8;
++j3s[dev];
}
);
// pipe 4
pipes[dev].emplace_back(
tf::PipeType::PARALLEL, [&, dev, BATCH_SIZE](auto& pf) mutable {
REQUIRE((*j4s[dev])++ < BATCH_SIZE);
REQUIRE(pf.token() % L == pf.line());
REQUIRE(source[dev_begins[dev] + pf.token()] + 8 == buffers[dev][pf.line()][pf.pipe() - 1]);
buffers[dev][pf.line()][pf.pipe()] = source[dev_begins[dev] + pf.token()] + 9;
}
);
// pipe 5
pipes[dev].emplace_back(
tf::PipeType::SERIAL, [&, dev, BATCH_SIZE](auto& pf) mutable {
REQUIRE(j5s[dev] < BATCH_SIZE);
REQUIRE(pf.token() % L == pf.line());
REQUIRE(source[dev_begins[dev] + j5s[dev]] + 9 == buffers[dev][pf.line()][pf.pipe() - 1]);
collections[dev].push_back(buffers[dev][pf.line()][pf.pipe() - 1] + 2);
++j5s[dev];
}
);
fetches[dev] = taskflow.emplace([&, dev, NUM_SOURCE, BATCH_SIZE](){
for(size_t b = 0; b < BATCH_SIZE; ++b) {
REQUIRE(source[dev_begins[dev] + b] + 9 + 2 == collections[dev][b]);
}
collections[dev].clear();
collections[dev].reserve(BATCH_SIZE);
size_t num = finished.fetch_add(BATCH_SIZE);
dev_begins[dev] = num;
j1s[dev] = 0;
*j2s[dev] = 0;
j3s[dev] = 0;
*j4s[dev] = 0;
j5s[dev] = 0;
pipelines[dev].reset();
return num >= NUM_SOURCE;
}).name("fetch");
}
for(size_t dev = 0; dev < NUM_DEVICE; ++dev) {
pipelines[dev].reset(L, pipes[dev].begin(), pipes[dev].end());
module_of_pipelines[dev] = taskflow.composed_of(pipelines[dev]).name("pipeline");
}
// dependencies
for(size_t dev = 0; dev < NUM_DEVICE; ++dev) {
start_t.precede(first_fetches[dev]);
first_fetches[dev].precede(
module_of_pipelines[dev],
dev_ends[dev]
);
module_of_pipelines[dev].precede(fetches[dev]);
fetches[dev].precede(module_of_pipelines[dev], dev_ends[dev]);
dev_ends[dev].precede(end_t);
}
executor.run(taskflow).wait();
}
}
TEST_CASE("ScalablePipeline.SNIG.1L.1W" * doctest::timeout(300)) {
snig_spipeline(1, 1);
}
TEST_CASE("ScalablePipeline.SNIG.1L.2W" * doctest::timeout(300)) {
snig_spipeline(1, 2);
}
TEST_CASE("ScalablePipeline.SNIG.1L.3W" * doctest::timeout(300)) {
snig_spipeline(1, 3);
}
TEST_CASE("ScalablePipeline.SNIG.3L.1W" * doctest::timeout(300)) {
snig_spipeline(3, 1);
}
TEST_CASE("ScalablePipeline.SNIG.3L.2W" * doctest::timeout(300)) {
snig_spipeline(3, 2);
}
TEST_CASE("ScalablePipeline.SNIG.3L.3W" * doctest::timeout(300)) {
snig_spipeline(3, 3);
}
TEST_CASE("ScalablePipeline.SNIG.5L.1W" * doctest::timeout(300)) {
snig_spipeline(5, 1);
}
TEST_CASE("ScalablePipeline.SNIG.5L.2W" * doctest::timeout(300)) {
snig_spipeline(5, 2);
}
TEST_CASE("ScalablePipeline.SNIG.5L.3W" * doctest::timeout(300)) {
snig_spipeline(5, 3);
}
TEST_CASE("ScalablePipeline.SNIG.7L.1W" * doctest::timeout(300)) {
snig_spipeline(7, 1);
}
TEST_CASE("ScalablePipeline.SNIG.7L.2W" * doctest::timeout(300)) {
snig_spipeline(7, 2);
}
TEST_CASE("ScalablePipeline.SNIG.7L.3W" * doctest::timeout(300)) {
snig_spipeline(7, 3);
}
// ----------------------------------------------------------------------
// Subflow pipeline
// -----------------------------------------------------------------------
void spawn(
tf::Subflow& sf,
size_t L,
size_t NUM_PIPES,
size_t NUM_RECURS,
size_t maxN,
size_t r,
std::vector<std::vector<int>>& buffer,
std::vector<std::vector<int>>& source,
std::vector<std::vector<tf::Pipe<>>>& pipes,
std::vector<tf::ScalablePipeline<typename std::vector<tf::Pipe<>>::iterator>>& spls,
size_t& counter
) {
// construct pipes
for(size_t p = 0; p < NUM_PIPES; ++p) {
pipes[r].emplace_back(tf::PipeType::SERIAL, [&, maxN, r](tf::Pipeflow& pf) mutable {
switch(pf.pipe()) {
case 0:
if(pf.token() == maxN) {
pf.stop();
++counter;
return;
}
buffer[r][pf.line()] = source[r][pf.token()];
break;
default:
++buffer[r][pf.line()];
}
REQUIRE(buffer[r][pf.line()] == source[r][pf.token()] + pf.pipe());
});
}
spls[r].reset(L, pipes[r].begin(), pipes[r].end());
auto spl_t = sf.composed_of(spls[r]).name("module_of_pipeline");
if(r + 1 < NUM_RECURS) {
auto spawn_t = sf.emplace([&, L, NUM_PIPES, NUM_RECURS, maxN, r](tf::Subflow& sf2) mutable {
spawn(sf2, L, NUM_PIPES, NUM_RECURS, maxN, r + 1, buffer, source, pipes, spls, counter);
});
spawn_t.precede(spl_t);
}
}
void subflow_spipeline(unsigned NUM_RECURS, unsigned w, size_t L) {
tf::Executor executor(w);
tf::Taskflow taskflow;
std::vector<tf::ScalablePipeline<typename std::vector<tf::Pipe<>>::iterator>> spls(NUM_RECURS);
std::vector<std::vector<tf::Pipe<>>> pipes(NUM_RECURS);
size_t maxN = 1123;
size_t NUM_PIPES = 5;
size_t counter = 0;
std::vector<std::vector<int>> source(NUM_RECURS);
for(auto&& each: source) {
each.resize(maxN);
std::iota(each.begin(), each.end(), 0);
}
std::vector<std::vector<int>> buffer(NUM_RECURS);
for(auto&& each:buffer) {
each.resize(L);
}
auto subflows = taskflow.emplace([&, L, NUM_PIPES, NUM_RECURS, maxN](tf::Subflow& sf){
spawn(sf, L, NUM_PIPES, NUM_RECURS, maxN, 0, buffer, source, pipes, spls, counter);
});
auto check = taskflow.emplace([&, NUM_RECURS](){
REQUIRE(counter == NUM_RECURS);
}).name("check");
subflows.precede(check);
executor.run(taskflow).wait();
}
TEST_CASE("ScalablePipeline.Subflow.1R.1W.1L" * doctest::timeout(300)) {
subflow_spipeline(1, 1, 1);
}
TEST_CASE("ScalablePipeline.Subflow.1R.1W.3L" * doctest::timeout(300)) {
subflow_spipeline(1, 1, 3);
}
TEST_CASE("ScalablePipeline.Subflow.1R.1W.4L" * doctest::timeout(300)) {
subflow_spipeline(1, 1, 4);
}
TEST_CASE("ScalablePipeline.Subflow.1R.2W.1L" * doctest::timeout(300)) {
subflow_spipeline(1, 2, 1);
}
TEST_CASE("ScalablePipeline.Subflow.1R.2W.3L" * doctest::timeout(300)) {
subflow_spipeline(1, 2, 3);
}
TEST_CASE("ScalablePipeline.Subflow.1R.2W.4L" * doctest::timeout(300)) {
subflow_spipeline(1, 2, 4);
}
TEST_CASE("ScalablePipeline.Subflow.3R.1W.1L" * doctest::timeout(300)) {
subflow_spipeline(3, 1, 1);
}
TEST_CASE("ScalablePipeline.Subflow.3R.1W.3L" * doctest::timeout(300)) {
subflow_spipeline(3, 1, 3);
}
TEST_CASE("ScalablePipeline.Subflow.3R.1W.4L" * doctest::timeout(300)) {
subflow_spipeline(3, 1, 4);
}
TEST_CASE("ScalablePipeline.Subflow.3R.2W.1L" * doctest::timeout(300)) {
subflow_spipeline(3, 2, 1);
}
TEST_CASE("ScalablePipeline.Subflow.3R.2W.3L" * doctest::timeout(300)) {
subflow_spipeline(3, 2, 3);
}
TEST_CASE("ScalablePipeline.Subflow.3R.2W.4L" * doctest::timeout(300)) {
subflow_spipeline(3, 2, 4);
}
TEST_CASE("ScalablePipeline.Subflow.5R.1W.1L" * doctest::timeout(300)) {
subflow_spipeline(5, 1, 1);
}
TEST_CASE("ScalablePipeline.Subflow.5R.1W.3L" * doctest::timeout(300)) {
subflow_spipeline(5, 1, 3);
}
TEST_CASE("ScalablePipeline.Subflow.5R.1W.4L" * doctest::timeout(300)) {
subflow_spipeline(5, 1, 4);
}
TEST_CASE("ScalablePipeline.Subflow.5R.2W.1L" * doctest::timeout(300)) {
subflow_spipeline(5, 2, 1);
}
TEST_CASE("ScalablePipeline.Subflow.5R.2W.3L" * doctest::timeout(300)) {
subflow_spipeline(5, 2, 3);
}
TEST_CASE("ScalablePipeline.Subflow.5R.2W.4L" * doctest::timeout(300)) {
subflow_spipeline(5, 2, 4);
}
TEST_CASE("ScalablePipeline.Subflow.7R.1W.1L" * doctest::timeout(300)) {
subflow_spipeline(7, 1, 1);
}
TEST_CASE("ScalablePipeline.Subflow.7R.1W.3L" * doctest::timeout(300)) {
subflow_spipeline(7, 1, 3);
}
TEST_CASE("ScalablePipeline.Subflow.7R.1W.4L" * doctest::timeout(300)) {
subflow_spipeline(7, 1, 4);
}
TEST_CASE("ScalablePipeline.Subflow.7R.2W.1L" * doctest::timeout(300)) {
subflow_spipeline(7, 2, 1);
}
TEST_CASE("ScalablePipeline.Subflow.7R.2W.3L" * doctest::timeout(300)) {
subflow_spipeline(7, 2, 3);
}
TEST_CASE("ScalablePipeline.Subflow.7R.2W.4L" * doctest::timeout(300)) {
subflow_spipeline(7, 2, 4);
}
// ------------------------------------------------------------------------
// Scalable Pipeline with move constructor and move assignment constructor
// ------------------------------------------------------------------------
TEST_CASE("ScalablePipeline.move" * doctest::timeout(300)) {
size_t N = 10;
std::atomic<int> counter{0};
std::vector< tf::Pipe<std::function<void(tf::Pipeflow&)>> > pipes;
for(size_t i=0; i<N; i++) {
pipes.emplace_back(tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {
if (pf.token() == 5) {
pf.stop();
}
else {
++counter;
}
});
}
using iterator_type = decltype(pipes)::iterator;
tf::ScalablePipeline<iterator_type> rhs;
REQUIRE(rhs.num_lines() == 0);
REQUIRE(rhs.num_pipes() == 0);
REQUIRE(rhs.num_tokens() == 0);
rhs.reset(1, pipes.begin(), pipes.end());
REQUIRE(rhs.num_lines() == 1);
REQUIRE(rhs.num_pipes() == N);
REQUIRE(rhs.num_tokens() == 0);
{
tf::Executor executor;
tf::Taskflow taskflow;
taskflow.composed_of(rhs);
executor.run(taskflow).wait();
REQUIRE(counter == 50);
}
auto lhs = std::move(rhs);
REQUIRE(rhs.num_lines() == 0);
REQUIRE(rhs.num_pipes() == 0);
REQUIRE(rhs.num_tokens() == 0);
REQUIRE(lhs.num_lines() == 1);
REQUIRE(lhs.num_pipes() == N);
REQUIRE(lhs.num_tokens() == 5);
{
tf::Executor executor;
tf::Taskflow taskflow;
taskflow.composed_of(lhs);
executor.run(taskflow).wait();
REQUIRE(counter == 50);
}
rhs = std::move(lhs);
REQUIRE(lhs.num_lines() == 0);
REQUIRE(lhs.num_pipes() == 0);
REQUIRE(lhs.num_tokens() == 0);
REQUIRE(rhs.num_lines() == 1);
REQUIRE(rhs.num_pipes() == N);
REQUIRE(rhs.num_tokens() == 5);
{
tf::Executor executor;
tf::Taskflow taskflow;
taskflow.composed_of(rhs);
executor.run(taskflow).wait();
REQUIRE(counter == 50);
}
}
Reference in a new issue View git blame Copy permalink