diff --git a/CMakeLists.txt b/CMakeLists.txt index 217501db..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") + 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 a7680d56..5f7b44d0 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -88,6 +88,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 6f9a26b1..4eb5d906 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -7,6 +7,7 @@ set(SRC WFTaskFactory.cc Workflow.cc HttpTaskImpl.cc + WFResourcePool.cc ) if (NOT MYSQL STREQUAL "n") diff --git a/src/factory/WFResourcePool.cc b/src/factory/WFResourcePool.cc new file mode 100644 index 00000000..38deaf4a --- /dev/null +++ b/src/factory/WFResourcePool.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 "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 **pres, + struct WFResourcePool::Data *data) : + WFConditional(task, pres) + { + 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 **pres) +{ + return new __WFConditional(task, pres, &this->data); +} + +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; + data->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..c0a1a55e --- /dev/null +++ b/src/factory/WFResourcePool.h @@ -0,0 +1,84 @@ +/* + 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 _WFRESOURCEPOOL_H_ +#define _WFRESOURCEPOOL_H_ + +#include +#include +#include "list.h" +#include "WFTask.h" + +class WFResourcePool +{ +public: + WFConditional *get(SubTask *task, void **pres); + void post(void *res); + +public: + struct Data + { + void **res; + std::atomic value; + std::atomic index; + struct list_head wait_list; + std::mutex mutex; + WFResourcePool *ptr; + + virtual void *pop() + { + return this->ptr->pop(); + } + + virtual void push(void *res) + { + this->ptr->push(res); + } + }; + +private: + virtual void *pop() + { + return this->data.res[this->data.index++]; + } + + virtual void push(void *res) + { + this->data.res[--this->data.index] = res; + } + +private: + struct Data data; + +public: + WFResourcePool(const 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; } +}; + +#endif + diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index c673b168..d1442e29 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -39,6 +39,7 @@ set(TEST_LIST memory_unittest upstream_unittest dns_unittest + resource_unittest ) if (APPLE) diff --git a/test/resource_unittest.cc b/test/resource_unittest.cc new file mode 100644 index 00000000..c5fd1c07 --- /dev/null +++ b/test/resource_unittest.cc @@ -0,0 +1,55 @@ +/* + 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 "workflow/WFTask.h" +#include "workflow/WFTaskFactory.h" +#include "workflow/WFResourcePool.h" +#include "workflow/WFFacilities.h" + +TEST(resource_unittest, resource_pool) +{ + int res_concurrency = 3; + int task_concurrency = 10; + const char *words[3] = {"workflow", "srpc", "pyworkflow"}; + WFResourcePool res_pool((const void **)words, res_concurrency); + WFFacilities::WaitGroup wg(task_concurrency); + + for (int i = 0; i < task_concurrency; i++) + { + auto *user_task = WFTaskFactory::create_timer_task(0, + [&wg, &res_pool](WFTimerTask *task) { + uint64_t id = (uint64_t)series_of(task)->get_context(); + printf("task-%lu get [%s]\n", id, (char *)task->user_data); + res_pool.post(task->user_data); + wg.done(); + }); + + auto *cond = res_pool.get(user_task, &user_task->user_data); + + SeriesWork *series = Workflow::create_series_work(cond, nullptr); + series->set_context(reinterpret_cast(i)); + series->start(); + } + + wg.wait(); +} +