add WFResourcePool for windows

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)

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

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

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 <string.h>
+#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);
+}
+

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 <mutex>
+#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
+