From 2a3e69f5afc250ebfea22a275a58440c81c84f2c Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Thu, 15 Oct 2020 16:26:33 +0800 Subject: [PATCH 1/2] Add WFGraphTask for simply building DAGs (#101) * add WFGraphTask * Update WFGraphNode Co-authored-by: XieHan --- CMakeLists_Headers.txt | 1 + src/factory/CMakeLists.txt | 1 + src/factory/WFGraphTask.cc | 95 ++++++++++++++++++++++++++++++++ src/factory/WFGraphTask.h | 106 ++++++++++++++++++++++++++++++++++++ src/factory/WFTaskFactory.h | 12 +++- src/factory/Workflow.h | 2 +- 6 files changed, 215 insertions(+), 2 deletions(-) create mode 100644 src/factory/WFGraphTask.cc create mode 100644 src/factory/WFGraphTask.h diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index f2fe307f..6e990558 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -74,6 +74,7 @@ set(INCLUDE_HEADERS src/factory/WFConnection.h src/factory/WFTask.h src/factory/WFTask.inl + src/factory/WFGraphTask.h src/factory/WFTaskError.h src/factory/WFTaskFactory.h src/factory/WFTaskFactory.inl diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index 53127c35..64ed8982 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -2,6 +2,7 @@ cmake_minimum_required(VERSION 3.6) project(factory) set(SRC + WFGraphTask.cc HttpTaskImpl.cc RedisTaskImpl.cc MySQLTaskImpl.cc diff --git a/src/factory/WFGraphTask.cc b/src/factory/WFGraphTask.cc new file mode 100644 index 00000000..1fec7324 --- /dev/null +++ b/src/factory/WFGraphTask.cc @@ -0,0 +1,95 @@ +/* + Copyright (c) 2020 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Xie Han (xiehan@sogou-inc.com) +*/ + +#include +#include "Workflow.h" +#include "WFGraphTask.h" + +SubTask *WFGraphNode::done() +{ + SeriesWork *series = series_of(this); + + if (!this->user_data) + { + this->value = 1; + this->user_data = (void *)1; + } + else + delete this; + + return series->pop(); +} + +WFGraphNode::~WFGraphNode() +{ + if (this->user_data) + { + for (WFGraphNode *node : this->successors) + node->WFCounterTask::count(); + } +} + +WFGraphNode& WFGraphTask::create_graph_node(SubTask *task) +{ + WFGraphNode *node = new WFGraphNode; + SeriesWork *series = Workflow::create_series_work(node, node, nullptr); + + series->push_back(task); + this->parallel->add_series(series); + return *node; +} + +SubTask *WFGraphTask::done() +{ + SeriesWork *series = series_of(this); + ParallelWork *parallel = this->parallel; + + if (parallel) + { + this->parallel = NULL; + series->push_front(this); + series->push_front(parallel); + } + else + { + if (this->callback) + this->callback(this); + + delete this; + } + + return series->pop(); +} + +WFGraphTask::~WFGraphTask() +{ + SeriesWork *series; + size_t i; + + if (this->parallel) + { + for (i = 0; i < this->parallel->size(); i++) + { + series = const_cast(this->parallel->series_at(i)); + series->set_last_task(new WFGenericTask); + } + + this->parallel->dismiss(); + } +} + diff --git a/src/factory/WFGraphTask.h b/src/factory/WFGraphTask.h new file mode 100644 index 00000000..920d355f --- /dev/null +++ b/src/factory/WFGraphTask.h @@ -0,0 +1,106 @@ +/* + Copyright (c) 2020 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Xie Han (xiehan@sogou-inc.com) +*/ + +#ifndef _WFGRAPHTASK_H_ +#define _WFGRAPHTASK_H_ + +#include +#include +#include +#include "Workflow.h" +#include "WFTask.h" + +class WFGraphNode : protected WFCounterTask +{ +public: + void precede(WFGraphNode& node) + { + node.value++; + this->successors.push_back(&node); + } + + void succeed(WFGraphNode& node) + { + node.precede(*this); + } + +protected: + virtual SubTask *done(); + +protected: + std::vector successors; + +protected: + WFGraphNode() : WFCounterTask(0, nullptr) { } + virtual ~WFGraphNode(); + friend class WFGraphTask; +}; + +static inline WFGraphNode& operator --(WFGraphNode& node, int) +{ + return node; +} + +static inline WFGraphNode& operator > (WFGraphNode& prec, WFGraphNode& succ) +{ + prec.precede(succ); + return succ; +} + +static inline WFGraphNode& operator < (WFGraphNode& succ, WFGraphNode& prec) +{ + succ.succeed(prec); + return prec; +} + +static inline WFGraphNode& operator --(WFGraphNode& node) +{ + return node; +} + +class WFGraphTask : public WFGenericTask +{ +public: + WFGraphNode& create_graph_node(SubTask *task); + +public: + void set_callback(std::function cb) + { + this->callback = std::move(cb); + } + +protected: + virtual SubTask *done(); + +protected: + ParallelWork *parallel; + std::function callback; + +public: + WFGraphTask(std::function&& cb) : + callback(std::move(cb)) + { + this->parallel = Workflow::create_parallel_work(nullptr); + } + +protected: + virtual ~WFGraphTask(); +}; + +#endif + diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index 6f8ccb65..436f8ead 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -26,8 +26,9 @@ #include "HttpMessage.h" #include "MySQLMessage.h" #include "DNSRoutine.h" -#include "WFTask.h" #include "Workflow.h" +#include "WFTask.h" +#include "WFGraphTask.h" #include "EndpointParams.h" #include "WFAlgoTaskFactory.h" @@ -81,6 +82,9 @@ using fsync_callback_t = std::function; using timer_callback_t = std::function; using counter_callback_t = std::function; +// Graph (DAG) task. +using graph_callback_t = std::function; + // DNS task. For internal usage only. using WFDNSTask = WFThreadTask; using dns_callback_t = std::function; @@ -197,6 +201,12 @@ public: static WFGoTask *create_go_task(const std::string& queue_name, FUNC&& func, ARGS&&... args); +public: + static WFGraphTask *create_graph_task(graph_callback_t callback) + { + return new WFGraphTask(std::move(callback)); + } + public: static WFDNSTask *create_dns_task(const std::string& host, unsigned short port, diff --git a/src/factory/Workflow.h b/src/factory/Workflow.h index ae46edca..f7cdb7cf 100644 --- a/src/factory/Workflow.h +++ b/src/factory/Workflow.h @@ -105,9 +105,9 @@ public: } public: + /* pop() and set_task_task() are intended for framework providers only. */ SubTask *pop(); -private: void set_last_task(SubTask *last) { last->set_pointer(this); From c98a73ad7b1899705edf387f4d6bc0d55500ba5c Mon Sep 17 00:00:00 2001 From: Sonullx Liu Date: Fri, 16 Oct 2020 17:50:47 +0800 Subject: [PATCH 2/2] Update unit tests (#104) * test: correct TestSuitName and TestName * test: add graph_unittest.cc Add unittests of WFGraphTask. --- test/CMakeLists.txt | 1 + test/algo_unittest.cc | 4 +- test/facilities_unittest.cc | 12 +++--- test/graph_unittest.cc | 85 +++++++++++++++++++++++++++++++++++++ test/http_unittest.cc | 6 +-- test/mysql_unittest.cc | 2 +- test/redis_unittest.cc | 4 +- test/task_unittest.cc | 12 +++--- 8 files changed, 106 insertions(+), 20 deletions(-) create mode 100644 test/graph_unittest.cc diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index e362bec0..83c579ff 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -35,6 +35,7 @@ set(TEST_LIST redis_unittest mysql_unittest facilities_unittest + graph_unittest ) if (APPLE) diff --git a/test/algo_unittest.cc b/test/algo_unittest.cc index 5a55e886..f3c0533e 100644 --- a/test/algo_unittest.cc +++ b/test/algo_unittest.cc @@ -35,7 +35,7 @@ static void __arr_check(int *arr, int n) EXPECT_LE(arr[i - 1], arr[i]); } -TEST(sort, algo_unittest) +TEST(algo_unittest, sort) { static constexpr int n = 100000; int *arr = new int[n]; @@ -65,7 +65,7 @@ TEST(sort, algo_unittest) delete []arr; } -TEST(parallel_sort, algo_unittest) +TEST(algo_unittest, parallel_sort) { static constexpr int n = 100000; int *arr = new int[n]; diff --git a/test/facilities_unittest.cc b/test/facilities_unittest.cc index 394540f3..2a4243fb 100644 --- a/test/facilities_unittest.cc +++ b/test/facilities_unittest.cc @@ -26,7 +26,7 @@ #define GET_CURRENT_MICRO std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() -TEST(usleep, facilities_unittest) +TEST(facilities_unittest, usleep) { int64_t st = GET_CURRENT_MICRO; WFFacilities::usleep(1000000); @@ -34,7 +34,7 @@ TEST(usleep, facilities_unittest) EXPECT_LE(ed - st, 10000000) << "usleep too slow"; } -TEST(async_usleep, facilities_unittest) +TEST(facilities_unittest, async_usleep) { int64_t st = GET_CURRENT_MICRO; WFFacilities::async_usleep(1000000).wait(); @@ -42,7 +42,7 @@ TEST(async_usleep, facilities_unittest) EXPECT_LE(ed - st, 10000000) << "async_usleep too slow"; } -TEST(request, facilities_unittest) +TEST(facilities_unittest, request) { protocol::HttpRequest req; req.set_method(HttpMethodGet); @@ -62,7 +62,7 @@ TEST(request, facilities_unittest) } } -TEST(async_request, facilities_unittest) +TEST(facilities_unittest, async_request) { protocol::HttpRequest req; req.set_method(HttpMethodGet); @@ -78,7 +78,7 @@ TEST(async_request, facilities_unittest) } } -TEST(fileIO, facilities_unittest) +TEST(facilities_unittest, fileIO) { uint64_t data = 0x1234; ssize_t sz; @@ -98,7 +98,7 @@ static inline void f(int i, WFFacilities::WaitGroup *wg) wg->done(); } -TEST(WaitGroup, facilities_unittest) +TEST(facilities_unittest, WaitGroup) { WFFacilities::WaitGroup wg(100); diff --git a/test/graph_unittest.cc b/test/graph_unittest.cc new file mode 100644 index 00000000..cf236913 --- /dev/null +++ b/test/graph_unittest.cc @@ -0,0 +1,85 @@ +/* + Copyright (c) 2020 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Liu Yang (liuyang216492@sogou-inc.com) +*/ + +#include +#include + +#include "workflow/WFTaskFactory.h" +#include "workflow/WFFacilities.h" + +static SubTask *create_task(int& target) +{ + static std::atomic generator; + return WFTaskFactory::create_timer_task(0, [&](WFTimerTask *) + { + target = generator++; + }); +} + +TEST(graph_unittest, WFGraphTask1) +{ + WFFacilities::WaitGroup wait_group(1); + + auto graph = WFTaskFactory::create_graph_task([&wait_group](WFGraphTask *){ wait_group.done(); }); + + int ta, tb, tc, td; + + auto& a = graph->create_graph_node(create_task(ta)); + auto& b = graph->create_graph_node(create_task(tb)); + auto& c = graph->create_graph_node(create_task(tc)); + auto& d = graph->create_graph_node(create_task(td)); + + a --> b <-- c --> d --> a; + c --> a; + + graph->start(); + wait_group.wait(); + + EXPECT_LT(ta, tb); + EXPECT_LT(tc, tb); + EXPECT_LT(tc, td); + EXPECT_LT(td, ta); + EXPECT_LT(tc, ta); +} + +TEST(graph_unittest, WFGraphTask2) +{ + WFFacilities::WaitGroup wait_group(1); + + auto graph = WFTaskFactory::create_graph_task([&wait_group](WFGraphTask *){ wait_group.done(); }); + + constexpr int N = 4096 - 1; + + auto target = new int[N]; + auto node = new WFGraphNode *[N]; + + for (int i = 0; i < N; i++) + node[i] = &graph->create_graph_node(create_task(target[i])); + + for (int i = 1; i < N; i++) + node[i]->precede(*node[(i - 1) / 2]); + + graph->start(); + wait_group.wait(); + + for (int i = 1; i < N; i++) + EXPECT_LT(target[i], target[(i - 1) / 2]); + + delete[] target; + delete[] node; +} diff --git a/test/http_unittest.cc b/test/http_unittest.cc index 060dfeab..2e0934c1 100644 --- a/test/http_unittest.cc +++ b/test/http_unittest.cc @@ -36,7 +36,7 @@ static void __http_process(WFHttpTask *task) resp->add_header_pair("Content-Type", "text/plain"); } -TEST(WFHttpTask1, http_unittest) +TEST(http_unittest, WFHttpTask1) { std::mutex mutex; std::condition_variable cond; @@ -69,7 +69,7 @@ TEST(WFHttpTask1, http_unittest) lock.unlock(); } -TEST(WFHttpTask2, http_unittest) +TEST(http_unittest, WFHttpTask2) { std::mutex mutex; std::condition_variable cond; @@ -98,7 +98,7 @@ TEST(WFHttpTask2, http_unittest) lock.unlock(); } -TEST(WFHttpTask3, http_unittest) +TEST(http_unittest, WFHttpTask3) { FILE *f; f = fopen("server.crt", "w"); diff --git a/test/mysql_unittest.cc b/test/mysql_unittest.cc index 29f186b1..a2d353f9 100644 --- a/test/mysql_unittest.cc +++ b/test/mysql_unittest.cc @@ -48,7 +48,7 @@ static void test_client(const char *url, const char *sql, std::mutex& mutex, std task->start(); } -TEST(WFMySQLTask1, mysql_unittest) +TEST(mysql_unittest, WFMySQLTask1) { std::mutex mutex; std::condition_variable cond; diff --git a/test/redis_unittest.cc b/test/redis_unittest.cc index 62664589..69bc19a9 100644 --- a/test/redis_unittest.cc +++ b/test/redis_unittest.cc @@ -136,7 +136,7 @@ static void test_client(const char *url, std::mutex& mutex, std::condition_varia flow.start(); } -TEST(WFRedisTask1, redis_unittest) +TEST(redis_unittest, WFRedisTask1) { std::mutex mutex; std::condition_variable cond; @@ -153,7 +153,7 @@ TEST(WFRedisTask1, redis_unittest) server.stop(); } -TEST(WFRedisTask2, redis_unittest) +TEST(redis_unittest, WFRedisTask2) { std::mutex mutex; std::condition_variable cond; diff --git a/test/task_unittest.cc b/test/task_unittest.cc index 3df54a6a..a85f1250 100644 --- a/test/task_unittest.cc +++ b/test/task_unittest.cc @@ -31,7 +31,7 @@ #define GET_CURRENT_MICRO std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() -TEST(WFTimerTask, task_unittest) +TEST(task_unittest, WFTimerTask) { std::mutex mutex; std::condition_variable cond; @@ -56,7 +56,7 @@ TEST(WFTimerTask, task_unittest) EXPECT_LE(ed - st, 10000000) << "Timer Task too slow"; } -TEST(WFCounterTask1, task_unittest) +TEST(task_unittest, WFCounterTask1) { std::mutex mutex; std::condition_variable cond; @@ -92,7 +92,7 @@ TEST(WFCounterTask1, task_unittest) lock.unlock(); } -TEST(WFCounterTask2, task_unittest) +TEST(task_unittest, WFCounterTask2) { std::mutex mutex; std::condition_variable cond; @@ -125,7 +125,7 @@ TEST(WFCounterTask2, task_unittest) lock.unlock(); } -TEST(WFGoTask, task_unittest) +TEST(task_unittest, WFGoTask) { srand(time(NULL)); std::mutex mutex; @@ -155,7 +155,7 @@ TEST(WFGoTask, task_unittest) EXPECT_EQ(edit_inner, 100); } -TEST(WFThreadTask, task_unittest) +TEST(task_unittest, WFThreadTask) { std::mutex mutex; std::condition_variable cond; @@ -195,7 +195,7 @@ TEST(WFThreadTask, task_unittest) lock.unlock(); } -TEST(WFFileIOTask, task_unittest) +TEST(task_unittest, WFFileIOTask) { srand(time(NULL)); std::mutex mutex;