add Semaphore and SemTaskFactory

CMakeLists_Headers.txt

@@ -76,7 +76,6 @@ set(INCLUDE_HEADERS
 	src/manager/WFFuture.h
 	src/manager/WFFacilities.h
 	src/manager/WFFacilities.inl
-	src/manager/WFCondition.h
 	src/util/EncodeStream.h
 	src/util/LRUCache.h
 	src/util/StringUtil.h
@@ -91,8 +90,10 @@ set(INCLUDE_HEADERS
 	src/factory/WFTaskFactory.inl
 	src/factory/WFAlgoTaskFactory.h
 	src/factory/WFAlgoTaskFactory.inl
+	src/factory/WFSemTaskFactory.h
 	src/factory/Workflow.h
 	src/factory/WFOperator.h
+	src/factory/WFSemaphore.h
 	src/factory/WFChannel.h
 	src/factory/WFChannel.inl
 	src/nameservice/WFNameService.h

src/factory/CMakeLists.txt

@@ -9,6 +9,8 @@ set(SRC
 	MySQLTaskImpl.cc
 	WebSocketTaskImpl.cc
 	WFTaskFactory.cc
+	WFSemaphore.cc
+	WFSemTaskFactory.cc
 	Workflow.cc
 )
 

src/factory/WFChannel.inl

@@ -21,7 +21,9 @@
 #include "TransRequest.h"
 #include "WFTask.h"
 #include "WFTaskFactory.h"
-#include "WFCondition.h"
+//#include "WFCondition.h"
+#include "WFSemaphore.h"
+#include "WFSemTaskFactory.h"
 #include "WFNameService.h"
 #include "RouteManager.h"
 #include "WFGlobal.h"
@@ -247,7 +249,7 @@ void WFComplexChannel<MSG>::handle_terminated()
 		this->sending = true;
 		shutdown = true;
 	} else {
-		WFCounterTask *counter = this->condition.create_wait_task(nullptr);
+		SubTask *counter = WFSemTaskFactory::create_wait_task(&this->condition, nullptr);
 		series_of(this)->push_front(this);
 		series_of(this)->push_front(counter);
 	}
@@ -349,8 +351,8 @@ void ComplexChannelOutTask<MSG>::dispatch()
 		}
 		else
 		{
-			WFCounterTask *counter = channel->condition.create_wait_task(
-				[this](WFCounterTask *task)
+			SubTask *counter = WFSemTaskFactory::create_wait_task(&channel->condition,
+				[this](WFMailboxTask *task)
 			{
 				auto *channel = (WFComplexChannel<MSG> *)this->get_request_channel();
 				channel->set_state(WFT_STATE_SUCCESS);
@@ -370,8 +372,8 @@ void ComplexChannelOutTask<MSG>::dispatch()
 		}
 		else
 		{
-			WFCounterTask *counter = channel->condition.create_wait_task(
-				[this](WFCounterTask *task)
+			SubTask *counter = WFSemTaskFactory::create_wait_task(&channel->condition,
+				[this](WFMailboxTask *task)
 			{
 				auto *channel = (WFComplexChannel<MSG> *)this->get_request_channel();
 				channel->set_state(WFT_STATE_SUCCESS);
@@ -416,7 +418,7 @@ SubTask *ComplexChannelOutTask<MSG>::done()
 
 	pthread_mutex_lock(&channel->mutex);
 	channel->set_sending(false);
-	channel->condition.signal();
+	channel->condition.signal(NULL);
 	pthread_mutex_unlock(&channel->mutex);
 
 	return WFChannelOutTask<MSG>::done();
@@ -433,7 +435,7 @@ SubTask *ComplexChannelOutTask<MSG>::upgrade()
 		channel->set_state(WFT_STATE_SUCCESS);
 		this->ready = true;
 		channel->set_sending(false);
-		channel->condition.signal();
+		channel->condition.signal(NULL);
 		pthread_mutex_unlock(&channel->mutex);
 	});
 

src/factory/WFSemTaskFactory.cc

@@ -0,0 +1,287 @@
+/*
+  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 <mutex>
+#include <time.h>
+#include <functional>
+#include "list.h"
+#include "WFTask.h"
+#include "WFTaskFactory.h"
+#include "WFGlobal.h"
+#include "WFSemTaskFactory.h"
+
+class __WFCondition : public WFCondition
+{
+public:
+	__WFCondition(const std::string& str) :
+		name(str)
+	{ }
+
+	struct rb_node rb;
+	std::string name;
+};
+
+class __ConditionMap
+{
+public:
+	void signal(const std::string& name, void *msg);
+	void broadcast(const std::string& name, void *msg);
+
+	WFMailboxTask *create(const std::string& name,
+						  mailbox_callback_t&& cb);
+	WFMailboxTask *create(const std::string& name,
+						  const struct timespec *abstime,
+						  mailbox_callback_t&& cb);
+
+	WFMailboxTask *create_switch(const std::string& name,
+								 mailbox_callback_t&& cb);
+	WFMailboxTask *create_switch(const std::string& name,
+								 const struct timespec *abstime,
+								 mailbox_callback_t&& cb);
+
+public:
+	static __ConditionMap *get_instance()
+	{
+		static __ConditionMap kInstance;
+		return &kInstance;
+	}
+
+	virtual ~__ConditionMap();
+
+private:
+	__ConditionMap()
+	{
+		this->condition_map.rb_node = NULL;
+	}
+
+	__WFCondition *find_condition(const std::string& name);
+	struct rb_root condition_map;
+	std::mutex mutex;
+};
+
+void __ConditionMap::signal(const std::string& name, void *msg)
+{
+	__WFCondition *cond = this->find_condition(name);
+
+	cond->signal(msg);
+}
+
+void __ConditionMap::broadcast(const std::string& name, void *msg)
+{
+	__WFCondition *cond = this->find_condition(name);
+
+	cond->broadcast(msg);
+}
+
+WFMailboxTask *__ConditionMap::create(const std::string& name,
+									  mailbox_callback_t&& cb)
+{
+	__WFCondition *cond = this->find_condition(name);
+
+	return WFSemTaskFactory::create_wait_task(cond, std::move(cb));
+}
+
+WFMailboxTask *__ConditionMap::create(const std::string& name,
+									  const struct timespec *abstime,
+									  mailbox_callback_t&& cb)
+{
+	__WFCondition *cond = this->find_condition(name);
+
+	return WFSemTaskFactory::create_timedwait_task(cond, abstime,
+												   std::move(cb));
+}
+
+WFMailboxTask *__ConditionMap::create_switch(const std::string& name,
+											 mailbox_callback_t&& cb)
+{
+	__WFCondition *cond = this->find_condition(name);
+
+	return WFSemTaskFactory::create_switch_wait_task(cond, std::move(cb));
+}
+
+WFMailboxTask *__ConditionMap::create_switch(const std::string& name,
+											 const struct timespec *abstime,
+											 mailbox_callback_t&& cb)
+{
+	__WFCondition *cond = this->find_condition(name);
+
+	return WFSemTaskFactory::create_switch_timedwait_task(cond, abstime,
+														  std::move(cb));
+}
+
+__ConditionMap::~__ConditionMap()
+{
+	__WFCondition *cond;
+	WFWaitTask *task;
+	struct list_head *pos;
+	struct list_head *tmp;
+
+	while (this->condition_map.rb_node)
+	{
+		cond = rb_entry(this->condition_map.rb_node, __WFCondition, rb);
+		list_for_each_safe(pos, tmp, &cond->waiter_list)
+		{
+			task = list_entry(pos, WFWaitTask, list);
+			list_del(pos);
+			delete task;
+		}
+
+		rb_erase(this->condition_map.rb_node, &this->condition_map);
+		delete cond;
+	}
+}
+
+__WFCondition *__ConditionMap::find_condition(const std::string& name)
+{
+	struct rb_node **p = &this->condition_map.rb_node;
+	struct rb_node *parent = NULL;
+	__WFCondition *cond;
+
+	this->mutex.lock();
+	while (*p)
+	{
+		parent = *p;
+		cond = rb_entry(*p, __WFCondition, rb);
+
+		if (name < cond->name)
+			p = &(*p)->rb_left;
+		else if (name > cond->name)
+			p = &(*p)->rb_right;
+		else
+			break;
+	}
+
+	if (*p == NULL)
+	{
+		cond = new __WFCondition(name);
+		rb_link_node(&cond->rb, parent, p);
+		rb_insert_color(&cond->rb, &this->condition_map);
+	}
+
+	this->mutex.unlock();
+
+	return cond;
+}
+
+/////////////// factory api ///////////////
+/*
+SubTask *acquire_by_name(const std::string& name, mailbox_callback_t callback)
+{
+	return __SemaphoreMap::get_instance()->acquire(name, std::move(callback));
+}
+
+void release_by_name(const std::string& name, void *msg)
+{
+	return __SemaphoreMap::get_instance()->release(name, msg);
+}
+*/
+void WFSemTaskFactory::signal_by_name(const std::string& name, void *msg)
+{
+	return __ConditionMap::get_instance()->signal(name, msg);
+}
+
+void WFSemTaskFactory::broadcast_by_name(const std::string& name, void *msg)
+{
+	return __ConditionMap::get_instance()->broadcast(name, msg);
+}
+
+WFMailboxTask *WFSemTaskFactory::create_wait_task(const std::string& name,
+												  mailbox_callback_t callback)
+{
+	return __ConditionMap::get_instance()->create(name, std::move(callback));
+}
+
+WFMailboxTask *WFSemTaskFactory::create_timedwait_task(const std::string& name,
+													   const struct timespec *abstime,
+													   mailbox_callback_t callback)
+{
+	return __ConditionMap::get_instance()->create(name, abstime,
+												  std::move(callback));
+}
+
+WFMailboxTask *WFSemTaskFactory::create_switch_wait_task(const std::string& name,
+														 mailbox_callback_t callback)
+{
+	return __ConditionMap::get_instance()->create_switch(name, std::move(callback));
+}
+
+WFMailboxTask *WFSemTaskFactory::create_switch_timedwait_task(const std::string& name,
+														 	  const struct timespec *abstime,
+															  mailbox_callback_t callback)
+{
+	return __ConditionMap::get_instance()->create_switch(name, abstime,
+														 std::move(callback));
+}
+
+WFMailboxTask *WFSemTaskFactory::create_wait_task(WFCondition *cond,
+												  mailbox_callback_t callback)
+{
+	WFWaitTask *task = new WFWaitTask(std::move(callback));
+
+	cond->mutex.lock();
+	list_add_tail(&task->list, &cond->waiter_list);
+	cond->mutex.unlock();
+
+	return task;
+}
+
+WFMailboxTask *WFSemTaskFactory::create_timedwait_task(WFCondition *cond,
+													   const struct timespec *abstime,
+													   mailbox_callback_t callback)
+{
+	WFWaitTask *waiter = new WFWaitTask(std::move(callback));
+	WFTimedWaitTask *task = new WFTimedWaitTask(waiter, &cond->mutex, abstime,
+												WFGlobal::get_scheduler(),
+												nullptr);
+	waiter->set_timer(task);
+
+	cond->mutex.lock();
+	list_add_tail(&waiter->list, &cond->waiter_list);
+	cond->mutex.unlock();
+
+	return waiter;
+}
+
+WFMailboxTask *WFSemTaskFactory::create_switch_wait_task(WFCondition *cond,
+														 mailbox_callback_t callback)
+{
+	WFWaitTask *task = new WFSwitchWaitTask(std::move(callback));
+
+	cond->mutex.lock();
+	list_add_tail(&task->list, &cond->waiter_list);
+	cond->mutex.unlock();
+
+	return task;
+}
+
+WFMailboxTask *WFSemTaskFactory::create_switch_timedwait_task(WFCondition *cond,
+															  const struct timespec *abstime,
+															  mailbox_callback_t callback)
+{
+	WFSwitchWaitTask *waiter = new WFSwitchWaitTask(std::move(callback));
+	WFTimedWaitTask *task = new WFTimedWaitTask(waiter, &cond->mutex, abstime,
+												WFGlobal::get_scheduler(),
+												nullptr);
+	waiter->set_timer(task);
+
+	cond->mutex.lock();
+	list_add_tail(&waiter->list, &cond->waiter_list);
+	cond->mutex.unlock();
+
+	return waiter;
+}

src/factory/WFSemTaskFactory.h

@@ -0,0 +1,76 @@
+/*
+  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)
+*/
+
+#ifndef _WFSEMTASKFACTORY_H_
+#define _WFSEMTASKFACTORY_H_
+
+#include <mutex>
+#include <time.h>
+#include <functional>
+#include <string>
+#include "list.h"
+#include "WFTask.h"
+#include "WFTaskFactory.h"
+#include "WFGlobal.h"
+#include "WFSemaphore.h"
+
+class WFSemTaskFactory
+{
+public:
+	// use semaphore by name
+	static SubTask *acquire_by_name(const std::string& name, mailbox_callback_t cb);
+
+	static void release_by_name(const std::string& name, void *msg);
+
+	// use condition by name
+	static void signal_by_name(const std::string& name, void *msg);
+
+	static void broadcast_by_name(const std::string& name, void *msg);
+
+	static WFMailboxTask *create_wait_task(const std::string& name,
+										   mailbox_callback_t callback);
+
+	static WFMailboxTask *create_timedwait_task(const std::string& name,
+												const struct timespec *abstime,
+												mailbox_callback_t callback);
+
+	static WFMailboxTask *create_switch_wait_task(const std::string& name,
+												  mailbox_callback_t callback);
+
+	static WFMailboxTask *create_switch_timedwait_task(const std::string& name,
+													   const struct timespec *abstime,
+													   mailbox_callback_t callback);
+
+	// use condition by ptr
+	static WFMailboxTask *create_wait_task(WFCondition *cond,
+										   mailbox_callback_t callback);
+
+	static WFMailboxTask *create_timedwait_task(WFCondition *cond,
+												const struct timespec *abstime,
+									  			mailbox_callback_t callback);
+
+	static WFMailboxTask *create_switch_wait_task(WFCondition *cond,
+												  mailbox_callback_t callback);
+
+	static WFMailboxTask *create_switch_timedwait_task(WFCondition *cond,
+													   const struct timespec *abstime,
+									 				   mailbox_callback_t callback);
+};
+
+#endif
+

src/factory/WFSemaphore.cc

@@ -0,0 +1,136 @@
+/*
+  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)
+          Liu Kai (liukaidx@sogou-inc.com)
+*/
+
+#include <mutex>
+#include <time.h>
+#include <functional>
+#include "list.h"
+#include "WFTask.h"
+#include "WFTaskFactory.h"
+#include "WFSemaphore.h"
+
+/////////////// Semaphore Impl ///////////////
+
+WFMailboxTask *WFSemaphore::acquire(std::function<void (WFMailboxTask *)> cb)
+{
+	WFSemaphoreTask *task = new WFSemaphoreTask(std::move(cb));
+
+	if (--this->concurrency >= 0)
+	{
+		task->count();
+	}
+	else
+	{
+		this->mutex.lock();
+		list_add_tail(&task->list, &this->waiter_list);
+		this->mutex.unlock();
+	}
+
+	return task;
+}
+
+void WFSemaphore::release(void *msg)
+{
+	WFSemaphoreTask *task;
+	struct list_head *pos;
+
+	this->mutex.lock();
+
+	if (++this->concurrency <= 0)// && !list_empty(&this->waiter_list))
+	{
+		pos = this->waiter_list.next;
+		task = list_entry(pos, WFSemaphoreTask, list);
+		list_del(pos);
+		task->send(msg);
+	}
+
+	this->mutex.unlock();
+}
+
+/////////////// Wait tasks Impl ///////////////
+
+void WFWaitTask::dispatch()
+{
+	if (this->timer)
+		timer->dispatch();
+	
+	this->WFMailboxTask::count();
+}
+
+void WFWaitTask::clear_timer_waiter()
+{
+	if (this->timer)
+		timer->clear_wait_task();
+}
+
+SubTask *WFTimedWaitTask::done()
+{
+	this->mutex->lock();
+	if (this->wait_task && this->wait_task->list.next)
+	{
+		list_del(&this->wait_task->list);
+		this->wait_task->set_error(ETIMEDOUT);
+		this->wait_task->count();
+		this->wait_task = NULL;
+	}
+	this->mutex->unlock();
+
+	delete this;
+	return NULL;
+}
+
+/////////////// Condition Impl ///////////////
+
+void WFCondition::signal(void *msg)
+{
+	WFWaitTask *task;
+	struct list_head *pos;
+
+	this->mutex.lock();
+
+	if (!list_empty(&this->waiter_list))
+	{
+		pos = this->waiter_list.next;
+		task = list_entry(pos, WFWaitTask, list);
+		list_del(pos);
+		task->clear_timer_waiter();
+		task->send(msg);
+	}
+
+	this->mutex.unlock();
+}
+
+void WFCondition::broadcast(void *msg)
+{
+	WFWaitTask *task;
+	struct list_head *pos, *tmp;
+
+	this->mutex.lock();
+	if (!list_empty(&this->waiter_list))
+	{
+		list_for_each_safe(pos, tmp, &this->waiter_list)
+		{
+			task = list_entry(pos, WFWaitTask, list);
+			list_del(pos);
+			task->send(msg);
+		}
+	}
+	this->mutex.unlock();
+}
+

src/factory/WFSemaphore.h

@@ -0,0 +1,164 @@
+/*
+  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)
+          Liu Kai (liukaidx@sogou-inc.com)
+*/
+
+#ifndef _WFSEMAPHORE_H_
+#define _WFSEMAPHORE_H_
+
+#include <mutex>
+#include <time.h>
+#include <functional>
+#include <atomic>
+#include "list.h"
+#include "WFTask.h"
+#include "WFTaskFactory.h"
+#include "WFGlobal.h"
+
+class WFSemaphore
+{
+public:
+	WFMailboxTask *acquire(std::function<void (WFMailboxTask *)> cb);
+	void release(void *msg);
+
+public:
+	std::mutex mutex;
+	struct list_head waiter_list;
+	
+public:
+	WFSemaphore(int value)
+	{
+		INIT_LIST_HEAD(&this->waiter_list);
+		this->concurrency = value;
+//		this->size = value;
+	}
+
+private:
+	std::atomic<int> concurrency;
+//	int size;
+};
+
+class WFCondition : public WFSemaphore
+{	
+public:
+	void signal(void *msg);
+	void broadcast(void *msg);
+
+public:
+	WFCondition() : WFSemaphore(1) { }
+	WFCondition(int value) : WFSemaphore(value) { }
+	~WFCondition() { }
+};
+
+class WFSemaphoreTask : public WFMailboxTask
+{
+public:
+	WFSemaphoreTask(std::function<void (WFMailboxTask *)>&& cb) :
+		WFMailboxTask(std::move(cb))
+//		WFMailboxTask(&this->msg, 1, std::move(cb))
+	{
+		this->list.next = NULL;
+	}
+
+	virtual ~WFSemaphoreTask() { }
+
+public:
+	struct list_head list;
+
+private:
+	void *msg; // TODO: user_data instead
+};
+
+class WFTimedWaitTask;
+
+class WFWaitTask : public WFSemaphoreTask
+{
+public:
+	void set_timer(WFTimedWaitTask *timer) { this->timer = timer; }
+	void clear_timer_waiter();
+	void set_error(int error) { this->error = error; }
+
+protected:
+	void dispatch();
+
+private:
+	WFTimedWaitTask *timer;
+
+public:
+	WFWaitTask(std::function<void (WFMailboxTask *)>&& cb) :
+		WFSemaphoreTask(std::move(cb))
+	{
+		this->timer = NULL;
+	}
+
+	virtual ~WFWaitTask() { }
+};
+
+class WFTimedWaitTask : public __WFTimerTask
+{
+public:
+	WFTimedWaitTask(WFWaitTask *wait_task, std::mutex *mutex,
+					const struct timespec *value,
+					CommScheduler *scheduler,
+					std::function<void (WFTimerTask *)> cb) :
+		__WFTimerTask(value, scheduler, std::move(cb))
+	{
+		this->mutex = mutex;
+		this->wait_task = wait_task;
+	}
+
+	void clear_wait_task()
+	{
+		this->mutex->lock();
+		this->wait_task = NULL;
+		this->mutex->unlock();
+	}
+
+protected:
+	virtual SubTask *done();
+
+private:
+	std::mutex *mutex;
+	WFWaitTask *wait_task;
+};
+
+class WFSwitchWaitTask : public WFWaitTask
+{
+public:
+	WFSwitchWaitTask(std::function<void (WFMailboxTask *)>&& cb) :
+			WFWaitTask(std::move(cb))
+	{ }
+
+protected:
+	SubTask *done()
+	{
+		SeriesWork *series = series_of(this);
+
+		WFTimerTask *switch_task = WFTaskFactory::create_timer_task(0,
+			[this](WFTimerTask *task) {
+				if (this->callback)
+					this->callback(this);
+				delete this;
+		});
+		series->push_front(switch_task);
+
+		return series->pop();
+	}
+};
+
+#endif
+

src/factory/WebSocketTaskImpl.cc

@@ -130,7 +130,7 @@ SubTask *ComplexWebSocketOutTask::done()
 
 	pthread_mutex_lock(&channel->mutex);
 	channel->set_sending(false);
-	channel->condition.signal();
+	channel->condition.signal(NULL);
 	pthread_mutex_unlock(&channel->mutex);
 
 	return WFChannelOutTask<WebSocketFrame>::done();
@@ -201,7 +201,7 @@ void ComplexWebSocketChannel::handle_in(CommMessageIn *in)
 	if (!parse_websocket) // so this is equal to should_count
 	{
 		pthread_mutex_lock(&this->mutex);
-		this->condition.signal();
+		this->condition.signal(NULL);
 		pthread_mutex_unlock(&this->mutex);
 		return;
 	}

src/manager/CMakeLists.txt

@@ -5,7 +5,6 @@ set(SRC
 	DnsCache.cc
 	UpstreamManager.cc
 	RouteManager.cc
-	WFCondition.cc
 	WFGlobal.cc
 )
 

src/manager/WFCondition.cc

@@ -24,7 +24,7 @@
 
 class WFTimedWaitTask;
 
-class WFWaitTask : public WFCounterTask
+class WFWaitTask : public WFMailboxTask
 {
 public:
 	void set_timer(WFTimedWaitTask *timer) { this->timer = timer; }
@@ -44,8 +44,8 @@ private:
 	WFTimedWaitTask *timer;
 
 public:
-	WFWaitTask(std::function<void (WFCounterTask *)>&& cb) :
-		WFCounterTask(1, std::move(cb))
+	WFWaitTask(std::function<void (WFMailboxTask *)>&& cb) :
+		WFMailboxTask(&this->user_data, 1, std::move(cb))
 	{
 		this->timer = NULL;
 		this->entry.list.next = NULL;
@@ -81,7 +81,7 @@ void WFWaitTask::dispatch()
 	if (this->timer)
 		timer->dispatch();
 	
-	this->WFCounterTask::count();
+	this->WFMailboxTask::count();
 }
 
 SubTask *WFWaitTask::done()
@@ -139,7 +139,7 @@ void WFTimedWaitTask::clear_wait_task()
 	this->mutex->unlock();
 }
 
-WFCounterTask *WFCondition::create_wait_task(counter_callback_t cb)
+WFMailboxTask *WFCondition::create_wait_task(mailbox_callback_t cb)
 {
 	WFWaitTask *task = new WFWaitTask(std::move(cb));
 
@@ -150,8 +150,8 @@ WFCounterTask *WFCondition::create_wait_task(counter_callback_t cb)
 	return task;
 }
 
-WFCounterTask *WFCondition::create_timedwait_task(const struct timespec *abstime,
-												  counter_callback_t cb)
+WFMailboxTask *WFCondition::create_timedwait_task(const struct timespec *abstime,
+												  mailbox_callback_t cb)
 {
 	WFWaitTask *waiter = new WFWaitTask(std::move(cb));
 	WFTimedWaitTask *task = new WFTimedWaitTask(waiter, &this->mutex, abstime,

src/manager/WFCondition.h

@@ -28,9 +28,9 @@
 class WFCondition
 {
 public:
-	WFCounterTask *create_wait_task(std::function<void (WFCounterTask *)> cb);
-	WFCounterTask *create_timedwait_task(const struct timespec *abstime,
-										 std::function<void (WFCounterTask *)> cb);
+	WFMailboxTask *create_wait_task(std::function<void (WFMailboxTask *)> cb);
+	WFMailboxTask *create_timedwait_task(const struct timespec *abstime,
+										 std::function<void (WFMailboxTask *)> cb);
 	void signal();
 	void broadcast();
 

test/CMakeLists.txt

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

test/condition_unittest.cc

@@ -0,0 +1,108 @@
+/*
+  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 <chrono>
+#include <mutex>
+#include <gtest/gtest.h>
+#include "workflow/WFTask.h"
+#include "workflow/WFTaskFactory.h"
+//#include "workflow/WFCondition.h"
+#include "workflow/WFSemaphore.h"
+#include "workflow/WFSemTaskFactory.h"
+#include "workflow/WFFacilities.h"
+
+TEST(condition_unittest, signal)
+{
+	WFCondition cond;
+	std::mutex mutex;
+	int ret = 1979;
+	int *ptr = &ret;
+
+	auto *task1 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) {
+		*ptr = 1412;
+		fprintf(stderr, "task1 callback\n");
+	});
+	SeriesWork *series1 = Workflow::create_series_work(task1, nullptr);
+
+	auto *task2 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) {
+		*ptr = 2;
+		fprintf(stderr, "task2 callback\n");
+	});
+	SeriesWork *series2 = Workflow::create_series_work(task2, nullptr);
+
+	series1->start();
+	series2->start();
+
+	mutex.lock();
+	cond.signal(NULL);
+	mutex.unlock();
+	usleep(10000);
+	EXPECT_EQ(ret, 1412);
+	cond.signal(NULL);
+	usleep(10000);
+	EXPECT_EQ(ret, 2);
+}
+
+TEST(condition_unittest, broadcast)
+{
+	WFCondition cond;
+	std::mutex mutex;
+	int ret = 0;
+	int *ptr = &ret;
+
+	auto *task1 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) {
+		(*ptr)++;
+		fprintf(stderr, "task1 callback\n");
+	});
+	SeriesWork *series1 = Workflow::create_series_work(task1, nullptr);
+
+	auto *task2 = WFSemTaskFactory::create_wait_task(&cond, [&ptr](WFMailboxTask *) {
+		(*ptr)++;
+		fprintf(stderr, "task2 callback\n");
+	});
+	SeriesWork *series2 = Workflow::create_series_work(task2, nullptr);
+
+	series1->start();
+	series2->start();
+
+	cond.broadcast(NULL);
+	usleep(10000);
+	EXPECT_EQ(ret, 2);
+}
+
+TEST(condition_unittest, timedwait)
+{
+	WFFacilities::WaitGroup wait_group(1);
+	struct timespec ts;
+	ts.tv_sec = 2;
+	ts.tv_nsec = 0;
+
+	auto *task1 = WFSemTaskFactory::create_timedwait_task("timedwait", &ts,
+		[&wait_group](WFMailboxTask *task) {
+		EXPECT_EQ(task->get_error(), ETIMEDOUT);
+		wait_group.done();
+	});
+
+	Workflow::start_series_work(task1, nullptr);
+	wait_group.wait();
+	usleep(10000);
+}
+