#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN

#include <doctest.h>

#include <taskflow/taskflow.hpp>

#include <vector>

#include <utility>

#include <chrono>

#include <limits.h>

TEST_CASE("Builder" * doctest::timeout(300)) {

const size_t num_tasks = 100;

tf::Taskflow taskflow;

tf::Executor executor;

std::atomic<int> counter {0};

std::vector<tf::Task> silent_tasks;

std::vector<tf::Task> tasks;

SUBCASE("Empty") {

for(size_t W=0; W<32; ++W) {

tf::Executor executor(W);

tf::Taskflow taskflow;

REQUIRE(taskflow.num_nodes() == 0);

REQUIRE(taskflow.empty() == true);

executor.run(taskflow).wait();

}

}

//SUBCASE("NoWorker") {

// tf::Executor executor(0);

// tf::Taskflow taskflow;

// taskflow.emplace([](){});

// REQUIRE_THROWS(executor.run(taskflow));

//}

SUBCASE("Placeholder") {

for(size_t i=0; i<num_tasks; ++i) {

silent_tasks.emplace_back(taskflow.placeholder().name(std::to_string(i)));

}

for(size_t i=0; i<num_tasks; ++i) {

REQUIRE(silent_tasks[i].name() == std::to_string(i));

REQUIRE(silent_tasks[i].num_dependents() == 0);

REQUIRE(silent_tasks[i].num_successors() == 0);

}

for(auto& task : silent_tasks) {

task.work([&counter](){ counter++; });

}

executor.run(taskflow).get();

REQUIRE(counter == num_tasks);

}

SUBCASE("EmbarrassinglyParallel"){

for(size_t i=0;i<num_tasks;i++) {

tasks.emplace_back(taskflow.emplace([&counter]() {counter += 1;}));

}

REQUIRE(taskflow.num_nodes() == num_tasks);

executor.run(taskflow).get();

REQUIRE(counter == num_tasks);

REQUIRE(taskflow.num_nodes() == 100);

counter = 0;

for(size_t i=0;i<num_tasks;i++){

silent_tasks.emplace_back(taskflow.emplace([&counter]() {counter += 1;}));

}

REQUIRE(taskflow.num_nodes() == num_tasks * 2);

executor.run(taskflow).get();

REQUIRE(counter == num_tasks * 2);

REQUIRE(taskflow.num_nodes() == 200);

}

SUBCASE("BinarySequence"){

for(size_t i=0;i<num_tasks;i++){

if(i%2 == 0){

tasks.emplace_back(

taskflow.emplace([&counter]() { REQUIRE(counter == 0); counter += 1;})

);

}

else{

tasks.emplace_back(

taskflow.emplace([&counter]() { REQUIRE(counter == 1); counter -= 1;})

);

}

if(i>0){

//tasks[i-1].first.precede(tasks[i].first);

tasks[i-1].precede(tasks[i]);

}

if(i==0) {

//REQUIRE(tasks[i].first.num_dependents() == 0);

REQUIRE(tasks[i].num_dependents() == 0);

}

else {

//REQUIRE(tasks[i].first.num_dependents() == 1);

REQUIRE(tasks[i].num_dependents() == 1);

}

}

executor.run(taskflow).get();

}

SUBCASE("LinearCounter"){

for(size_t i=0;i<num_tasks;i++){

tasks.emplace_back(

taskflow.emplace([&counter, i]() {

REQUIRE(counter == i); counter += 1;}

)

);

if(i>0){

taskflow.precede(tasks[i-1], tasks[i]);

}

}

executor.run(taskflow).get();

REQUIRE(counter == num_tasks);

REQUIRE(taskflow.num_nodes() == num_tasks);

}

SUBCASE("Broadcast"){

auto src = taskflow.emplace([&counter]() {counter -= 1;});

for(size_t i=1; i<num_tasks; i++){

silent_tasks.emplace_back(

taskflow.emplace([&counter]() {REQUIRE(counter == -1);})

);

}

taskflow.broadcast(src, silent_tasks);

executor.run(taskflow).get();

REQUIRE(counter == - 1);

REQUIRE(taskflow.num_nodes() == num_tasks);

}

SUBCASE("Succeed"){

auto dst = taskflow.emplace([&]() { REQUIRE(counter == num_tasks - 1);});

for(size_t i=1;i<num_tasks;i++){

silent_tasks.emplace_back(

taskflow.emplace([&counter]() {counter += 1;})

);

}

dst.succeed(silent_tasks);

executor.run(taskflow).get();

REQUIRE(counter == num_tasks - 1);

REQUIRE(taskflow.num_nodes() == num_tasks);

}

SUBCASE("MapReduce"){

auto src = taskflow.emplace([&counter]() {counter = 0;});

for(size_t i=0;i<num_tasks;i++){

silent_tasks.emplace_back(

taskflow.emplace([&counter]() {counter += 1;})

);

}

taskflow.broadcast(src, silent_tasks);

auto dst = taskflow.emplace(

[&counter, num_tasks]() { REQUIRE(counter == num_tasks);}

);

taskflow.gather(silent_tasks, dst);

executor.run(taskflow).get();

REQUIRE(taskflow.num_nodes() == num_tasks + 2);

}

SUBCASE("Linearize"){

for(size_t i=0;i<num_tasks;i++){

silent_tasks.emplace_back(

taskflow.emplace([&counter, i]() {

REQUIRE(counter == i); counter += 1;}

)

);

}

taskflow.linearize(silent_tasks);

executor.run(taskflow).get();

REQUIRE(counter == num_tasks);

REQUIRE(taskflow.num_nodes() == num_tasks);

}

SUBCASE("Kite"){

auto src = taskflow.emplace([&counter]() {counter = 0;});

for(size_t i=0;i<num_tasks;i++){

silent_tasks.emplace_back(

taskflow.emplace([&counter, i]() {

REQUIRE(counter == i); counter += 1; }

)

);

}

taskflow.broadcast(src, silent_tasks);

taskflow.linearize(silent_tasks);

auto dst = taskflow.emplace(

[&counter, num_tasks]() { REQUIRE(counter == num_tasks);}

);

taskflow.gather(silent_tasks, dst);

executor.run(taskflow).get();

REQUIRE(taskflow.num_nodes() == num_tasks + 2);

}

}

