seperate WFSemaphore; override dispatch() and signal();

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)

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

src/factory/CMakeLists.txt

@@ -7,6 +7,7 @@ set(SRC
 	Workflow.cc
 	WFTaskFactory.cc
 	WFCondTaskFactory.cc
+	WFSemaphore.cc
 	HttpTaskImpl.cc
 	WebSocketTaskImpl.cc
 )

src/factory/WFChannel.inl

@@ -429,7 +429,7 @@ SubTask *ComplexChannelOutTask<MSG>::done()
 template<class MSG>
 SubTask *ComplexChannelOutTask<MSG>::upgrade()
 {
-	WFCounterTask *counter =  new WFCounterTask(0, [this](WFCounterTask *task)
+	WFCounterTask *counter = new WFCounterTask(0, [this](WFCounterTask *task)
 	{
 		auto *channel = (WFComplexChannel<MSG> *)this->get_request_channel();
 

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

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

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<int> value;
-	std::atomic<int> index;
-
-protected:
-	void **sems;
-};
-
 class WFCondTaskFactory
 {
 public:

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

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

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))