From 8d2ab2821146d6b492a0d70ec0062e8506cfd925 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Mon, 19 Jul 2021 22:26:17 +0800 Subject: [PATCH] add BaseResource for WFCondition --- src/manager/WFCondition.cc | 62 ++++++++++++++++++++++++++- src/manager/WFCondition.h | 53 +++++++++++++++++++---- test/condition_unittest.cc | 87 +++++++++++++++++++++++++++++++++++++- 3 files changed, 190 insertions(+), 12 deletions(-) diff --git a/src/manager/WFCondition.cc b/src/manager/WFCondition.cc index 6fa3333b..44597c23 100644 --- a/src/manager/WFCondition.cc +++ b/src/manager/WFCondition.cc @@ -22,14 +22,64 @@ #include "WFCondTask.h" #include "WFCondition.h" +int WFCondition::get(void **pmsg) +{ +// if (this->res == NULL) +// return -2; + int ret; + + this->mutex->lock(); + if (this->empty == 1) + { + *pmsg = this->res->get(); + ret = 1; + } + else if (--this->empty == 0) + { + ret = 0; + } + else + { + WFCondWaitTask *task = new WFCondWaitTask(nullptr); + list_add_tail(&task->list, &this->get_list); + ret = -1; + } + this->mutex->unlock(); + + return ret; +} + +WFWaitTask *WFCondition::get_wait_task(wait_callback_t callback) +{ + WFCondWaitTask *task = NULL; + struct list_head *pos; + + this->mutex->lock(); + if (!list_empty(&this->get_list)) + { + pos = this->get_list.next; + list_move_tail(pos, &this->wait_list); + task = list_entry(pos, WFCondWaitTask, list); + task->set_callback(std::move(callback)); + } + + this->mutex->unlock(); + return task; +} + void WFCondition::signal(void *msg) { WFCondWaitTask *task = NULL; this->mutex->lock(); + if (!list_empty(&this->wait_list)) - { task = list_entry(this->wait_list.next, WFCondWaitTask, list); + else if (!list_empty(&this->get_list)) + task = list_entry(this->get_list.next, WFCondWaitTask, list); + + if (task) + { list_del(&task->list); task->clear_locked(); } @@ -56,6 +106,16 @@ void WFCondition::broadcast(void *msg) } } + if (!list_empty(&this->get_list)) + { + list_for_each_safe(pos, tmp, &this->get_list) + { + list_move_tail(pos, &tmp_list); + task = list_entry(pos, WFCondWaitTask, list); + task->clear_locked(); + } + } + this->mutex->unlock(); while (!list_empty(&tmp_list)) { diff --git a/src/manager/WFCondition.h b/src/manager/WFCondition.h index c9725570..037d042d 100644 --- a/src/manager/WFCondition.h +++ b/src/manager/WFCondition.h @@ -36,20 +36,55 @@ class WFCondition public: void signal(void *msg); void broadcast(void *msg); - -public: - WFCondition() - { - this->mutex = new std::mutex; - this->ref = new std::atomic(1); - INIT_LIST_HEAD(&this->wait_list); - } - virtual ~WFCondition(); + // 1: existed; 0: should get; -1: should wait; + int get(void **pmsg); + WFWaitTask *get_wait_task(wait_callback_t callback); public: std::atomic *ref; std::mutex *mutex; + struct list_head get_list; struct list_head wait_list; + +public: + class BaseResource + { + public: + virtual void *get() = 0; + virtual void set() { this->ptr->empty = 0; } + void clear() { this->ptr->empty = 1; } + + private: + WFCondition *ptr; + friend WFCondition; + }; + +private: + BaseResource *res; + int empty; + +public: + WFCondition(BaseResource *res) + { + this->empty = 1; + this->res = res; + this->res->ptr = this; + this->mutex = new std::mutex; + this->ref = new std::atomic(1); + INIT_LIST_HEAD(&this->get_list); + INIT_LIST_HEAD(&this->wait_list); + } + + WFCondition() + { + this->empty = 0; + this->res = NULL; + this->mutex = new std::mutex; + this->ref = new std::atomic(1); + INIT_LIST_HEAD(&this->get_list); + INIT_LIST_HEAD(&this->wait_list); + } + virtual ~WFCondition(); }; #endif diff --git a/test/condition_unittest.cc b/test/condition_unittest.cc index f71fbfe5..4dff2bfc 100644 --- a/test/condition_unittest.cc +++ b/test/condition_unittest.cc @@ -122,10 +122,10 @@ TEST(condition_unittest, timedwait) TEST(condition_unittest, resource_pool) { - int res__concurrency = 3; + 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 **)words, res_concurrency); WFFacilities::WaitGroup wg(task_concurrency); for (int i = 0; i < task_concurrency; i++) @@ -148,3 +148,86 @@ 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(); +} +