TEST_CASE("Creation" * doctest::timeout(300)) {

std::vector<int> dummy(1024, -1);

auto create_taskflow = [&] () {

for(int i=0; i<1024; ++i) {

tf::Taskflow tf;

tf.parallel_for(dummy.begin(), dummy.end(), [] (int) {});

}

};

SUBCASE("One") {

create_taskflow();

}

SUBCASE("Two") {

std::thread t1(create_taskflow);

std::thread t2(create_taskflow);

t1.join();

t2.join();

}

SUBCASE("Four") {

std::thread t1(create_taskflow);

std::thread t2(create_taskflow);

std::thread t3(create_taskflow);

std::thread t4(create_taskflow);

t1.join();

t2.join();

t3.join();

t4.join();

}

}

TEST_CASE("SequentialRun" * doctest::timeout(300)) {

using namespace std::chrono_literals;

size_t num_workers = 4;

size_t num_tasks = 100;

tf::Executor executor(num_workers);

tf::Taskflow taskflow;

REQUIRE(executor.num_workers() == num_workers);

std::atomic<int> counter {0};

std::vector<tf::Task> silent_tasks;

for(size_t i=0;i<num_tasks;i++){

silent_tasks.emplace_back(

taskflow.emplace([&counter]() {counter += 1;})

);

}

SUBCASE("RunOnce"){

auto fu = executor.run(taskflow);

REQUIRE(taskflow.num_nodes() == num_tasks);

fu.get();

REQUIRE(counter == num_tasks);

}

SUBCASE("WaitForAll") {

executor.run(taskflow);

executor.wait_for_all();

REQUIRE(counter == num_tasks);

}

SUBCASE("RunWithFuture") {

// Empty subflow test

for(unsigned W=1; W<=4; ++W) {

std::atomic<size_t> count {0};

tf::Taskflow f;

auto A = f.emplace([&](){ count ++; });

auto B = f.emplace([&](auto& subflow){

count ++;

auto B1 = subflow.emplace([&](){ count++; });

auto B2 = subflow.emplace([&](){ count++; });

auto B3 = subflow.emplace([&](){ count++; });

B1.precede(B3); B2.precede(B3);

});

auto C = f.emplace([&](){ count ++; });

auto D = f.emplace([&](){ count ++; });

A.precede(B, C);

B.precede(D);

C.precede(D);

tf::Executor executor(W);

std::list<std::future<void>> fu_list;

for(size_t i=0; i<500; i++) {

if(i == 499) {

executor.run(f).get(); // Synchronize the first 500 runs

executor.run_n(f, 500); // Run 500 times more

}

else if(i % 2) {

fu_list.push_back(executor.run(f));

}

else {

fu_list.push_back(executor.run(f, [&, i=i](){

REQUIRE(count == (i+1)*7); })

);

}

}

executor.wait_for_all();

for(auto& fu: fu_list) {

REQUIRE(fu.valid());

REQUIRE(fu.wait_for(std::chrono::seconds(1)) == std::future_status::ready);

}

REQUIRE(count == 7000);

}

}

SUBCASE("RunWithChange") {

// test correctness when taskflow got changed between runs

for(unsigned W=1; W<=4; ++W) {

std::atomic<size_t> count {0};

tf::Taskflow f;

auto A = f.emplace([&](){ count ++; });

auto B = f.emplace([&](auto& subflow){

count ++;

auto B1 = subflow.emplace([&](){ count++; });

auto B2 = subflow.emplace([&](){ count++; });

auto B3 = subflow.emplace([&](){ count++; });

B1.precede(B3); B2.precede(B3);

});

auto C = f.emplace([&](){ count ++; });

auto D = f.emplace([&](){ count ++; });

A.precede(B, C);

B.precede(D);

C.precede(D);

tf::Executor executor(W);

executor.run_n(f, 10).get();

REQUIRE(count == 70);

auto E = f.emplace([](){});

D.precede(E);

executor.run_n(f, 10).get();

REQUIRE(count == 140);

auto F = f.emplace([](){});

E.precede(F);

executor.run_n(f, 10);

executor.wait_for_all();

REQUIRE(count == 210);

}

}

SUBCASE("RunWithPred") {

// Test run_until

for(unsigned W=1; W<=4; ++W) {

std::atomic<size_t> count {0};

tf::Taskflow f;

auto A = f.emplace([&](){ count ++; });

auto B = f.emplace([&](auto& subflow){

count ++;

auto B1 = subflow.emplace([&](){ count++; });

auto B2 = subflow.emplace([&](){ count++; });

auto B3 = subflow.emplace([&](){ count++; });

B1.precede(B3); B2.precede(B3);

});

auto C = f.emplace([&](){ count ++; });

auto D = f.emplace([&](){ count ++; });

A.precede(B, C);

B.precede(D);

C.precede(D);

tf::Executor executor(W);

executor.run_until(f, [run=10]() mutable { return run-- == 0; },

[&](){

REQUIRE(count == 70);

count = 0;

}

).get();

executor.run_until(f, [run=10]() mutable { return run-- == 0; },

[&](){

REQUIRE(count == 70);

count = 0;

auto E = f.emplace([&](){ count ++; });

auto F = f.emplace([&](){ count ++; });

A.precede(E).precede(F);

});

executor.run_until(f, [run=10]() mutable { return run-- == 0; },

[&](){

REQUIRE(count == 90);

count = 0;

}

).get();

}

}

SUBCASE("MultipleRuns") {

for(size_t W=1; W<=8; ++W) {

tf::Executor executor(W);

std::atomic<size_t> counter(0);

tf::Taskflow tf1, tf2, tf3, tf4;

for(size_t n=0; n<16; ++n) {

tf1.emplace([&](){counter.fetch_add(1, std::memory_order_relaxed);});

}

for(size_t n=0; n<1024; ++n) {

tf2.emplace([&](){counter.fetch_add(1, std::memory_order_relaxed);});

}

for(size_t n=0; n<32; ++n) {

tf3.emplace([&](){counter.fetch_add(1, std::memory_order_relaxed);});

}

for(size_t n=0; n<128; ++n) {

tf4.emplace([&](){counter.fetch_add(1, std::memory_order_relaxed);});

}

for(int i=0; i<200; ++i) {

executor.run(tf1);

executor.run(tf2);

executor.run(tf3);

executor.run(tf4);

}

executor.wait_for_all();

REQUIRE(counter == 240000);

}

}

}

