Merge branch 'master' of https://github.com/holmes1412/workflow-major

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)

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

src/factory/CMakeLists.txt

@@ -7,6 +7,7 @@ set(SRC
 	WFTaskFactory.cc
 	Workflow.cc
 	HttpTaskImpl.cc
+	WFResourcePool.cc
 )
 
 if (NOT MYSQL STREQUAL "n")

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);
+}
+

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 <mutex>
+#include <atomic>
+#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<int> value;
+		std::atomic<int> 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
+

test/CMakeLists.txt

@@ -39,6 +39,7 @@ set(TEST_LIST
 	memory_unittest
 	upstream_unittest
 	dns_unittest
+	resource_unittest
 )
 
 if (APPLE)

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 <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <gtest/gtest.h>
+#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<uint64_t *>(i));
+		series->start();
+	}
+
+	wg.wait();
+}
+