From bcd128c1ab14014952e56ecc3c5f9d8d287c95e5 Mon Sep 17 00:00:00 2001 From: XieHan Date: Sat, 18 Sep 2021 18:42:37 +0800 Subject: [PATCH] add WFResourcePool for windows --- CMakeLists.txt | 2 +- CMakeLists_Headers.txt | 1 + src/factory/CMakeLists.txt | 1 + src/factory/WFResourcePool.cc | 117 ++++++++++++++++++++++++++++++++++ src/factory/WFResourcePool.h | 72 +++++++++++++++++++++ 5 files changed, 192 insertions(+), 1 deletion(-) create mode 100644 src/factory/WFResourcePool.cc create mode 100644 src/factory/WFResourcePool.h diff --git a/CMakeLists.txt b/CMakeLists.txt index b8c34a67..2321428f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -68,7 +68,7 @@ if (WIN32) set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /MP /wd4200 /Zc:__cplusplus /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") + 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 3b73c6ac..8abc4cfc 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -96,6 +96,7 @@ set(INCLUDE_HEADERS src/factory/WFAlgoTaskFactory.inl src/factory/Workflow.h src/factory/WFOperator.h + src/factory/WFResourcePool.h src/nameservice/WFNameService.h src/nameservice/WFDNSResolver.h src/nameservice/WFServiceGovernance.h diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index aac274e2..dd3dd8ec 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -2,6 +2,7 @@ cmake_minimum_required(VERSION 3.6) project(factory) set(SRC + WFResourcePool.cc WFGraphTask.cc HttpTaskImpl.cc RedisTaskImpl.cc diff --git a/src/factory/WFResourcePool.cc b/src/factory/WFResourcePool.cc new file mode 100644 index 00000000..d01219b7 --- /dev/null +++ b/src/factory/WFResourcePool.cc @@ -0,0 +1,117 @@ +/* + 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. + + Authors: Li Yingxin (liyingxin@sogou-inc.com) + Xie Han (xiehan@sogou-inc.com) +*/ + +#include +#include "list.h" +#include "WFTask.h" +#include "WFResourcePool.h" + +class __WFConditional : public WFConditional +{ +public: + struct list_head list; + struct WFResourcePool::Data *data; + +public: + virtual void dispatch(); + virtual void signal(void *res) { } + +public: + __WFConditional(SubTask *task, void **resbuf, + struct WFResourcePool::Data *data) : + WFConditional(task, resbuf) + { + this->data = data; + } + + __WFConditional(SubTask *task, + struct WFResourcePool::Data *data) : + WFConditional(task) + { + this->data = data; + } +}; + +void __WFConditional::dispatch() +{ + struct WFResourcePool::Data *data = this->data; + + data->mutex.lock(); + if (--data->value >= 0) + this->WFConditional::signal(data->pop()); + else + list_add_tail(&this->list, &data->wait_list); + + data->mutex.unlock(); + this->WFConditional::dispatch(); +} + +WFConditional *WFResourcePool::get(SubTask *task, void **resbuf) +{ + return new __WFConditional(task, resbuf, &this->data); +} + +WFConditional *WFResourcePool::get(SubTask *task) +{ + return new __WFConditional(task, &this->data); +} + +void WFResourcePool::create(size_t n) +{ + this->data.res = new void *[n]; + this->data.value = n; + this->data.index = 0; + INIT_LIST_HEAD(&this->data.wait_list); + this->data.pool = this; +} + +WFResourcePool::WFResourcePool(void *const *res, size_t n) +{ + this->create(n); + memcpy(this->data.res, res, n * sizeof (void *)); +} + +WFResourcePool::WFResourcePool(size_t n) +{ + this->create(n); + memset(this->data.res, 0, n * sizeof (void *)); +} + +void WFResourcePool::post(void *res) +{ + struct WFResourcePool::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; + this->push(res); + } + + data->mutex.unlock(); + if (cond) + cond->WFConditional::signal(res); +} + diff --git a/src/factory/WFResourcePool.h b/src/factory/WFResourcePool.h new file mode 100644 index 00000000..1cdce2f6 --- /dev/null +++ b/src/factory/WFResourcePool.h @@ -0,0 +1,72 @@ +/* + 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. + + Authors: Li Yingxin (liyingxin@sogou-inc.com) + Xie Han (xiehan@sogou-inc.com) +*/ + +#ifndef _WFRESOURCEPOOL_H_ +#define _WFRESOURCEPOOL_H_ + +#include +#include "list.h" +#include "WFTask.h" + +class WFResourcePool +{ +public: + WFConditional *get(SubTask *task, void **resbuf); + WFConditional *get(SubTask *task); + void post(void *res); + +public: + struct Data + { + void *pop() { return this->pool->pop(); } + void push(void *res) { this->pool->push(res); } + + void **res; + long value; + size_t index; + struct list_head wait_list; + std::mutex mutex; + WFResourcePool *pool; + }; + +protected: + virtual void *pop() + { + return this->data.res[this->data.index++]; + } + + virtual void push(void *res) + { + this->data.res[--this->data.index] = res; + } + +protected: + struct Data data; + +private: + void create(size_t n); + +public: + WFResourcePool(void *const *res, size_t n); + WFResourcePool(size_t n); + virtual ~WFResourcePool() { delete []this->data.res; } +}; + +#endif +