TEST_CASE("ParallelRuns" * doctest::timeout(300)) {

std::atomic<int> counter;

std::vector<std::thread> threads;

auto make_taskflow = [&] (tf::Taskflow& tf) {

for(int i=0; i<1024; i++) {

auto A = tf.emplace([&] () {

counter.fetch_add(1, std::memory_order_relaxed);

});

auto B = tf.emplace([&] () {

counter.fetch_add(1, std::memory_order_relaxed);

});

A.precede(B);

}

};

SUBCASE("RunAndWait") {

for(size_t w=1; w<=32; ++w) {

tf::Executor executor(w);

counter = 0;

for(int t=0; t<32; t++) {

threads.emplace_back([&] () {

tf::Taskflow taskflow;

make_taskflow(taskflow);

executor.run(taskflow).wait();

});

}

for(auto& t : threads) {

t.join();

}

threads.clear();

REQUIRE(counter.load() == 32*1024*2);

}

}

SUBCASE("RunAndWaitForAll") {

for(size_t w=1; w<=32; ++w) {

tf::Executor executor(w);

counter = 0;

std::vector<std::unique_ptr<tf::Taskflow>> taskflows(32);

std::atomic<size_t> barrier(0);

for(int t=0; t<32; t++) {

threads.emplace_back([&, t=t] () {

taskflows[t] = std::make_unique<tf::Taskflow>();

make_taskflow(*taskflows[t]);

executor.run(*taskflows[t]);

++barrier; // make sure all runs are issued

});

}

while(barrier != 32);

executor.wait_for_all();

REQUIRE(counter.load() == 32*1024*2);

for(auto& t : threads) {

t.join();

}

threads.clear();

}

}

}

