From 908c016ecd2bb15ab1098b6abcf7e22b82ff486c Mon Sep 17 00:00:00 2001 From: XieHan Date: Sun, 29 May 2022 03:33:45 +0800 Subject: [PATCH 1/9] rename __WFConditional->__RPConditional --- src/factory/WFResourcePool.cc | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/factory/WFResourcePool.cc b/src/factory/WFResourcePool.cc index d01219b7..865bae33 100644 --- a/src/factory/WFResourcePool.cc +++ b/src/factory/WFResourcePool.cc @@ -22,7 +22,7 @@ #include "WFTask.h" #include "WFResourcePool.h" -class __WFConditional : public WFConditional +class __RPConditional : public WFConditional { public: struct list_head list; @@ -33,14 +33,14 @@ public: virtual void signal(void *res) { } public: - __WFConditional(SubTask *task, void **resbuf, + __RPConditional(SubTask *task, void **resbuf, struct WFResourcePool::Data *data) : WFConditional(task, resbuf) { this->data = data; } - __WFConditional(SubTask *task, + __RPConditional(SubTask *task, struct WFResourcePool::Data *data) : WFConditional(task) { @@ -48,7 +48,7 @@ public: } }; -void __WFConditional::dispatch() +void __RPConditional::dispatch() { struct WFResourcePool::Data *data = this->data; @@ -64,12 +64,12 @@ void __WFConditional::dispatch() WFConditional *WFResourcePool::get(SubTask *task, void **resbuf) { - return new __WFConditional(task, resbuf, &this->data); + return new __RPConditional(task, resbuf, &this->data); } WFConditional *WFResourcePool::get(SubTask *task) { - return new __WFConditional(task, &this->data); + return new __RPConditional(task, &this->data); } void WFResourcePool::create(size_t n) @@ -101,7 +101,7 @@ void WFResourcePool::post(void *res) data->mutex.lock(); if (++data->value <= 0) { - cond = list_entry(data->wait_list.next, __WFConditional, list); + cond = list_entry(data->wait_list.next, __RPConditional, list); list_del(data->wait_list.next); } else From e09445732830ad7ac05f9916b25c35c7989d4998 Mon Sep 17 00:00:00 2001 From: XieHan Date: Sun, 29 May 2022 03:34:48 +0800 Subject: [PATCH 2/9] add named conditional (observer mode) --- src/factory/WFTaskFactory.cc | 239 +++++++++++++++++++++++++++++++++++ src/factory/WFTaskFactory.h | 8 ++ 2 files changed, 247 insertions(+) diff --git a/src/factory/WFTaskFactory.cc b/src/factory/WFTaskFactory.cc index a1546547..950d19f8 100644 --- a/src/factory/WFTaskFactory.cc +++ b/src/factory/WFTaskFactory.cc @@ -343,3 +343,242 @@ WFMailboxTask *WFTaskFactory::create_mailbox_task(mailbox_callback_t callback) { return new WFMailboxTask(std::move(callback)); } + +/****************** Named Conditional ******************/ + +class __WFConditional; + +struct __conditional_node +{ + struct list_head list; + __WFConditional *cond; +}; + +struct __ConditionalList +{ + __ConditionalList(const std::string& str): + name(str) + { + INIT_LIST_HEAD(&this->head); + } + + void push_back(struct __conditional_node *node) + { + list_add_tail(&node->list, &this->head); + } + + bool empty() const + { + return list_empty(&this->head); + } + + void del(struct __conditional_node *node) + { + list_del(&node->list); + } + + struct rb_node rb; + struct list_head head; + std::string name; + friend class __ConditionalMap; +}; + +class __ConditionalMap +{ +public: + static __ConditionalMap *get_instance() + { + static __ConditionalMap kInstance; + return &kInstance; + } + + WFConditional *create(const std::string& name, + SubTask *task, void **msgbuf); + + WFConditional *create(const std::string& name, SubTask *task); + + void signal(const std::string& name, void *msg); + void signal(struct __ConditionalList *conds, + struct __conditional_node *node, + void *msg); + void remove(struct __ConditionalList *conds, + struct __conditional_node *node); + +private: + struct __ConditionalList *get_list(const std::string& name); + + __ConditionalMap() + { + conds_map_.rb_node = NULL; + } + + struct rb_root conds_map_; + std::mutex mutex_; +}; + +class __WFConditional : public WFConditional +{ +public: + __WFConditional(SubTask *task, void **msgbuf, + struct __ConditionalList *conds) : + WFConditional(task, msgbuf), + conds_(conds) + { + node_.cond = this; + conds_->push_back(&node_); + } + + __WFConditional(SubTask *task, struct __ConditionalList *conds) : + WFConditional(task), + conds_(conds) + { + node_.cond = this; + conds_->push_back(&node_); + } + + virtual ~__WFConditional() + { + if (!this->flag) + __ConditionalMap::get_instance()->remove(conds_, &node_); + } + + virtual void signal(void *msg) + { + __ConditionalMap::get_instance()->signal(conds_, &node_, msg); + } + +private: + struct __conditional_node node_; + struct __ConditionalList *conds_; + friend class __ConditionalMap; +}; + +WFConditional *__ConditionalMap::create(const std::string& name, + SubTask *task, void **msgbuf) +{ + std::lock_guard lock(mutex_); + struct __ConditionalList *conds = get_list(name); + return new __WFConditional(task, msgbuf, conds); +} + +WFConditional *__ConditionalMap::create(const std::string& name, + SubTask *task) +{ + std::lock_guard lock(mutex_); + struct __ConditionalList *conds = get_list(name); + return new __WFConditional(task, conds); +} + +struct __ConditionalList *__ConditionalMap::get_list(const std::string& name) +{ + struct rb_node **p = &conds_map_.rb_node; + struct rb_node *parent = NULL; + struct __ConditionalList *conds; + + while (*p) + { + parent = *p; + conds = rb_entry(*p, struct __ConditionalList, rb); + if (name < conds->name) + p = &(*p)->rb_left; + else if (name > conds->name) + p = &(*p)->rb_right; + else + break; + } + + if (*p == NULL) + { + conds = new struct __ConditionalList(name); + rb_link_node(&conds->rb, parent, p); + rb_insert_color(&conds->rb, &conds_map_); + } + + return conds; +} + +void __ConditionalMap::signal(const std::string& name, void *msg) +{ + struct rb_node **p = &conds_map_.rb_node; + struct __ConditionalList *conds = NULL; + + mutex_.lock(); + while (*p) + { + conds = rb_entry(*p, struct __ConditionalList, rb); + + if (name < conds->name) + p = &(*p)->rb_left; + else if (name > conds->name) + p = &(*p)->rb_right; + else + { + rb_erase(&conds->rb, &conds_map_); + break; + } + } + + mutex_.unlock(); + if (!conds) + return; + + struct list_head *pos; + struct list_head *tmp; + struct __conditional_node *node; + + list_for_each_safe(pos, tmp, &conds->head) + { + node = list_entry(pos, struct __conditional_node, list); + node->cond->WFConditional::signal(msg); + } + + delete conds; +} + +void __ConditionalMap::signal(struct __ConditionalList *conds, + struct __conditional_node *node, + void *msg) +{ + mutex_.lock(); + conds->del(node); + if (conds->empty()) + { + rb_erase(&conds->rb, &conds_map_); + delete conds; + } + + mutex_.unlock(); + node->cond->WFConditional::signal(msg); +} + +void __ConditionalMap::remove(struct __ConditionalList *conds, + struct __conditional_node *node) +{ + mutex_.lock(); + conds->del(node); + if (conds->empty()) + { + rb_erase(&conds->rb, &conds_map_); + delete conds; + } + + mutex_.unlock(); +} + +WFConditional *WFTaskFactory::create_conditional(const std::string& cond_name, + SubTask *task, void **msgbuf) +{ + return __ConditionalMap::get_instance()->create(cond_name, task, msgbuf); +} + +WFConditional *WFTaskFactory::create_conditional(const std::string& cond_name, + SubTask *task) +{ + return __ConditionalMap::get_instance()->create(cond_name, task); +} + +void WFTaskFactory::signal_by_name(const std::string& cond_name, void *msg) +{ + __ConditionalMap::get_instance()->signal(cond_name, msg); +} + diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index 120eb7e0..046be6bc 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -275,6 +275,14 @@ public: return new WFConditional(task); } + static WFConditional *create_conditional(const std::string& cond_name, + SubTask *task, void **msgbuf); + + static WFConditional *create_conditional(const std::string& cond_name, + SubTask *task); + + static void signal_by_name(const std::string& cond_name, void *msg); + public: template static WFGoTask *create_go_task(const std::string& queue_name, From 9b96ae40d3a67badf9ac09936fd843c1907797e6 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 01:02:10 +0800 Subject: [PATCH 3/9] Rename about-conditional.md to about-resourcepool.md --- docs/{about-conditional.md => about-resourcepool.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/{about-conditional.md => about-resourcepool.md} (100%) diff --git a/docs/about-conditional.md b/docs/about-resourcepool.md similarity index 100% rename from docs/about-conditional.md rename to docs/about-resourcepool.md From d8e7a84e5f8a647aeffc727f08ef8722e0a6c4a3 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 01:02:41 +0800 Subject: [PATCH 4/9] Rename about-resourcepool.md to about-resource-pool.md --- docs/{about-resourcepool.md => about-resource-pool.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/{about-resourcepool.md => about-resource-pool.md} (100%) diff --git a/docs/about-resourcepool.md b/docs/about-resource-pool.md similarity index 100% rename from docs/about-resourcepool.md rename to docs/about-resource-pool.md From 796e9d817186a166a38fa11d0579f95215b5bfa4 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 01:03:33 +0800 Subject: [PATCH 5/9] Update README_cn.md --- README_cn.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README_cn.md b/README_cn.md index 8b785980..f0ca7d92 100644 --- a/README_cn.md +++ b/README_cn.md @@ -85,7 +85,7 @@ make * 其它一些重要任务与组件 * [关于定时器](docs/about-timer.md) * [关于计数器](docs/about-counter.md) - * [条件任务与资源池](docs/about-conditional.md) + * [关于资源池](docs/about-resource-pool.md) * [模块任务](/docs/about-module.md) * [DAG图任务](/docs/tutorial-11-graph_task.md) * 服务治理 From a6456261140157a5b62d475c9ec83ec08fc78d9b Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 01:04:16 +0800 Subject: [PATCH 6/9] Update README.md --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 7a97ffe0..ae9271df 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,7 @@ sudo apt-get install libworkflow1 * Timing tasks and counting tasks * [About timer](docs/en/about-timer.md) * [About counter](docs/en/about-counter.md) - * [Conditional and resource pool](docs/en/about-conditional.md) + * [About resource pool](docs/en/about-resource-pool.md) * [About module](docs/en/about-module.md) * Service governance * [About service governance](docs/en/about-service-governance.md) From 1bf613f6e6bc0f452f2bdf00ac69327766b0b2b9 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 01:05:10 +0800 Subject: [PATCH 7/9] Rename about-conditional.md to about-resource-pool.md --- docs/en/{about-conditional.md => about-resource-pool.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/en/{about-conditional.md => about-resource-pool.md} (100%) diff --git a/docs/en/about-conditional.md b/docs/en/about-resource-pool.md similarity index 100% rename from docs/en/about-conditional.md rename to docs/en/about-resource-pool.md From d2d6d6db3d1fa2c20315fd54229de5b5196a7756 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 02:13:17 +0800 Subject: [PATCH 8/9] Create about-conditional.md --- docs/about-conditional.md | 104 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 docs/about-conditional.md diff --git a/docs/about-conditional.md b/docs/about-conditional.md new file mode 100644 index 00000000..24c0d889 --- /dev/null +++ b/docs/about-conditional.md @@ -0,0 +1,104 @@ +# 条件任务与观察者模式 + +有的时候，我们需要让任务在某个条件下才被执行。条件任务（WFConditional）就是用于解决这种问题。 +条件任务是一种任务包装器，可以包装任何的任务并取代原任务。通过对条件任务发送信号来触发被包装任务的执行。 + +# 条件任务的创建 +在[WFTaskFactory.h](/src/factory/WFTaskFactory.h)里，可以看到条件任务的创建接口。 +~~~cpp +class WFTaskFactory +{ +public: + static WFConditional *create_conditional(SubTask *task); + static WFConditional *create_conditional(SubTask *task, void **msgbuf); +}; +~~~ +可以看到，我们通过工厂的create_conditional接口创建条件任务。 +其中，task为被包装的任务。msgbuf是用于接收消息的缓冲区，如果无需关注消息的具体内容，msgbuf可以缺省。 +WFConditional的主要接口： +~~~cpp +class WFConditional : public WFGenericTask +{ +public: + virtual void signal(void *msg); + ... +}; +~~~ +WFConditional是一种任务，所以，它满足普通workflow任务的一切属性。特别的接口只有signal，用于发送信号。 + +# 示例 + +以下示例，通过timer和conditional，实现一个延迟1秒执行的计算任务。 +~~~cpp +int main() +{ + WFGoTask *task = WFTaskFactory::create_go_task("test", [](){ printf("Done\n"); }); + WFConditional *cond = WFTaskFactory::create_conditional(task); + WFTimerTask *timer = WFTaskFactory::create_timer_task(1, 0, [cond](void *){ + cond->signal(NULL); + }); + timer->start(); + cond->start(); + getchar(); +} +~~~ +这个示例里，在定时器的回调里向cond发送信号，让被包装的go task可以被执行。 +注意，无论cond->signal()与cond->start()哪一个先被调用，程序都完全正确。 + +# 观察者模式 + +我们看到，如果直接对cond发送信息，需要发送者直接持有cond的指针，这在一些情况下并不是很方便。 +于是，我们引入了观察者模式，也就是命名的条件任务。通过向某个名称发送信号，同时唤醒所有在这个名称下的条件任务。 +命名条件任务的创建与唤醒： +~~~cpp +class WFTaskFactory +{ +public: + static WFConditional *create_conditional(const std::string& cond_name, SubTask *task); + static WFConditional *create_conditional(const std::string& cond_name, SubTask *task, void **msgbuf); + static void signal_by_name(const std::string& cond_name, void *msg); +}; +~~~ +我们看到，与普通条件任务唯一区别是，命名条件任务创建时，需要传入一个cond_name。 +而signal_by_name()接口，将msg发送到所有在这个名称上等待的条件任务，将它们全部唤醒。这就相当于实现了观察者模式。 +# 示例 + +还是上面的延迟计算示例，我们增加到两个计算任务并用观察者模式来实现。用”slot1”作为条件任务名。 +~~~cpp +int main() +{ + WFGoTask *task1 = WFTaskFactory::create_go_task("test”, [](){ printf(“test1 done\n"); }); + WFGoTask *task2 = WFTaskFactory::create_go_task("test”, [](){ printf(“test2 done\n"); }); + WFConditional *cond1 = WFTaskFactory::create_conditional(“slot1”, task1); + WFConditional *cond2 = WFTaskFactory::create_conditional(“slot1”, task2); + WFTimerTask *timer = WFTaskFactory::create_timer_task(1, 0, [](void *){ + WFTaskFactory::signal_by_name(“slot1”, NULL); + }); + timer->start(); + cond1->start(); + cond2->start(); + getchar(); +} +~~~ +我们看到，在这个示例里，timer在回调中通过signal_by_name方法，同时唤醒了slot1下两个计算任务。 + +# 使用条件任务注意事项 + +Workflow里的任何任务，如果创建之后不想运行，都可以通过dismiss接口直接释放。 +对于条件任务，如果要被dismiss（或者在某个被cancel的series里），必须保证这个条件任务没有被signal过。 +以下代码的行为无定义： +~~~cpp +int main() +{ + WFEmptyTask *task = WFTaskFactory::create_empty_task(); + WFConditional *cond = WFTaskFactory::create_conditional(“slot1”, task); + WFTimerTask *timer = WFTaskFactory::create_timer_task(0, 0, [](void *) { + WFTaskFactory::signal_by_name(“slot1”); + }); + timer->start(); + cond->dismiss(); // 取消任务 + getchar(); +} +~~~ +显然，如果timer的callback里已经执行或正在执行了signal_by_name，cond被signal，再dismiss()是一种错误行为。 +这种情况一般也只会出现在命名条件任务里。所以，dismiss一个命名条件任务，需要特别的小心。 From 185013171f4205b72bfe607b0936baef41699ba7 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 30 May 2022 02:14:10 +0800 Subject: [PATCH 9/9] Update README_cn.md --- README_cn.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README_cn.md b/README_cn.md index f0ca7d92..e10fdf0f 100644 --- a/README_cn.md +++ b/README_cn.md @@ -85,6 +85,7 @@ make * 其它一些重要任务与组件 * [关于定时器](docs/about-timer.md) * [关于计数器](docs/about-counter.md) + * [条件任务与观察者模式](docs/about-conditional.md) * [关于资源池](docs/about-resource-pool.md) * [模块任务](/docs/about-module.md) * [DAG图任务](/docs/tutorial-11-graph_task.md)