diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index ab029bb6..8360f878 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -76,7 +76,6 @@ set(INCLUDE_HEADERS src/manager/WFFuture.h src/manager/WFFacilities.h src/manager/WFFacilities.inl - src/manager/WFCondition.h src/util/EncodeStream.h src/util/LRUCache.h src/util/StringUtil.h @@ -91,8 +90,10 @@ set(INCLUDE_HEADERS src/factory/WFTaskFactory.inl src/factory/WFAlgoTaskFactory.h src/factory/WFAlgoTaskFactory.inl + src/factory/WFSemTaskFactory.h src/factory/Workflow.h src/factory/WFOperator.h + src/factory/WFSemaphore.h src/factory/WFChannel.h src/factory/WFChannel.inl src/nameservice/WFNameService.h diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index ec43f9a7..07213e34 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -9,6 +9,8 @@ set(SRC MySQLTaskImpl.cc WebSocketTaskImpl.cc WFTaskFactory.cc + WFSemaphore.cc + WFSemTaskFactory.cc Workflow.cc ) diff --git a/src/factory/WFChannel.inl b/src/factory/WFChannel.inl index 54da3960..7664a4cb 100644 --- a/src/factory/WFChannel.inl +++ b/src/factory/WFChannel.inl @@ -21,7 +21,9 @@ #include "TransRequest.h" #include "WFTask.h" #include "WFTaskFactory.h" -#include "WFCondition.h" +//#include "WFCondition.h" +#include "WFSemaphore.h" +#include "WFSemTaskFactory.h" #include "WFNameService.h" #include "RouteManager.h" #include "WFGlobal.h" @@ -247,7 +249,7 @@ void WFComplexChannel::handle_terminated() this->sending = true; shutdown = true; } else { - WFCounterTask *counter = this->condition.create_wait_task(nullptr); + SubTask *counter = WFSemTaskFactory::create_wait_task(&this->condition, nullptr); series_of(this)->push_front(this); series_of(this)->push_front(counter); } @@ -349,8 +351,8 @@ void ComplexChannelOutTask::dispatch() } else { - WFCounterTask *counter = channel->condition.create_wait_task( - [this](WFCounterTask *task) + SubTask *counter = WFSemTaskFactory::create_wait_task(&channel->condition, + [this](WFMailboxTask *task) { auto *channel = (WFComplexChannel *)this->get_request_channel(); channel->set_state(WFT_STATE_SUCCESS); @@ -370,8 +372,8 @@ void ComplexChannelOutTask::dispatch() } else { - WFCounterTask *counter = channel->condition.create_wait_task( - [this](WFCounterTask *task) + SubTask *counter = WFSemTaskFactory::create_wait_task(&channel->condition, + [this](WFMailboxTask *task) { auto *channel = (WFComplexChannel *)this->get_request_channel(); channel->set_state(WFT_STATE_SUCCESS); @@ -416,7 +418,7 @@ SubTask *ComplexChannelOutTask::done() pthread_mutex_lock(&channel->mutex); channel->set_sending(false); - channel->condition.signal(); + channel->condition.signal(NULL); pthread_mutex_unlock(&channel->mutex); return WFChannelOutTask::done(); @@ -433,7 +435,7 @@ SubTask *ComplexChannelOutTask::upgrade() channel->set_state(WFT_STATE_SUCCESS); this->ready = true; channel->set_sending(false); - channel->condition.signal(); + channel->condition.signal(NULL); pthread_mutex_unlock(&channel->mutex); }); diff --git a/src/factory/WFSemTaskFactory.cc b/src/factory/WFSemTaskFactory.cc new file mode 100644 index 00000000..d054ea5c --- /dev/null +++ b/src/factory/WFSemTaskFactory.cc @@ -0,0 +1,287 @@ +/* + Copyright (c) 2021 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: Li Yingxin (liyingxin@sogou-inc.com) +*/ + +#include +#include +#include +#include "list.h" +#include "WFTask.h" +#include "WFTaskFactory.h" +#include "WFGlobal.h" +#include "WFSemTaskFactory.h" + +class __WFCondition : public WFCondition +{ +public: + __WFCondition(const std::string& str) : + name(str) + { } + + struct rb_node rb; + std::string name; +}; + +class __ConditionMap +{ +public: + void signal(const std::string& name, void *msg); + void broadcast(const std::string& name, void *msg); + + WFMailboxTask *create(const std::string& name, + mailbox_callback_t&& cb); + WFMailboxTask *create(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t&& cb); + + WFMailboxTask *create_switch(const std::string& name, + mailbox_callback_t&& cb); + WFMailboxTask *create_switch(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t&& cb); + +public: + static __ConditionMap *get_instance() + { + static __ConditionMap kInstance; + return &kInstance; + } + + virtual ~__ConditionMap(); + +private: + __ConditionMap() + { + this->condition_map.rb_node = NULL; + } + + __WFCondition *find_condition(const std::string& name); + struct rb_root condition_map; + std::mutex mutex; +}; + +void __ConditionMap::signal(const std::string& name, void *msg) +{ + __WFCondition *cond = this->find_condition(name); + + cond->signal(msg); +} + +void __ConditionMap::broadcast(const std::string& name, void *msg) +{ + __WFCondition *cond = this->find_condition(name); + + cond->broadcast(msg); +} + +WFMailboxTask *__ConditionMap::create(const std::string& name, + mailbox_callback_t&& cb) +{ + __WFCondition *cond = this->find_condition(name); + + return WFSemTaskFactory::create_wait_task(cond, std::move(cb)); +} + +WFMailboxTask *__ConditionMap::create(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t&& cb) +{ + __WFCondition *cond = this->find_condition(name); + + return WFSemTaskFactory::create_timedwait_task(cond, abstime, + std::move(cb)); +} + +WFMailboxTask *__ConditionMap::create_switch(const std::string& name, + mailbox_callback_t&& cb) +{ + __WFCondition *cond = this->find_condition(name); + + return WFSemTaskFactory::create_switch_wait_task(cond, std::move(cb)); +} + +WFMailboxTask *__ConditionMap::create_switch(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t&& cb) +{ + __WFCondition *cond = this->find_condition(name); + + return WFSemTaskFactory::create_switch_timedwait_task(cond, abstime, + std::move(cb)); +} + +__ConditionMap::~__ConditionMap() +{ + __WFCondition *cond; + WFWaitTask *task; + struct list_head *pos; + struct list_head *tmp; + + while (this->condition_map.rb_node) + { + cond = rb_entry(this->condition_map.rb_node, __WFCondition, rb); + list_for_each_safe(pos, tmp, &cond->waiter_list) + { + task = list_entry(pos, WFWaitTask, list); + list_del(pos); + delete task; + } + + rb_erase(this->condition_map.rb_node, &this->condition_map); + delete cond; + } +} + +__WFCondition *__ConditionMap::find_condition(const std::string& name) +{ + struct rb_node **p = &this->condition_map.rb_node; + struct rb_node *parent = NULL; + __WFCondition *cond; + + this->mutex.lock(); + while (*p) + { + parent = *p; + cond = rb_entry(*p, __WFCondition, rb); + + if (name < cond->name) + p = &(*p)->rb_left; + else if (name > cond->name) + p = &(*p)->rb_right; + else + break; + } + + if (*p == NULL) + { + cond = new __WFCondition(name); + rb_link_node(&cond->rb, parent, p); + rb_insert_color(&cond->rb, &this->condition_map); + } + + this->mutex.unlock(); + + return cond; +} + +/////////////// factory api /////////////// +/* +SubTask *acquire_by_name(const std::string& name, mailbox_callback_t callback) +{ + return __SemaphoreMap::get_instance()->acquire(name, std::move(callback)); +} + +void release_by_name(const std::string& name, void *msg) +{ + return __SemaphoreMap::get_instance()->release(name, msg); +} +*/ +void WFSemTaskFactory::signal_by_name(const std::string& name, void *msg) +{ + return __ConditionMap::get_instance()->signal(name, msg); +} + +void WFSemTaskFactory::broadcast_by_name(const std::string& name, void *msg) +{ + return __ConditionMap::get_instance()->broadcast(name, msg); +} + +WFMailboxTask *WFSemTaskFactory::create_wait_task(const std::string& name, + mailbox_callback_t callback) +{ + return __ConditionMap::get_instance()->create(name, std::move(callback)); +} + +WFMailboxTask *WFSemTaskFactory::create_timedwait_task(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t callback) +{ + return __ConditionMap::get_instance()->create(name, abstime, + std::move(callback)); +} + +WFMailboxTask *WFSemTaskFactory::create_switch_wait_task(const std::string& name, + mailbox_callback_t callback) +{ + return __ConditionMap::get_instance()->create_switch(name, std::move(callback)); +} + +WFMailboxTask *WFSemTaskFactory::create_switch_timedwait_task(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t callback) +{ + return __ConditionMap::get_instance()->create_switch(name, abstime, + std::move(callback)); +} + +WFMailboxTask *WFSemTaskFactory::create_wait_task(WFCondition *cond, + mailbox_callback_t callback) +{ + WFWaitTask *task = new WFWaitTask(std::move(callback)); + + cond->mutex.lock(); + list_add_tail(&task->list, &cond->waiter_list); + cond->mutex.unlock(); + + return task; +} + +WFMailboxTask *WFSemTaskFactory::create_timedwait_task(WFCondition *cond, + const struct timespec *abstime, + mailbox_callback_t callback) +{ + WFWaitTask *waiter = new WFWaitTask(std::move(callback)); + WFTimedWaitTask *task = new WFTimedWaitTask(waiter, &cond->mutex, abstime, + WFGlobal::get_scheduler(), + nullptr); + waiter->set_timer(task); + + cond->mutex.lock(); + list_add_tail(&waiter->list, &cond->waiter_list); + cond->mutex.unlock(); + + return waiter; +} + +WFMailboxTask *WFSemTaskFactory::create_switch_wait_task(WFCondition *cond, + mailbox_callback_t callback) +{ + WFWaitTask *task = new WFSwitchWaitTask(std::move(callback)); + + cond->mutex.lock(); + list_add_tail(&task->list, &cond->waiter_list); + cond->mutex.unlock(); + + return task; +} + +WFMailboxTask *WFSemTaskFactory::create_switch_timedwait_task(WFCondition *cond, + const struct timespec *abstime, + mailbox_callback_t callback) +{ + WFSwitchWaitTask *waiter = new WFSwitchWaitTask(std::move(callback)); + WFTimedWaitTask *task = new WFTimedWaitTask(waiter, &cond->mutex, abstime, + WFGlobal::get_scheduler(), + nullptr); + waiter->set_timer(task); + + cond->mutex.lock(); + list_add_tail(&waiter->list, &cond->waiter_list); + cond->mutex.unlock(); + + return waiter; +} diff --git a/src/factory/WFSemTaskFactory.h b/src/factory/WFSemTaskFactory.h new file mode 100644 index 00000000..3a91f787 --- /dev/null +++ b/src/factory/WFSemTaskFactory.h @@ -0,0 +1,76 @@ +/* + Copyright (c) 2021 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: Li Yingxin (liyingxin@sogou-inc.com) +*/ + +#ifndef _WFSEMTASKFACTORY_H_ +#define _WFSEMTASKFACTORY_H_ + +#include +#include +#include +#include +#include "list.h" +#include "WFTask.h" +#include "WFTaskFactory.h" +#include "WFGlobal.h" +#include "WFSemaphore.h" + +class WFSemTaskFactory +{ +public: + // use semaphore by name + static SubTask *acquire_by_name(const std::string& name, mailbox_callback_t cb); + + static void release_by_name(const std::string& name, void *msg); + + // use condition by name + static void signal_by_name(const std::string& name, void *msg); + + static void broadcast_by_name(const std::string& name, void *msg); + + static WFMailboxTask *create_wait_task(const std::string& name, + mailbox_callback_t callback); + + static WFMailboxTask *create_timedwait_task(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t callback); + + static WFMailboxTask *create_switch_wait_task(const std::string& name, + mailbox_callback_t callback); + + static WFMailboxTask *create_switch_timedwait_task(const std::string& name, + const struct timespec *abstime, + mailbox_callback_t callback); + + // use condition by ptr + static WFMailboxTask *create_wait_task(WFCondition *cond, + mailbox_callback_t callback); + + static WFMailboxTask *create_timedwait_task(WFCondition *cond, + const struct timespec *abstime, + mailbox_callback_t callback); + + static WFMailboxTask *create_switch_wait_task(WFCondition *cond, + mailbox_callback_t callback); + + static WFMailboxTask *create_switch_timedwait_task(WFCondition *cond, + const struct timespec *abstime, + mailbox_callback_t callback); +}; + +#endif + diff --git a/src/factory/WFSemaphore.cc b/src/factory/WFSemaphore.cc new file mode 100644 index 00000000..e944af90 --- /dev/null +++ b/src/factory/WFSemaphore.cc @@ -0,0 +1,136 @@ +/* + Copyright (c) 2021 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: Li Yingxin (liyingxin@sogou-inc.com) + Liu Kai (liukaidx@sogou-inc.com) +*/ + +#include +#include +#include +#include "list.h" +#include "WFTask.h" +#include "WFTaskFactory.h" +#include "WFSemaphore.h" + +/////////////// Semaphore Impl /////////////// + +WFMailboxTask *WFSemaphore::acquire(std::function cb) +{ + WFSemaphoreTask *task = new WFSemaphoreTask(std::move(cb)); + + if (--this->concurrency >= 0) + { + task->count(); + } + else + { + this->mutex.lock(); + list_add_tail(&task->list, &this->waiter_list); + this->mutex.unlock(); + } + + return task; +} + +void WFSemaphore::release(void *msg) +{ + WFSemaphoreTask *task; + struct list_head *pos; + + this->mutex.lock(); + + if (++this->concurrency <= 0)// && !list_empty(&this->waiter_list)) + { + pos = this->waiter_list.next; + task = list_entry(pos, WFSemaphoreTask, list); + list_del(pos); + task->send(msg); + } + + this->mutex.unlock(); +} + +/////////////// Wait tasks Impl /////////////// + +void WFWaitTask::dispatch() +{ + if (this->timer) + timer->dispatch(); + + this->WFMailboxTask::count(); +} + +void WFWaitTask::clear_timer_waiter() +{ + if (this->timer) + timer->clear_wait_task(); +} + +SubTask *WFTimedWaitTask::done() +{ + this->mutex->lock(); + if (this->wait_task && this->wait_task->list.next) + { + list_del(&this->wait_task->list); + this->wait_task->set_error(ETIMEDOUT); + this->wait_task->count(); + this->wait_task = NULL; + } + this->mutex->unlock(); + + delete this; + return NULL; +} + +/////////////// Condition Impl /////////////// + +void WFCondition::signal(void *msg) +{ + WFWaitTask *task; + struct list_head *pos; + + this->mutex.lock(); + + if (!list_empty(&this->waiter_list)) + { + pos = this->waiter_list.next; + task = list_entry(pos, WFWaitTask, list); + list_del(pos); + task->clear_timer_waiter(); + task->send(msg); + } + + this->mutex.unlock(); +} + +void WFCondition::broadcast(void *msg) +{ + WFWaitTask *task; + struct list_head *pos, *tmp; + + this->mutex.lock(); + if (!list_empty(&this->waiter_list)) + { + list_for_each_safe(pos, tmp, &this->waiter_list) + { + task = list_entry(pos, WFWaitTask, list); + list_del(pos); + task->send(msg); + } + } + this->mutex.unlock(); +} + diff --git a/src/factory/WFSemaphore.h b/src/factory/WFSemaphore.h new file mode 100644 index 00000000..eabd4888 --- /dev/null +++ b/src/factory/WFSemaphore.h @@ -0,0 +1,164 @@ +/* + Copyright (c) 2021 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: Li Yingxin (liyingxin@sogou-inc.com) + Liu Kai (liukaidx@sogou-inc.com) +*/ + +#ifndef _WFSEMAPHORE_H_ +#define _WFSEMAPHORE_H_ + +#include +#include +#include +#include +#include "list.h" +#include "WFTask.h" +#include "WFTaskFactory.h" +#include "WFGlobal.h" + +class WFSemaphore +{ +public: + WFMailboxTask *acquire(std::function cb); + void release(void *msg); + +public: + std::mutex mutex; + struct list_head waiter_list; + +public: + WFSemaphore(int value) + { + INIT_LIST_HEAD(&this->waiter_list); + this->concurrency = value; +// this->size = value; + } + +private: + std::atomic concurrency; +// int size; +}; + +class WFCondition : public WFSemaphore +{ +public: + void signal(void *msg); + void broadcast(void *msg); + +public: + WFCondition() : WFSemaphore(1) { } + WFCondition(int value) : WFSemaphore(value) { } + ~WFCondition() { } +}; + +class WFSemaphoreTask : public WFMailboxTask +{ +public: + WFSemaphoreTask(std::function&& cb) : + WFMailboxTask(std::move(cb)) +// WFMailboxTask(&this->msg, 1, std::move(cb)) + { + this->list.next = NULL; + } + + virtual ~WFSemaphoreTask() { } + +public: + struct list_head list; + +private: + void *msg; // TODO: user_data instead +}; + +class WFTimedWaitTask; + +class WFWaitTask : public WFSemaphoreTask +{ +public: + void set_timer(WFTimedWaitTask *timer) { this->timer = timer; } + void clear_timer_waiter(); + void set_error(int error) { this->error = error; } + +protected: + void dispatch(); + +private: + WFTimedWaitTask *timer; + +public: + WFWaitTask(std::function&& cb) : + WFSemaphoreTask(std::move(cb)) + { + this->timer = NULL; + } + + virtual ~WFWaitTask() { } +}; + +class WFTimedWaitTask : public __WFTimerTask +{ +public: + WFTimedWaitTask(WFWaitTask *wait_task, std::mutex *mutex, + const struct timespec *value, + CommScheduler *scheduler, + std::function cb) : + __WFTimerTask(value, scheduler, std::move(cb)) + { + this->mutex = mutex; + this->wait_task = wait_task; + } + + void clear_wait_task() + { + this->mutex->lock(); + this->wait_task = NULL; + this->mutex->unlock(); + } + +protected: + virtual SubTask *done(); + +private: + std::mutex *mutex; + WFWaitTask *wait_task; +}; + +class WFSwitchWaitTask : public WFWaitTask +{ +public: + WFSwitchWaitTask(std::function&& cb) : + WFWaitTask(std::move(cb)) + { } + +protected: + SubTask *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(); + } +}; + +#endif + diff --git a/src/factory/WebSocketTaskImpl.cc b/src/factory/WebSocketTaskImpl.cc index 8b288595..4b278ddb 100644 --- a/src/factory/WebSocketTaskImpl.cc +++ b/src/factory/WebSocketTaskImpl.cc @@ -130,7 +130,7 @@ SubTask *ComplexWebSocketOutTask::done() pthread_mutex_lock(&channel->mutex); channel->set_sending(false); - channel->condition.signal(); + channel->condition.signal(NULL); pthread_mutex_unlock(&channel->mutex); return WFChannelOutTask::done(); @@ -201,7 +201,7 @@ void ComplexWebSocketChannel::handle_in(CommMessageIn *in) if (!parse_websocket) // so this is equal to should_count { pthread_mutex_lock(&this->mutex); - this->condition.signal(); + this->condition.signal(NULL); pthread_mutex_unlock(&this->mutex); return; } diff --git a/src/manager/CMakeLists.txt b/src/manager/CMakeLists.txt index 5571ade8..91af981c 100644 --- a/src/manager/CMakeLists.txt +++ b/src/manager/CMakeLists.txt @@ -5,7 +5,6 @@ set(SRC DnsCache.cc UpstreamManager.cc RouteManager.cc - WFCondition.cc WFGlobal.cc ) diff --git a/src/manager/WFCondition.cc b/src/manager/WFCondition.cc index 52ea7efa..12c5d621 100644 --- a/src/manager/WFCondition.cc +++ b/src/manager/WFCondition.cc @@ -24,7 +24,7 @@ class WFTimedWaitTask; -class WFWaitTask : public WFCounterTask +class WFWaitTask : public WFMailboxTask { public: void set_timer(WFTimedWaitTask *timer) { this->timer = timer; } @@ -44,8 +44,8 @@ private: WFTimedWaitTask *timer; public: - WFWaitTask(std::function&& cb) : - WFCounterTask(1, std::move(cb)) + WFWaitTask(std::function&& cb) : + WFMailboxTask(&this->user_data, 1, std::move(cb)) { this->timer = NULL; this->entry.list.next = NULL; @@ -81,7 +81,7 @@ void WFWaitTask::dispatch() if (this->timer) timer->dispatch(); - this->WFCounterTask::count(); + this->WFMailboxTask::count(); } SubTask *WFWaitTask::done() @@ -139,7 +139,7 @@ void WFTimedWaitTask::clear_wait_task() this->mutex->unlock(); } -WFCounterTask *WFCondition::create_wait_task(counter_callback_t cb) +WFMailboxTask *WFCondition::create_wait_task(mailbox_callback_t cb) { WFWaitTask *task = new WFWaitTask(std::move(cb)); @@ -150,8 +150,8 @@ WFCounterTask *WFCondition::create_wait_task(counter_callback_t cb) return task; } -WFCounterTask *WFCondition::create_timedwait_task(const struct timespec *abstime, - counter_callback_t cb) +WFMailboxTask *WFCondition::create_timedwait_task(const struct timespec *abstime, + mailbox_callback_t cb) { WFWaitTask *waiter = new WFWaitTask(std::move(cb)); WFTimedWaitTask *task = new WFTimedWaitTask(waiter, &this->mutex, abstime, diff --git a/src/manager/WFCondition.h b/src/manager/WFCondition.h index 9f996dc0..9c343f31 100644 --- a/src/manager/WFCondition.h +++ b/src/manager/WFCondition.h @@ -28,9 +28,9 @@ class WFCondition { public: - WFCounterTask *create_wait_task(std::function cb); - WFCounterTask *create_timedwait_task(const struct timespec *abstime, - std::function cb); + WFMailboxTask *create_wait_task(std::function cb); + WFMailboxTask *create_timedwait_task(const struct timespec *abstime, + std::function cb); void signal(); void broadcast(); diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index c673b168..1eb5fa8b 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -39,6 +39,7 @@ set(TEST_LIST memory_unittest upstream_unittest dns_unittest + condition_unittest ) if (APPLE) diff --git a/test/condition_unittest.cc b/test/condition_unittest.cc new file mode 100644 index 00000000..16a96c82 --- /dev/null +++ b/test/condition_unittest.cc @@ -0,0 +1,108 @@ +/* + Copyright (c) 2021 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: Li Yingxin (liyingxin@sogou-inc.com) +*/ + +#include +#include +#include +#include +#include +#include +#include "workflow/WFTask.h" +#include "workflow/WFTaskFactory.h" +//#include "workflow/WFCondition.h" +#include "workflow/WFSemaphore.h" +#include "workflow/WFSemTaskFactory.h" +#include "workflow/WFFacilities.h" + +TEST(condition_unittest, signal) +{ + WFCondition cond; + std::mutex mutex; + int ret = 1979; + int *ptr = &ret; + + auto *task1 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) { + *ptr = 1412; + fprintf(stderr, "task1 callback\n"); + }); + SeriesWork *series1 = Workflow::create_series_work(task1, nullptr); + + auto *task2 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) { + *ptr = 2; + fprintf(stderr, "task2 callback\n"); + }); + SeriesWork *series2 = Workflow::create_series_work(task2, nullptr); + + series1->start(); + series2->start(); + + mutex.lock(); + cond.signal(NULL); + mutex.unlock(); + usleep(10000); + EXPECT_EQ(ret, 1412); + cond.signal(NULL); + usleep(10000); + EXPECT_EQ(ret, 2); +} + +TEST(condition_unittest, broadcast) +{ + WFCondition cond; + std::mutex mutex; + int ret = 0; + int *ptr = &ret; + + auto *task1 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) { + (*ptr)++; + fprintf(stderr, "task1 callback\n"); + }); + SeriesWork *series1 = Workflow::create_series_work(task1, nullptr); + + auto *task2 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) { + (*ptr)++; + fprintf(stderr, "task2 callback\n"); + }); + SeriesWork *series2 = Workflow::create_series_work(task2, nullptr); + + series1->start(); + series2->start(); + + cond.broadcast(NULL); + usleep(10000); + EXPECT_EQ(ret, 2); +} + +TEST(condition_unittest, timedwait) +{ + WFFacilities::WaitGroup wait_group(1); + struct timespec ts; + ts.tv_sec = 2; + ts.tv_nsec = 0; + + auto *task1 = WFSemTaskFactory::create_timedwait_task("timedwait", &ts, + [&wait_group](WFMailboxTask *task) { + EXPECT_EQ(task->get_error(), ETIMEDOUT); + wait_group.done(); + }); + + Workflow::start_series_work(task1, nullptr); + wait_group.wait(); + usleep(10000); +} +