TEST_CASE("ParallelFor" * doctest::timeout(300)) {

using namespace std::chrono_literals;

const auto mapper = [](size_t w, size_t num_data, bool group){

tf::Executor executor(static_cast<unsigned>(w));

tf::Taskflow tf;

std::vector<int> vec(num_data, 0);

tf.parallel_for(

vec.begin(), vec.end(), [] (int& v) { v = 64; }, group ? ::rand() : 0

);

for(const auto v : vec) {

REQUIRE(v == 0);

}

executor.run(tf);

executor.wait_for_all();

for(const auto v : vec) {

REQUIRE(v == 64);

}

};

const auto reducer = [](size_t w, size_t num_data, bool group){

tf::Executor executor(static_cast<unsigned>(w));

tf::Taskflow tf;

std::vector<int> vec(num_data, 0);

std::atomic<int> sum(0);

tf.parallel_for(vec.begin(), vec.end(), [&](auto) { ++sum; }, group ? ::rand() : 0);

REQUIRE(sum == 0);

executor.run(tf);

executor.wait_for_all();

REQUIRE(sum == vec.size());

};

// map

SUBCASE("Map") {

for(size_t W=1; W<=4; ++W){

for(size_t num_data=1; num_data<=59049; num_data *= 3){

mapper(W, num_data, true);

mapper(W, num_data, false);

}

}

}

// reduce

SUBCASE("Reduce") {

for(size_t W=1; W<=4; ++W){

for(size_t num_data=1; num_data<=59049; num_data *= 3){

reducer(W, num_data, true);

reducer(W, num_data, false);

}

}

}

}

