From e884a6902ccbd626c486920454412b1ee5b6f850 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Thu, 15 Oct 2020 16:40:01 +0800 Subject: [PATCH 1/3] Add WFGraphTask to windows branch. (#103) * Add WFGraphTask 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/Workflow.h | 2 +- 5 files changed, 204 insertions(+), 1 deletion(-) 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 17ee42f8..24d75827 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -87,6 +87,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 2cbcecd3..aac274e2 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/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 5606f23ef0310af5ccf1881d9ce1703a76676581 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Sun, 18 Oct 2020 01:09:42 +0800 Subject: [PATCH 2/3] Update Windows branch (#109) ]* fix EncodeStream bug Co-authored-by: XieHan --- src/util/EncodeStream.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/util/EncodeStream.cc b/src/util/EncodeStream.cc index fde4b480..fa01f91c 100644 --- a/src/util/EncodeStream.cc +++ b/src/util/EncodeStream.cc @@ -70,7 +70,7 @@ void EncodeStream::check_merge() next = pos->next; } - while (pos != head) + if (pos != head) { __buff *x = list_entry(pos, __buff, buffer_list); From b8f12881a4c389db2615783f8a7a7083477059ec Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 19 Oct 2020 01:30:30 +0800 Subject: [PATCH 3/3] Windows (#110) * update WFAlgoTaskFactory Co-authored-by: XieHan --- src/factory/WFAlgoTaskFactory.h | 6 +++++- src/factory/WFAlgoTaskFactory.inl | 13 +++++++++---- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/src/factory/WFAlgoTaskFactory.h b/src/factory/WFAlgoTaskFactory.h index 6c33d0b8..88722c30 100644 --- a/src/factory/WFAlgoTaskFactory.h +++ b/src/factory/WFAlgoTaskFactory.h @@ -36,7 +36,11 @@ struct SortInput }; template -using SortOutput = SortInput; +struct SortOutput +{ + T *first; + T *last; +}; template struct MergeInput diff --git a/src/factory/WFAlgoTaskFactory.inl b/src/factory/WFAlgoTaskFactory.inl index f94ca6de..deff567b 100644 --- a/src/factory/WFAlgoTaskFactory.inl +++ b/src/factory/WFAlgoTaskFactory.inl @@ -33,7 +33,8 @@ protected: virtual void execute() { std::sort(this->input.first, this->input.last); - this->output = this->input; + this->output.first = this->input.first; + this->output.last = this->input.last; } public: @@ -67,6 +68,7 @@ public: this->input.last2 = last2; this->input.d_first = d_first; this->output.first = NULL; + this->output.last = NULL; } }; @@ -173,7 +175,8 @@ void __WFParSortTask::execute() T *middle = this->input.first + n / 2; std::inplace_merge(this->input.first, middle, this->input.last); - this->output = this->input; + this->output.first = this->input.first; + this->output.last = this->input.last; this->flag = 0; } else @@ -190,7 +193,8 @@ protected: { std::sort(this->input.first, this->input.last, std::move(this->compare)); - this->output = this->input; + this->output.first = this->input.first; + this->output.last = this->input.last; } protected: @@ -335,7 +339,8 @@ void __WFParSortTaskCmp::execute() std::inplace_merge(this->input.first, middle, this->input.last, std::move(this->compare)); - this->output = this->input; + this->output.first = this->input.first; + this->output.last = this->input.last; this->flag = 0; } else