add BaseResource for WFCondition

src/manager/WFCondition.cc

@@ -22,14 +22,64 @@
 #include "WFCondTask.h"
 #include "WFCondition.h"
 
+int WFCondition::get(void **pmsg)
+{
+//	if (this->res == NULL)
+//		return -2;
+	int ret;
+
+	this->mutex->lock();
+	if (this->empty == 1)
+	{
+		*pmsg = this->res->get();
+		ret = 1;
+	}
+	else if (--this->empty == 0)
+	{
+		ret = 0;
+	}
+	else
+	{
+		WFCondWaitTask *task = new WFCondWaitTask(nullptr);
+		list_add_tail(&task->list, &this->get_list);
+		ret = -1;
+	}
+	this->mutex->unlock();
+
+	return ret;
+}
+
+WFWaitTask *WFCondition::get_wait_task(wait_callback_t callback)
+{
+	WFCondWaitTask *task = NULL;
+ 	struct list_head *pos;
+
+ 	this->mutex->lock();
+ 	if (!list_empty(&this->get_list))
+ 	{
+ 		pos = this->get_list.next;
+ 		list_move_tail(pos, &this->wait_list);
+ 		task = list_entry(pos, WFCondWaitTask, list);
+ 		task->set_callback(std::move(callback));
+ 	}
+
+ 	this->mutex->unlock();
+ 	return task;
+}
+
 void WFCondition::signal(void *msg)
 {
 	WFCondWaitTask *task = NULL;
 
 	this->mutex->lock();
+
 	if (!list_empty(&this->wait_list))
-	{
 		task = list_entry(this->wait_list.next, WFCondWaitTask, list);
+	else if (!list_empty(&this->get_list))
+		task = list_entry(this->get_list.next, WFCondWaitTask, list);
+
+	if (task)
+	{
 		list_del(&task->list);
 		task->clear_locked();
 	}
@@ -56,6 +106,16 @@ void WFCondition::broadcast(void *msg)
 		}
 	}
 
+	if (!list_empty(&this->get_list))
+	{
+		list_for_each_safe(pos, tmp, &this->get_list)
+		{
+			list_move_tail(pos, &tmp_list);
+			task = list_entry(pos, WFCondWaitTask, list);
+			task->clear_locked();
+		}
+	}
+
 	this->mutex->unlock();
 	while (!list_empty(&tmp_list))
 	{

src/manager/WFCondition.h

@@ -36,20 +36,55 @@ class WFCondition
 public:
 	void signal(void *msg);
 	void broadcast(void *msg);
-
-public:
-	WFCondition()
-	{
-		this->mutex = new std::mutex;
-		this->ref = new std::atomic<int>(1);
-		INIT_LIST_HEAD(&this->wait_list);
-	}
-	virtual ~WFCondition();
+	// 1: existed; 0: should get; -1: should wait;
+	int get(void **pmsg);
+	WFWaitTask *get_wait_task(wait_callback_t callback);
 
 public:
 	std::atomic<int> *ref;
 	std::mutex *mutex;
+	struct list_head get_list;
 	struct list_head wait_list;
+
+public:
+	class BaseResource
+	{
+	public:
+		virtual void *get() = 0;
+		virtual void set() { this->ptr->empty = 0; }
+		void clear() { this->ptr->empty = 1; }
+
+	private:
+		WFCondition *ptr;
+		friend WFCondition;
+	};
+
+private:
+	BaseResource *res;
+	int empty;
+
+public:
+	WFCondition(BaseResource *res)
+	{
+		this->empty = 1;
+		this->res = res;
+		this->res->ptr = this;
+		this->mutex = new std::mutex;
+		this->ref = new std::atomic<int>(1);
+		INIT_LIST_HEAD(&this->get_list);
+		INIT_LIST_HEAD(&this->wait_list);
+	}
+
+	WFCondition()
+	{
+		this->empty = 0;
+		this->res = NULL;
+		this->mutex = new std::mutex;
+		this->ref = new std::atomic<int>(1);
+		INIT_LIST_HEAD(&this->get_list);
+		INIT_LIST_HEAD(&this->wait_list);
+	}
+	virtual ~WFCondition();
 };
 
 #endif

test/condition_unittest.cc

@@ -122,10 +122,10 @@ TEST(condition_unittest, timedwait)
 
 TEST(condition_unittest, resource_pool)
 {
-	int res__concurrency = 3;
+	int res_concurrency = 3;
 	int task_concurrency = 10;
 	const char *words[3] = {"workflow", "srpc", "pyworkflow"};
-	WFResourcePool res_pool((void **)words, res__concurrency);
+	WFResourcePool res_pool((void **)words, res_concurrency);
 	WFFacilities::WaitGroup wg(task_concurrency);
 
 	for (int i = 0; i < task_concurrency; i++)
@@ -148,3 +148,86 @@ TEST(condition_unittest, resource_pool)
 	wg.wait();
 }
 
+class TestResource : public WFCondition::BaseResource
+{
+public:
+	TestResource(std::string msg) : msg(msg) { }
+	void *get() { return &this->msg; }
+	void set(std::string&& msg) { this->msg = std::move(msg);}
+
+private:
+	std::string msg;
+};
+
+TestResource res("");
+
+void wait_callback(WFWaitTask *task)
+{
+	void **msg;
+	size_t n;
+	msg = task->get_mailbox(&n);
+	std::string *str = *msg;
+	fprintf(stderr, "waiter get msg:%s\n", str->c_str());
+	series->set_context(*msg);
+}
+
+void work(std::string *msg, WFFacilities::WaitGroup &wg)
+{
+	fprintf(stderr, "working on message:%s\n", msg->c_str());
+	wg.done();
+}
+
+SubTask *create_router_task(WFCondition *cond, std::function<WFTimerTask *task> cb)
+{
+	auto *router = WFTaskFactory::create_timer_task(100000, std::move(cb));
+	router->user_data = cond;
+}
+
+void router_callback(WFTimerTask *task)
+{
+	fprintf(stderr, "finish routing and should broadcast every one else\n");
+	
+	((WFCondition *)(task->user_data))->broadcast();
+}
+
+TEST(condition_unittest, res_condition)
+{
+	int task_concurrency = 10;
+	WFFacilities::WaitGroup wg(task_concurrency);
+	WFCondition cond((WFCondition::BaseResource *)&res);
+
+	for (int i = 0; i < task_concurrency; i++)
+	{
+		auto *timer = WFTaskFactory::create_timer_task(i * 100000,
+		[&cond](WFTimerTask *task) {
+			std::string *msg;
+			int ret = cond.get(&msg);
+			if (ret == -1)
+			{
+				auto *waiter = cond.create_wait_task(wait_callback);
+				series_of(task)->push_front(waiter);
+				fprintf(stderr, "task-%lu waiting\n", id);
+			}
+			else if (ret == 0)
+			{
+				auto *router = create_router_task(route_callback);
+				series_of(task)->push_front(router);
+				fprintf(stderr, "task-%lu routing\n", id);
+			}
+			else
+			{
+				fprintf(stderr, "task-%lu get [%s]\n", id, (*msg).c_str());
+			}
+		});
+
+		SeriesWork *series = Workflow::create_series_work(timer, nullptr);
+		auto *worker = WFTaskFactory::create_go_task(work,
+													 (std::string *)series->get_context(),
+													 &wg);
+		series->push_back(worker);
+		series->start();
+	}
+
+	wg.wait();
+}
+