Merge pull request #895 from Barenboim/master

Add WFModuleTask

src/factory/WFTask.h

@@ -299,12 +299,12 @@ protected:
 		CommRequest(object, scheduler),
 		callback(std::move(cb))
 	{
-		this->user_data = NULL;
 		this->send_timeo = -1;
 		this->receive_timeo = -1;
 		this->keep_alive_timeo = 0;
 		this->target = NULL;
 		this->timeout_reason = TOR_NOT_TIMEOUT;
+		this->user_data = NULL;
 		this->state = WFT_STATE_UNDEFINED;
 		this->error = 0;
 	}
@@ -721,6 +721,75 @@ protected:
 	virtual ~WFGoTask() { }
 };
 
+class WFModuleTask : protected SeriesWork, public ParallelTask
+{
+public:
+	void start()
+	{
+		assert(!series_of(this));
+		Workflow::start_series_work(this, nullptr);
+	}
+
+	void dismiss()
+	{
+		assert(!series_of(this));
+		delete this;
+	}
+
+public:
+	SeriesWork *sub_series() { return this; }
+
+	const SeriesWork *sub_series() const { return this; }
+
+public:
+	void *user_data;
+
+protected:
+	virtual SubTask *done()
+	{
+		SeriesWork *series = series_of(this);
+
+		if (this->callback)
+			this->callback(this);
+
+		this->first = NULL;
+		delete this;
+		return series->pop();
+	}
+
+protected:
+	SubTask *first;
+	std::function<void (const WFModuleTask *)> callback;
+
+public:
+	WFModuleTask(SubTask *first,
+				 std::function<void (const WFModuleTask *)>&& cb) :
+		SeriesWork(first, nullptr),
+		ParallelTask(&this->first, 1),
+		callback(std::move(cb))
+	{
+		this->first = first;
+		this->set_in_parallel();
+		this->user_data = NULL;
+	}
+
+protected:
+	virtual ~WFModuleTask()
+	{
+		SubTask *task = this->first;
+
+		if (task)
+		{
+			this->SeriesWork::callback = nullptr;
+			do
+			{
+				delete task;
+				task = this->pop_task();
+			} while (task);
+		}
+	}
+};
+
 #include "WFTask.inl"
 
 #endif

src/factory/WFTaskFactory.h

@@ -101,6 +101,8 @@ using WFEmptyTask = WFGenericTask;
 using WFDynamicTask = WFGenericTask;
 using dynamic_create_t = std::function<SubTask *(WFDynamicTask *)>;
 
+using module_callback_t = std::function<void (const WFModuleTask *)>;
+
 class WFTaskFactory
 {
 public:
@@ -291,6 +293,21 @@ public:
 	}
 
 	static WFDynamicTask *create_dynamic_task(dynamic_create_t create);
+
+public:
+	static WFModuleTask *create_module_task(SubTask *first,
+											module_callback_t callback)
+	{
+		return new WFModuleTask(first, std::move(callback));
+	}
+
+	static WFModuleTask *create_module_task(SubTask *first, SubTask *last,
+											module_callback_t callback)
+	{
+		WFModuleTask *task = new WFModuleTask(first, std::move(callback));
+		task->sub_series()->set_last_task(last);
+		return task;
+	}
 };
 
 template<class REQ, class RESP>

src/factory/Workflow.cc

@@ -33,6 +33,7 @@ SeriesWork::SeriesWork(SubTask *first, series_callback_t&& cb) :
 	this->back = 0;
 	this->in_parallel = false;
 	this->canceled = false;
+	assert(!series_of(first));
 	first->set_pointer(this);
 	this->first = first;
 	this->last = NULL;
@@ -47,13 +48,7 @@ void SeriesWork::dismiss_recursive()
 	this->callback = nullptr;
 	do
 	{
-		ParallelWork *parallel = dynamic_cast<ParallelWork *>(task);
-
-		if (parallel)
-			parallel->dismiss_recursive();
-		else
-			delete task;
-
+		delete task;
 		task = this->pop_task();
 	} while (task);
 }
@@ -118,13 +113,7 @@ SubTask *SeriesWork::pop()
 
 	while (task)
 	{
-		ParallelWork *parallel = dynamic_cast<ParallelWork *>(task);
-
-		if (parallel)
-			parallel->dismiss_recursive();
-		else
-			delete task;
-
+		delete task;
 		task = this->pop_task();
 	}
 
@@ -218,16 +207,6 @@ void ParallelWork::add_series(SeriesWork *series)
 	this->subtasks_nr++;
 }
 
-void ParallelWork::dismiss_recursive()
-{
-	size_t i;
-
-	for (i = 0; i < this->subtasks_nr; i++)
-		this->all_series[i]->dismiss_recursive();
-
-	delete this;
-}
-
 SubTask *ParallelWork::done()
 {
 	SeriesWork *series = series_of(this);
@@ -239,7 +218,18 @@ SubTask *ParallelWork::done()
 	for (i = 0; i < this->subtasks_nr; i++)
 		delete this->all_series[i];
 
+	this->subtasks_nr = 0;
 	delete this;
 	return series->pop();
 }
 
+ParallelWork::~ParallelWork()
+{
+	size_t i;
+
+	for (i = 0; i < this->subtasks_nr; i++)
+		this->all_series[i]->dismiss_recursive();
+
+	delete []this->subtasks;
+}
+

src/factory/Workflow.h

@@ -105,9 +105,11 @@ public:
 	}
 
 public:
-	/* The next 3 methods are intended for task implementations only. */
+	/* Call only in tasks' done() implementations. */
 	SubTask *pop();
 
+public:
+	/* The following functions are intended for internal usages only. */
 	void set_last_task(SubTask *last)
 	{
 		last->set_pointer(this);
@@ -116,14 +118,18 @@ public:
 
 	void unset_last_task() { this->last = NULL; }
 
+	void set_in_parallel() { this->in_parallel = true; }
+
+	SubTask *pop_task();
+
+	void dismiss_recursive();
+
 protected:
 	void *context;
 	series_callback_t callback;
 
 private:
-	SubTask *pop_task();
 	void expand_queue();
-	void dismiss_recursive();
 
 private:
 	SubTask *first;
@@ -202,7 +208,7 @@ public:
 	void dismiss()
 	{
 		assert(!series_of(this));
-		this->dismiss_recursive();
+		delete this;
 	}
 
 public:
@@ -243,7 +249,6 @@ protected:
 
 private:
 	void expand_buf();
-	void dismiss_recursive();
 
 private:
 	size_t buf_size;
@@ -253,8 +258,7 @@ protected:
 	ParallelWork(parallel_callback_t&& callback);
 	ParallelWork(SeriesWork *const all_series[], size_t n,
 				 parallel_callback_t&& callback);
-	virtual ~ParallelWork() { delete []this->subtasks; }
-	friend class SeriesWork;
+	virtual ~ParallelWork();
 	friend class Workflow;
 };