From 18baa0f83c3c0a3bd6f0d9922b30632b40d53b4b Mon Sep 17 00:00:00 2001 From: XieHan Date: Tue, 10 May 2022 23:29:35 +0800 Subject: [PATCH 1/5] Add WFModuleTask --- src/factory/WFTask.h | 57 ++++++++++++++++++++++++++++++++++++- src/factory/WFTaskFactory.h | 9 ++++++ src/factory/Workflow.h | 12 +++++--- 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index a5de6dad..489ab9fe 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -299,12 +299,12 @@ protected: CommRequest(object, scheduler), callback(std::move(cb)) { - this->user_data = NULL; this->send_timeo = -1; this->receive_timeo = -1; this->keep_alive_timeo = 0; this->target = NULL; this->timeout_reason = TOR_NOT_TIMEOUT; + this->user_data = NULL; this->state = WFT_STATE_UNDEFINED; this->error = 0; } @@ -721,6 +721,61 @@ protected: virtual ~WFGoTask() { } }; +class WFModuleTask : protected SeriesWork, public ParallelTask +{ +public: + void start() + { + assert(!series_of(this)); + Workflow::start_series_work(this, nullptr); + } + + void dismiss() + { + assert(!series_of(this)); + this->dismiss_recursive(); + } + +public: + SeriesWork *as_series() { return this; } + + const SeriesWork *as_series() const { return this; } + +public: + void *user_data; + +protected: + virtual SubTask *done() + { + SeriesWork *series = series_of(this); + + if (this->callback) + this->callback(this); + + delete this; + return series->pop(); + } + +protected: + SubTask *subtask; + std::function callback; + +public: + WFModuleTask(SubTask *first, + std::function&& cb) : + SeriesWork(first, nullptr), + ParallelTask(&this->subtask, 1), + callback(std::move(cb)) + { + this->subtask = first; + this->set_in_parallel(); + this->user_data = NULL; + } + +protected: + virtual ~WFModuleTask() { } +}; + #include "WFTask.inl" #endif diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index e13e51a4..59816938 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -101,6 +101,8 @@ using WFEmptyTask = WFGenericTask; using WFDynamicTask = WFGenericTask; using dynamic_create_t = std::function; +using module_callback_t = std::function; + class WFTaskFactory { public: @@ -291,6 +293,13 @@ public: } static WFDynamicTask *create_dynamic_task(dynamic_create_t create); + +public: + static WFModuleTask *create_module_task(SubTask *first, + module_callback_t callback) + { + return new WFModuleTask(first, std::move(callback)); + } }; template diff --git a/src/factory/Workflow.h b/src/factory/Workflow.h index 855751d3..cf913703 100644 --- a/src/factory/Workflow.h +++ b/src/factory/Workflow.h @@ -105,7 +105,7 @@ public: } public: - /* The next 3 methods are intended for task implementations only. */ + /* The following functions are intended for task implementations only. */ SubTask *pop(); void set_last_task(SubTask *last) @@ -116,6 +116,10 @@ public: void unset_last_task() { this->last = NULL; } + void set_in_parallel() { this->in_parallel = true; } + + void dismiss_recursive(); + protected: void *context; series_callback_t callback; @@ -123,7 +127,6 @@ protected: private: SubTask *pop_task(); void expand_queue(); - void dismiss_recursive(); private: SubTask *first; @@ -237,13 +240,15 @@ public: protected: virtual SubTask *done(); +public: + void dismiss_recursive(); + protected: void *context; parallel_callback_t callback; private: void expand_buf(); - void dismiss_recursive(); private: size_t buf_size; @@ -254,7 +259,6 @@ protected: ParallelWork(SeriesWork *const all_series[], size_t n, parallel_callback_t&& callback); virtual ~ParallelWork() { delete []this->subtasks; } - friend class SeriesWork; friend class Workflow; }; From 0ad2b1c351adf5d118ff8f62b271161ac81bfbe3 Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 11 May 2022 03:41:39 +0800 Subject: [PATCH 2/5] Simplily 'Workflow' codes. Update WFModuleTask --- src/factory/WFTask.h | 24 +++++++++++++++++------- src/factory/Workflow.cc | 37 +++++++++++++------------------------ src/factory/Workflow.h | 14 +++++++------- 3 files changed, 37 insertions(+), 38 deletions(-) diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 489ab9fe..f66f000a 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -733,13 +733,13 @@ public: void dismiss() { assert(!series_of(this)); - this->dismiss_recursive(); + delete this; } public: - SeriesWork *as_series() { return this; } + SeriesWork *sub_series() { return this; } - const SeriesWork *as_series() const { return this; } + const SeriesWork *sub_series() const { return this; } public: void *user_data; @@ -752,28 +752,38 @@ protected: if (this->callback) this->callback(this); + this->first = NULL; delete this; return series->pop(); } protected: - SubTask *subtask; + SubTask *first; std::function callback; public: WFModuleTask(SubTask *first, std::function&& cb) : SeriesWork(first, nullptr), - ParallelTask(&this->subtask, 1), + ParallelTask(&this->first, 1), callback(std::move(cb)) { - this->subtask = first; + this->first = first; this->set_in_parallel(); this->user_data = NULL; } protected: - virtual ~WFModuleTask() { } + virtual ~WFModuleTask() + { + SubTask *task = this->first; + + while (task) + { + delete task; + task = this->pop_task(); + } + } }; #include "WFTask.inl" diff --git a/src/factory/Workflow.cc b/src/factory/Workflow.cc index 48db2857..32cdd385 100644 --- a/src/factory/Workflow.cc +++ b/src/factory/Workflow.cc @@ -47,13 +47,7 @@ void SeriesWork::dismiss_recursive() this->callback = nullptr; do { - ParallelWork *parallel = dynamic_cast(task); - - if (parallel) - parallel->dismiss_recursive(); - else - delete task; - + delete task; task = this->pop_task(); } while (task); } @@ -118,13 +112,7 @@ SubTask *SeriesWork::pop() while (task) { - ParallelWork *parallel = dynamic_cast(task); - - if (parallel) - parallel->dismiss_recursive(); - else - delete task; - + delete task; task = this->pop_task(); } @@ -218,16 +206,6 @@ void ParallelWork::add_series(SeriesWork *series) this->subtasks_nr++; } -void ParallelWork::dismiss_recursive() -{ - size_t i; - - for (i = 0; i < this->subtasks_nr; i++) - this->all_series[i]->dismiss_recursive(); - - delete this; -} - SubTask *ParallelWork::done() { SeriesWork *series = series_of(this); @@ -239,7 +217,18 @@ SubTask *ParallelWork::done() for (i = 0; i < this->subtasks_nr; i++) delete this->all_series[i]; + this->subtasks_nr = 0; delete this; return series->pop(); } +ParallelWork::~ParallelWork() +{ + size_t i; + + for (i = 0; i < this->subtasks_nr; i++) + this->all_series[i]->dismiss_recursive(); + + delete []this->subtasks; +} + diff --git a/src/factory/Workflow.h b/src/factory/Workflow.h index cf913703..a3f5b655 100644 --- a/src/factory/Workflow.h +++ b/src/factory/Workflow.h @@ -105,9 +105,11 @@ public: } public: - /* The following functions are intended for task implementations only. */ + /* Call only in tasks' done() implementation. */ SubTask *pop(); +public: + /* The following functions are intended for internal usages only. */ void set_last_task(SubTask *last) { last->set_pointer(this); @@ -118,6 +120,8 @@ public: void set_in_parallel() { this->in_parallel = true; } + SubTask *pop_task(); + void dismiss_recursive(); protected: @@ -125,7 +129,6 @@ protected: series_callback_t callback; private: - SubTask *pop_task(); void expand_queue(); private: @@ -205,7 +208,7 @@ public: void dismiss() { assert(!series_of(this)); - this->dismiss_recursive(); + delete this; } public: @@ -240,9 +243,6 @@ public: protected: virtual SubTask *done(); -public: - void dismiss_recursive(); - protected: void *context; parallel_callback_t callback; @@ -258,7 +258,7 @@ protected: ParallelWork(parallel_callback_t&& callback); ParallelWork(SeriesWork *const all_series[], size_t n, parallel_callback_t&& callback); - virtual ~ParallelWork() { delete []this->subtasks; } + virtual ~ParallelWork(); friend class Workflow; }; From ace21b44c581dcd3ba21bdc45f2b35ec391ec153 Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 11 May 2022 04:10:06 +0800 Subject: [PATCH 3/5] Fix WFModuleTask's dismiss interface --- src/factory/WFTask.h | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index f66f000a..3c37dfe5 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -778,10 +778,14 @@ protected: { SubTask *task = this->first; - while (task) + if (task) { - delete task; - task = this->pop_task(); + this->SeriesWork::callback = nullptr; + do + { + delete task; + task = this->pop_task(); + } while (task); } } }; From 87e52404b315549e342e4d621a84cbe60165a0ef Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 11 May 2022 15:55:24 +0800 Subject: [PATCH 4/5] Add a module task creator function --- src/factory/WFTaskFactory.h | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index 59816938..120eb7e0 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -300,6 +300,14 @@ public: { return new WFModuleTask(first, std::move(callback)); } + + static WFModuleTask *create_module_task(SubTask *first, SubTask *last, + module_callback_t callback) + { + WFModuleTask *task = new WFModuleTask(first, std::move(callback)); + task->sub_series()->set_last_task(last); + return task; + } }; template From 3508061607ea14665ea7efe240b0eb92769b1554 Mon Sep 17 00:00:00 2001 From: XieHan Date: Wed, 11 May 2022 17:00:30 +0800 Subject: [PATCH 5/5] add an assert() --- src/factory/Workflow.cc | 1 + src/factory/Workflow.h | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/factory/Workflow.cc b/src/factory/Workflow.cc index 32cdd385..dd3cbe7c 100644 --- a/src/factory/Workflow.cc +++ b/src/factory/Workflow.cc @@ -33,6 +33,7 @@ SeriesWork::SeriesWork(SubTask *first, series_callback_t&& cb) : this->back = 0; this->in_parallel = false; this->canceled = false; + assert(!series_of(first)); first->set_pointer(this); this->first = first; this->last = NULL; diff --git a/src/factory/Workflow.h b/src/factory/Workflow.h index a3f5b655..e9bf953b 100644 --- a/src/factory/Workflow.h +++ b/src/factory/Workflow.h @@ -105,7 +105,7 @@ public: } public: - /* Call only in tasks' done() implementation. */ + /* Call only in tasks' done() implementations. */ SubTask *pop(); public: