diff --git a/CMakeLists.txt b/CMakeLists.txt index f6aa89a3..dde67fbe 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -68,7 +68,7 @@ if (WIN32) set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /MP /wd4200 /std:c++14") else () set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wall -fPIC -pipe -std=gnu90") - set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -fPIC -pipe -std=c++11 -fno-exceptions -Wno-invalid-offsetof") + set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -fPIC -pipe -std=c++11 -fno-exceptions -Wno-invalid-offsetof") endif () add_subdirectory(src) diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index 35a15778..bb1a3ec7 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -96,6 +96,7 @@ set(INCLUDE_HEADERS src/factory/WFAlgoTaskFactory.h src/factory/WFAlgoTaskFactory.inl src/factory/WFCondTaskFactory.h + src/factory/WFSemaphore.h src/factory/Workflow.h src/factory/WFOperator.h src/factory/WFChannel.h diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index 5a582735..28690e90 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -7,6 +7,7 @@ set(SRC Workflow.cc WFTaskFactory.cc WFCondTaskFactory.cc + WFSemaphore.cc HttpTaskImpl.cc WebSocketTaskImpl.cc ) diff --git a/src/factory/WFChannel.inl b/src/factory/WFChannel.inl index 6ffafac4..89082d51 100644 --- a/src/factory/WFChannel.inl +++ b/src/factory/WFChannel.inl @@ -429,7 +429,7 @@ SubTask *ComplexChannelOutTask::done() template SubTask *ComplexChannelOutTask::upgrade() { - WFCounterTask *counter = new WFCounterTask(0, [this](WFCounterTask *task) + WFCounterTask *counter = new WFCounterTask(0, [this](WFCounterTask *task) { auto *channel = (WFComplexChannel *)this->get_request_channel(); diff --git a/src/factory/WFCondTask.h b/src/factory/WFCondTask.h index 02db6c91..06fc0914 100644 --- a/src/factory/WFCondTask.h +++ b/src/factory/WFCondTask.h @@ -34,7 +34,7 @@ class WFCondWaitTask : public WFMailboxTask public: virtual void clear_locked() { } -public: +private: struct list_head list; private: @@ -48,24 +48,9 @@ public: virtual ~WFCondWaitTask() { } friend class __WFWaitTimerTask; -/* friend class WFCondition; friend class WFCondTaskFactory; - friend class __WFCondition; -*/ -}; - -class WFSemaphoreTask : public WFConditional -{ -private: - struct list_head list; - -public: - WFSemaphoreTask(SubTask *task, void **psem) : - WFConditional(task, psem) - { } - - friend class WFSemaphore; + friend class __ConditionMap; }; #endif diff --git a/src/factory/WFCondTaskFactory.cc b/src/factory/WFCondTaskFactory.cc index 6b27abcd..9e8cca2a 100644 --- a/src/factory/WFCondTaskFactory.cc +++ b/src/factory/WFCondTaskFactory.cc @@ -309,46 +309,6 @@ __WFWaitTimerTask::~__WFWaitTimerTask() } } -/////////////// WFSemaphore impl /////////////// - -WFConditional *WFSemaphore::get(SubTask *task, void **psem) -{ - WFConditional *cond; - WFSemaphoreTask *sem_task; - - if (--this->value >= 0) - { - cond = new WFConditional(task, psem); - cond->signal(this->sems[--this->index]); - return cond; - } - - sem_task = new WFSemaphoreTask(task, psem); - this->mutex.lock(); - list_add_tail(&sem_task->list, &this->wait_list); - this->mutex.unlock(); - - return sem_task; -} - -void WFSemaphore::post(void *sem) -{ - WFSemaphoreTask *task = NULL; - - if (++this->value <= 0) - { - this->mutex.lock(); - task = list_entry(this->wait_list.next, WFSemaphoreTask, list); - list_del(&task->list); - this->mutex.unlock(); - } - else - this->sems[this->index++] = sem; - - if (task) - task->signal(sem); -} - /////////////// factory impl /////////////// void WFCondTaskFactory::signal_by_name(const std::string& name, void *msg) @@ -399,10 +359,10 @@ WFWaitTask *WFCondTaskFactory::create_timedwait_task(WFCondition *cond, wait_callback_t callback) { WFTimedWaitTask *waiter = new WFTimedWaitTask(std::move(callback)); - __WFWaitTimerTask *task = new __WFWaitTimerTask(waiter, timeout, - cond->mutex, cond->ref, - WFGlobal::get_scheduler()); - waiter->set_timer(task); + __WFWaitTimerTask *timer = new __WFWaitTimerTask(waiter, timeout, + cond->mutex, cond->ref, + WFGlobal::get_scheduler()); + waiter->set_timer(timer); cond->mutex->lock(); list_add_tail(&waiter->list, &cond->wait_list); diff --git a/src/factory/WFCondTaskFactory.h b/src/factory/WFCondTaskFactory.h index 5f7199a1..09e2c363 100644 --- a/src/factory/WFCondTaskFactory.h +++ b/src/factory/WFCondTaskFactory.h @@ -29,41 +29,6 @@ #include "WFGlobal.h" #include "WFCondition.h" -class WFSemaphore -{ -public: - WFConditional *get(SubTask *task, void **psem); - void post(void *sem); - -private: - std::mutex mutex; - struct list_head wait_list; - - struct entry - { - struct list_head list; - WFConditional *ptr; - }; - -public: - WFSemaphore(void **sems, int n) - { - INIT_LIST_HEAD(&this->wait_list); - this->value = n; - this->index = n; - this->sems = sems; - } - - virtual ~WFSemaphore() { } - -private: - std::atomic value; - std::atomic index; - -protected: - void **sems; -}; - class WFCondTaskFactory { public: diff --git a/src/factory/WFSemaphore.cc b/src/factory/WFSemaphore.cc new file mode 100644 index 00000000..973d07ea --- /dev/null +++ b/src/factory/WFSemaphore.cc @@ -0,0 +1,83 @@ +/* + 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) + Xie Han (xiehan@sogou-inc.com) +*/ + +#include "list.h" +#include "WFTask.h" +#include "WFSemaphore.h" + +class __WFConditional : public WFConditional +{ +public: + struct list_head list; + struct WFSemaphore::Data *data; + +public: + virtual void dispatch(); + virtual void signal(void *msg) { } + +public: + __WFConditional(SubTask *task, void **pmsg, + struct WFSemaphore::Data *data) : + WFConditional(task, pmsg) + { + this->data = data; + } +}; + +void __WFConditional::dispatch() +{ + struct WFSemaphore::Data *data = this->data; + + data->mutex.lock(); + if (--data->value >= 0) + this->WFConditional::signal(data->sembuf[data->index++]); + else + list_add_tail(&this->list, &data->wait_list); + + data->mutex.unlock(); + this->WFConditional::dispatch(); +} + +WFConditional *WFSemaphore::get(SubTask *task, void **pmsg) +{ + return new __WFConditional(task, pmsg, &this->data); +} + +void WFSemaphore::post(void *msg) +{ + struct WFSemaphore::Data *data = &this->data; + WFConditional *cond; + + data->mutex.lock(); + if (++data->value <= 0) + { + cond = list_entry(data->wait_list.next, __WFConditional, list); + list_del(data->wait_list.next); + } + else + { + cond = NULL; + data->sembuf[--data->index] = msg; + } + + data->mutex.unlock(); + if (cond) + cond->WFConditional::signal(msg); +} + diff --git a/src/factory/WFSemaphore.h b/src/factory/WFSemaphore.h new file mode 100644 index 00000000..a9177b14 --- /dev/null +++ b/src/factory/WFSemaphore.h @@ -0,0 +1,59 @@ +/* + 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) + Xie Han (xiehan@sogou-inc.com) +*/ + +#ifndef _WFSEMAPHORE_H_ +#define _WFSEMAPHORE_H_ + +#include +#include +#include "list.h" +#include "WFTask.h" +#include "WFCondition.h" + +class WFSemaphore +{ +public: + WFConditional *get(SubTask *task, void **pmsg); + void post(void *msg); + +public: + struct Data + { + void **sembuf; + std::atomic value; + std::atomic index; + struct list_head wait_list; + std::mutex mutex; + }; + +private: + struct Data data; + +public: + WFSemaphore(void **sembuf, int nsems) + { + this->data.sembuf = sembuf; + this->data.value = nsems; + this->data.index = 0; + INIT_LIST_HEAD(&this->data.wait_list); + } +}; + +#endif + diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 5dc337df..123f11d5 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -634,7 +634,7 @@ protected: class WFConditional : public WFGenericTask { public: - void signal(void *msg) + virtual void signal(void *msg) { *this->msgbuf = msg; if (this->flag.exchange(true))