From f1cf95c379dfef74bf1bafbdeff267cd24bc4cc4 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Mon, 26 Jul 2021 18:44:31 +0800 Subject: [PATCH] remove WFSwitchWaitTask --- src/factory/WFChannel.inl | 8 +-- src/factory/WFCondTask.h | 3 ++ src/factory/WFCondTaskFactory.cc | 64 +++++------------------- src/factory/WFCondTaskFactory.h | 10 ++-- src/factory/WFResourcePool.cc | 21 ++++++-- src/factory/WFResourcePool.h | 35 ++++--------- src/manager/WFCondition.h | 1 + test/condition_unittest.cc | 85 +------------------------------- 8 files changed, 50 insertions(+), 177 deletions(-) diff --git a/src/factory/WFChannel.inl b/src/factory/WFChannel.inl index 89082d51..816bc43a 100644 --- a/src/factory/WFChannel.inl +++ b/src/factory/WFChannel.inl @@ -249,8 +249,8 @@ void WFComplexChannel::handle_terminated() this->sending = true; shutdown = true; } else { - waiter = WFCondTaskFactory::create_swait_task(&this->condition, - nullptr); + waiter = WFCondTaskFactory::create_wait_task(&this->condition, + nullptr); series_of(this)->push_front(this); series_of(this)->push_front(waiter); } @@ -353,7 +353,7 @@ void ComplexChannelOutTask::dispatch() } else { - waiter = WFCondTaskFactory::create_swait_task(&channel->condition, + waiter = WFCondTaskFactory::create_wait_task(&channel->condition, [this](WFMailboxTask *task) { auto *channel = (WFComplexChannel *)this->get_request_channel(); @@ -374,7 +374,7 @@ void ComplexChannelOutTask::dispatch() } else { - waiter = WFCondTaskFactory::create_swait_task(&channel->condition, + waiter = WFCondTaskFactory::create_wait_task(&channel->condition, [this](WFMailboxTask *task) { auto *channel = (WFComplexChannel *)this->get_request_channel(); diff --git a/src/factory/WFCondTask.h b/src/factory/WFCondTask.h index 06fc0914..ee3d3e72 100644 --- a/src/factory/WFCondTask.h +++ b/src/factory/WFCondTask.h @@ -34,6 +34,9 @@ class WFCondWaitTask : public WFMailboxTask public: virtual void clear_locked() { } +protected: + virtual SubTask *done(); + private: struct list_head list; diff --git a/src/factory/WFCondTaskFactory.cc b/src/factory/WFCondTaskFactory.cc index ac24c6d9..0fa983cf 100644 --- a/src/factory/WFCondTaskFactory.cc +++ b/src/factory/WFCondTaskFactory.cc @@ -50,8 +50,6 @@ public: WFWaitTask *create(const std::string& name, const struct timespec *timeout, wait_callback_t&& cb); - WFWaitTask *create_switch(const std::string& name, - wait_callback_t&& cb); public: static __ConditionMap *get_instance() @@ -105,14 +103,6 @@ WFWaitTask *__ConditionMap::create(const std::string& name, std::move(cb)); } -WFWaitTask *__ConditionMap::create_switch(const std::string& name, - wait_callback_t&& cb) -{ - __WFCondition *cond = this->find_condition(name); - - return WFCondTaskFactory::create_swait_task(cond, std::move(cb)); -} - __ConditionMap::~__ConditionMap() { __WFCondition *cond; @@ -235,16 +225,20 @@ public: virtual ~__WFWaitTimerTask(); }; -class WFSwitchWaitTask : public WFCondWaitTask +SubTask *WFCondWaitTask::done() { -public: - WFSwitchWaitTask(wait_callback_t&& cb) : - WFCondWaitTask(std::move(cb)) - { } + SeriesWork *series = series_of(this); -protected: - SubTask *done(); -}; + WFTimerTask *switch_task = WFTaskFactory::create_timer_task(0, + [this](WFTimerTask *task) { + if (this->callback) + this->callback(this); + delete this; + }); + series->push_front(switch_task); + + return series->pop(); +} void WFTimedWaitTask::clear_locked() { @@ -276,21 +270,6 @@ WFTimedWaitTask::~WFTimedWaitTask() delete this->timer; } -SubTask *WFSwitchWaitTask::done() -{ - SeriesWork *series = series_of(this); - - WFTimerTask *switch_task = WFTaskFactory::create_timer_task(0, - [this](WFTimerTask *task) { - if (this->callback) - this->callback(this); - delete this; - }); - series->push_front(switch_task); - - return series->pop(); -} - SubTask *__WFWaitTimerTask::done() { WFTimedWaitTask *waiter = NULL; @@ -339,13 +318,6 @@ WFWaitTask *WFCondTaskFactory::create_wait_task(const std::string& name, return __ConditionMap::get_instance()->create(name, std::move(callback)); } -WFWaitTask *WFCondTaskFactory::create_swait_task(const std::string& name, - wait_callback_t callback) -{ - return __ConditionMap::get_instance()->create_switch(name, - std::move(callback)); -} - WFWaitTask *WFCondTaskFactory::create_timedwait_task(const std::string& name, const struct timespec *timeout, wait_callback_t callback) @@ -383,15 +355,3 @@ WFWaitTask *WFCondTaskFactory::create_timedwait_task(WFCondition *cond, return waiter; } -WFWaitTask *WFCondTaskFactory::create_swait_task(WFCondition *cond, - wait_callback_t callback) -{ - WFSwitchWaitTask *task = new WFSwitchWaitTask(std::move(callback)); - - cond->mutex->lock(); - list_add_tail(&task->list, &cond->wait_list); - cond->mutex->unlock(); - - return task; -} - diff --git a/src/factory/WFCondTaskFactory.h b/src/factory/WFCondTaskFactory.h index 09e2c363..8a4b3764 100644 --- a/src/factory/WFCondTaskFactory.h +++ b/src/factory/WFCondTaskFactory.h @@ -32,6 +32,10 @@ class WFCondTaskFactory { public: + // simple timedwait task +// static WFWaitTask *create_timedwait_task(const struct timespec *timeout, +// wait_callback_t callback); + // use condition by name static void signal_by_name(const std::string& name, void *msg); @@ -44,9 +48,6 @@ public: const struct timespec *timeout, wait_callback_t callback); - static WFWaitTask *create_swait_task(const std::string& name, - wait_callback_t callback); - // use condition by ptr static WFWaitTask *create_wait_task(WFCondition *cond, wait_callback_t callback); @@ -55,9 +56,6 @@ public: const struct timespec *timeout, wait_callback_t callback); - static WFWaitTask *create_swait_task(WFCondition *cond, - wait_callback_t callback); - }; #endif diff --git a/src/factory/WFResourcePool.cc b/src/factory/WFResourcePool.cc index 38deaf4a..f52bb810 100644 --- a/src/factory/WFResourcePool.cc +++ b/src/factory/WFResourcePool.cc @@ -17,6 +17,7 @@ Xie Han (xiehan@sogou-inc.com) */ +#include #include "list.h" #include "WFTask.h" #include "WFResourcePool.h" @@ -32,9 +33,9 @@ public: virtual void signal(void *res) { } public: - __WFConditional(SubTask *task, void **pres, + __WFConditional(SubTask *task, void **resbuf, struct WFResourcePool::Data *data) : - WFConditional(task, pres) + WFConditional(task, resbuf) { this->data = data; } @@ -54,9 +55,19 @@ void __WFConditional::dispatch() this->WFConditional::dispatch(); } -WFConditional *WFResourcePool::get(SubTask *task, void **pres) +WFConditional *WFResourcePool::get(SubTask *task, void **resbuf) { - return new __WFConditional(task, pres, &this->data); + return new __WFConditional(task, resbuf, &this->data); +} + +WFResourcePool::WFResourcePool(void *const *res, size_t n) +{ + this->data.res = new void *[n]; + memcpy(this->data.res, res, n * sizeof (void *)); + this->data.value = n; + this->data.index = 0; + INIT_LIST_HEAD(&this->data.wait_list); + this->data.pool = this; } void WFResourcePool::post(void *res) @@ -73,7 +84,7 @@ void WFResourcePool::post(void *res) else { cond = NULL; - data->push(res); + this->push(res); } data->mutex.unlock(); diff --git a/src/factory/WFResourcePool.h b/src/factory/WFResourcePool.h index fa3e5996..38f44797 100644 --- a/src/factory/WFResourcePool.h +++ b/src/factory/WFResourcePool.h @@ -21,35 +21,27 @@ #define _WFRESOURCEPOOL_H_ #include -#include #include "list.h" #include "WFTask.h" class WFResourcePool { public: - WFConditional *get(SubTask *task, void **pres); + WFConditional *get(SubTask *task, void **resbuf); void post(void *res); public: struct Data { + void *pop() { return this->pool->pop(); } + void push(void *res) { this->pool->push(res); } + void **res; - std::atomic value; - std::atomic index; + long value; + size_t index; struct list_head wait_list; std::mutex mutex; - WFResourcePool *ptr; - - virtual void *pop() - { - return this->ptr->pop(); - } - - virtual void push(void *res) - { - ptr->push(res); - } + WFResourcePool *pool; }; private: @@ -67,17 +59,8 @@ private: struct Data data; public: - WFResourcePool(void **res, int n) - { - this->data.ptr = this; - this->data.res = new void *[n]; - memcpy(this->data.res, res, n * sizeof(void *)); - this->data.value = n; - this->data.index = 0; - INIT_LIST_HEAD(&this->data.wait_list); - } - - virtual ~WFResourcePool() { delete[] this->data.res; } + WFResourcePool(void *const *res, size_t n); + virtual ~WFResourcePool() { delete []this->data.res; } }; #endif diff --git a/src/manager/WFCondition.h b/src/manager/WFCondition.h index 037d042d..dfb47085 100644 --- a/src/manager/WFCondition.h +++ b/src/manager/WFCondition.h @@ -84,6 +84,7 @@ public: INIT_LIST_HEAD(&this->get_list); INIT_LIST_HEAD(&this->wait_list); } + virtual ~WFCondition(); }; diff --git a/test/condition_unittest.cc b/test/condition_unittest.cc index 4dff2bfc..69d26a3d 100644 --- a/test/condition_unittest.cc +++ b/test/condition_unittest.cc @@ -125,7 +125,7 @@ TEST(condition_unittest, resource_pool) int res_concurrency = 3; int task_concurrency = 10; const char *words[3] = {"workflow", "srpc", "pyworkflow"}; - WFResourcePool res_pool((void **)words, res_concurrency); + WFResourcePool res_pool((void *const *)words, res_concurrency); WFFacilities::WaitGroup wg(task_concurrency); for (int i = 0; i < task_concurrency; i++) @@ -148,86 +148,3 @@ TEST(condition_unittest, resource_pool) wg.wait(); } -class TestResource : public WFCondition::BaseResource -{ -public: - TestResource(std::string msg) : msg(msg) { } - void *get() { return &this->msg; } - void set(std::string&& msg) { this->msg = std::move(msg);} - -private: - std::string msg; -}; - -TestResource res(""); - -void wait_callback(WFWaitTask *task) -{ - void **msg; - size_t n; - msg = task->get_mailbox(&n); - std::string *str = *msg; - fprintf(stderr, "waiter get msg:%s\n", str->c_str()); - series->set_context(*msg); -} - -void work(std::string *msg, WFFacilities::WaitGroup &wg) -{ - fprintf(stderr, "working on message:%s\n", msg->c_str()); - wg.done(); -} - -SubTask *create_router_task(WFCondition *cond, std::function cb) -{ - auto *router = WFTaskFactory::create_timer_task(100000, std::move(cb)); - router->user_data = cond; -} - -void router_callback(WFTimerTask *task) -{ - fprintf(stderr, "finish routing and should broadcast every one else\n"); - - ((WFCondition *)(task->user_data))->broadcast(); -} - -TEST(condition_unittest, res_condition) -{ - int task_concurrency = 10; - WFFacilities::WaitGroup wg(task_concurrency); - WFCondition cond((WFCondition::BaseResource *)&res); - - for (int i = 0; i < task_concurrency; i++) - { - auto *timer = WFTaskFactory::create_timer_task(i * 100000, - [&cond](WFTimerTask *task) { - std::string *msg; - int ret = cond.get(&msg); - if (ret == -1) - { - auto *waiter = cond.create_wait_task(wait_callback); - series_of(task)->push_front(waiter); - fprintf(stderr, "task-%lu waiting\n", id); - } - else if (ret == 0) - { - auto *router = create_router_task(route_callback); - series_of(task)->push_front(router); - fprintf(stderr, "task-%lu routing\n", id); - } - else - { - fprintf(stderr, "task-%lu get [%s]\n", id, (*msg).c_str()); - } - }); - - SeriesWork *series = Workflow::create_series_work(timer, nullptr); - auto *worker = WFTaskFactory::create_go_task(work, - (std::string *)series->get_context(), - &wg); - series->push_back(worker); - series->start(); - } - - wg.wait(); -} -