TEST_CASE("ParallelForOnIndex" * doctest::timeout(300)) {

using namespace std::chrono_literals;

auto exception_test = [] () {

tf::Taskflow tf;

// invalid index

REQUIRE_THROWS(tf.parallel_for(0, 10, 0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0, 10, -1, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10, 0, 0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10, 0, 1, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0u, 10u, 0u, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10u, 0u, 0u, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10u, 0u, 1u, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0.0f, 10.0f, 0.0f, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0.0f, 10.0f, -1.0f, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10.0f, 0.0f, 0.0f, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10.0f, 0.0f, 1.0f, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0.0, 10.0, 0.0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0.0, 10.0, -1.0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10.0, 0.0, 0.0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(10.0, 0.0, 1.0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0, 0, 0, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0u, 0u, 0u, [] (auto) {}));

REQUIRE_THROWS(tf.parallel_for(0.0, 0.0, 0.0, [] (auto) {}));

// graceful case

REQUIRE_NOTHROW(tf.parallel_for(0, 0, -1, [] (auto) {}));

REQUIRE_NOTHROW(tf.parallel_for(0, 0, 1, [] (auto) {}));

REQUIRE_NOTHROW(tf.parallel_for(0u, 0u, 1u, [] (auto) {}));

REQUIRE_NOTHROW(tf.parallel_for(0.0f, 0.0f, -1.0f, [] (auto) {}));

REQUIRE_NOTHROW(tf.parallel_for(0.0f, 0.0f, 1.0f, [] (auto) {}));

REQUIRE_NOTHROW(tf.parallel_for(0.0, 0.0, -1.0, [] (auto) {}));

REQUIRE_NOTHROW(tf.parallel_for(0.0, 0.0, 1.0, [] (auto) {}));

};

auto positive_integer_step = [] (unsigned w) {

tf::Executor executor(w);

for(int beg=-10; beg<=10; ++beg) {

for(int end=beg; end<=10; ++end) {

for(int s=1; s<=end-beg; ++s) {

int n = 0;

for(int b = beg; b<end; b+=s) {

++n;

}

tf::Taskflow tf;

std::atomic<int> counter {0};

tf.parallel_for(beg, end, s, [&] (auto) {

counter.fetch_add(1, std::memory_order_relaxed);

});

executor.run(tf);

executor.wait_for_all();

REQUIRE(n == counter);

}

}

}

};

auto negative_integer_step = [] (unsigned w) {

tf::Executor executor(w);

for(int beg=10; beg>=-10; --beg) {

for(int end=beg; end>=-10; --end) {

for(int s=1; s<=beg-end; ++s) {

int n = 0;

for(int b = beg; b>end; b-=s) {

++n;

}

tf::Taskflow tf;

std::atomic<int> counter {0};

tf.parallel_for(beg, end, -s, [&] (auto) {

counter.fetch_add(1, std::memory_order_relaxed);

});

executor.run(tf);

executor.wait_for_all();

REQUIRE(n == counter);

}

}

}

};

auto positive_floating_step = [] (unsigned w) {

tf::Executor executor(w);

for(float beg=-10.0f; beg<=10.0f; ++beg) {

for(float end=beg; end<=10.0f; ++end) {

for(float s=1.0f; s<=end-beg; s+=0.1f) {

int n = 0;

for(float b = beg; b<end; b+=s) {

++n;

}

tf::Taskflow tf;

std::atomic<int> counter {0};

tf.parallel_for(beg, end, s, [&] (auto) {

counter.fetch_add(1, std::memory_order_relaxed);

});

executor.run(tf);

executor.wait_for_all();

REQUIRE(n == counter);

}

}

}

};

auto negative_floating_step = [] (unsigned w) {

tf::Executor executor(w);

for(float beg=10.0f; beg>=-10.0f; --beg) {

for(float end=beg; end>=-10.0f; --end) {

for(float s=1.0f; s<=beg-end; s+=0.1f) {

int n = 0;

for(float b = beg; b>end; b-=s) {

++n;

}

tf::Taskflow tf;

std::atomic<int> counter {0};

tf.parallel_for(beg, end, -s, [&] (auto) {

counter.fetch_add(1, std::memory_order_relaxed);

});

executor.run(tf);

executor.wait_for_all();

REQUIRE(n == counter);

}

}

}

};

SUBCASE("Exception") {

exception_test();

}

SUBCASE("PositiveIntegerStep") {

for(unsigned w=1; w<=4; w++) {

positive_integer_step(w);

}

}

SUBCASE("NegativeIntegerStep") {

for(unsigned w=1; w<=4; w++) {

negative_integer_step(w);

}

}

SUBCASE("PositiveFloatingStep") {

for(unsigned w=1; w<=4; w++) {

positive_floating_step(w);

}

}

SUBCASE("NegativeFloatingStep") {

for(unsigned w=1; w<=4; w++) {

negative_floating_step(w);

}

}

}

TEST_CASE("Reduce" * doctest::timeout(300)) {

const auto plus_test = [](size_t num_workers, auto &&data){

tf::Executor executor(static_cast<unsigned>(num_workers));

tf::Taskflow tf;

int result {0};

std::iota(data.begin(), data.end(), 1);

tf.reduce(data.begin(), data.end(), result, std::plus<int>());

executor.run(tf).get();

REQUIRE(result == std::accumulate(data.begin(), data.end(), 0, std::plus<int>()));

};

const auto multiply_test = [](size_t num_workers, auto &&data){

tf::Executor executor(static_cast<unsigned>(num_workers));

tf::Taskflow tf;

std::fill(data.begin(), data.end(), 1.0);

double result {2.0};

tf.reduce(data.begin(), data.end(), result, std::multiplies<double>());

executor.run(tf).get();

REQUIRE(result == std::accumulate(data.begin(), data.end(), 2.0, std::multiplies<double>()));

};

const auto max_test = [](size_t num_workers, auto &&data){

tf::Executor executor(static_cast<unsigned>(num_workers));

tf::Taskflow tf;

std::iota(data.begin(), data.end(), 1);

int result {0};

auto lambda = [](const auto& l, const auto& r){return std::max(l, r);};

tf.reduce(data.begin(), data.end(), result, lambda);

executor.run(tf).get();

REQUIRE(result == std::accumulate(data.begin(), data.end(), 0, lambda));

};

const auto min_test = [](size_t num_workers, auto &&data){

tf::Executor executor(static_cast<unsigned>(num_workers));

tf::Taskflow tf;

std::iota(data.begin(), data.end(), 1);

int result {std::numeric_limits<int>::max()};

auto lambda = [](const auto& l, const auto& r){return std::min(l, r);};

tf.reduce(data.begin(), data.end(), result, lambda);

executor.run(tf).get();

REQUIRE(result == std::accumulate(

data.begin(), data.end(), std::numeric_limits<int>::max(), lambda)

);

};

for(size_t w=1; w<=4; ++w){

for(size_t j=0; j<=256; j=j*2+1){

plus_test(w, std::vector<int>(j));

plus_test(w, std::list<int>(j));

multiply_test(w, std::vector<double>(j));

multiply_test(w, std::list<double>(j));

max_test(w, std::vector<int>(j));

max_test(w, std::list<int>(j));

min_test(w, std::vector<int>(j));

min_test(w, std::list<int>(j));

}

}

}

TEST_CASE("ReduceMin" * doctest::timeout(300)) {

for(int w=1; w<=4; w++) {

tf::Executor executor(w);

for(int i=0; i<=65536; i = (i <= 1024) ? i + 1 : i*2 + 1) {

tf::Taskflow tf;

std::vector<int> data(i);

int gold = std::numeric_limits<int>::max();

int test = std::numeric_limits<int>::max();

for(auto& d : data) {

d = ::rand();

gold = std::min(gold, d);

}

tf.reduce_min(data.begin(), data.end(), test);

executor.run(tf).get();

REQUIRE(test == gold);

}

}

}

TEST_CASE("ReduceMax" * doctest::timeout(300)) {

for(int w=1; w<=4; w++) {

tf::Executor executor(w);

for(int i=0; i<=65536; i = (i <= 1024) ? i + 1 : i*2 + 1) {

tf::Taskflow tf;

std::vector<int> data(i);

int gold = std::numeric_limits<int>::min();

int test = std::numeric_limits<int>::min();

for(auto& d : data) {

d = ::rand();

gold = std::max(gold, d);

}

tf.reduce_max(data.begin(), data.end(), test);

executor.run(tf).get();

REQUIRE(test == gold);

}

}

}

TEST_CASE("JoinedSubflow" * doctest::timeout(300)){

using namespace std::literals::chrono_literals;

SUBCASE("Trivial") {

// Empty subflow test

for(unsigned W=1; W<=4; ++W) {

tf::Executor executor(W);

tf::Taskflow tf;

// empty flow with future

tf::Task subflow3, subflow3_;

//std::future<int> fu3, fu3_;

std::atomic<int> fu3v{0}, fu3v_{0};

// empty flow

auto subflow1 = tf.emplace([&] (auto& fb) {

fu3v++;

fb.join();

}).name("subflow1");

// nested empty flow

auto subflow2 = tf.emplace([&] (auto& fb) {

fu3v++;

fb.emplace([&] (auto& fb) {

fu3v++;

fb.emplace( [&] (auto& fb) {

fu3v++;

fb.join();

}).name("subflow2_1_1");

}).name("subflow2_1");

}).name("subflow2");

//std::tie(subflow3, fu3) = tf.emplace([&] (auto& fb) {

subflow3 = tf.emplace([&] (auto& fb) {

REQUIRE(fu3v == 4);

fu3v++;

fu3v_++;

//std::tie(subflow3_, fu3_) = fb.emplace([&] (auto& fb) {

subflow3_ = fb.emplace([&] (auto& fb) {

REQUIRE(fu3v_ == 3);

fu3v++;

fu3v_++;

//return 200;

fb.join();

});

subflow3_.name("subflow3_");

// hereafter we use 100us to avoid dangling reference ...

auto s1 = fb.emplace([&] () {

fu3v_++;

fu3v++;

}).name("s1");

auto s2 = fb.emplace([&] () {

fu3v_++;

fu3v++;

}).name("s2");

auto s3 = fb.emplace([&] () {

fu3v++;

REQUIRE(fu3v_ == 4);

}).name("s3");

s1.precede(subflow3_);

s2.precede(subflow3_);

subflow3_.precede(s3);

REQUIRE(fu3v_ == 1);

//return 100;

});

subflow3.name("subflow3");

// empty flow to test future

auto subflow4 = tf.emplace([&] () {

fu3v++;

}).name("subflow4");

subflow1.precede(subflow2);

subflow2.precede(subflow3);

subflow3.precede(subflow4);

executor.run(tf).get();

} // End of for loop

}

// Mixed intra- and inter- operations

SUBCASE("Complex") {

for(unsigned W=1; W<=4; ++W) {

tf::Executor executor(W);

tf::Taskflow tf;

std::vector<int> data;

int sum {0};

auto A = tf.emplace([&data] () {

for(int i=0; i<10; ++i) {

data.push_back(1);

}

});

std::atomic<size_t> count = 0;

auto B = tf.emplace([&count, &data, &sum](auto& fb){

auto [src, tgt] = fb.reduce(data.begin(), data.end(), sum, std::plus<int>());

fb.emplace([&sum] () { REQUIRE(sum == 0); }).precede(src);

tgt.precede(fb.emplace([&sum] () { REQUIRE(sum == 10); }));

for(size_t i=0; i<10; i ++){

++count;

}

auto n = fb.emplace([&count](auto& fb){

REQUIRE(count == 20);

++count;

auto prev = fb.emplace([&count](){

REQUIRE(count == 21);

++count;

});

for(size_t i=0; i<10; i++){

auto next = fb.emplace([&count, i](){

REQUIRE(count == 22+i);

++count;

});

prev.precede(next);

prev = next;

}

});

for(size_t i=0; i<10; i++){

fb.emplace([&count](){ ++count; }).precede(n);

}

});

A.precede(B);