Merge pull request #1289 from Barenboim/master

Add more algorithm tasks.

src/factory/WFAlgoTaskFactory.h

@@ -59,6 +59,78 @@ struct MergeOutput
 	T *last;
 };
 
+template<typename T>
+struct ShuffleInput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct ShuffleOutput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct RemoveInput
+{
+	T *first;
+	T *last;
+	T value;
+};
+
+template<typename T>
+struct RemoveOutput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct UniqueInput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct UniqueOutput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct ReverseInput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct ReverseOutput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct RotateInput
+{
+	T *first;
+	T *middle;
+	T *last;
+};
+
+template<typename T>
+struct RotateOutput
+{
+	T *first;
+	T *last;
+};
+
 template<typename KEY = std::string, typename VAL = std::string>
 using ReduceInput = std::vector<std::pair<KEY, VAL>>;
 
@@ -79,6 +151,36 @@ using WFMergeTask = WFThreadTask<algorithm::MergeInput<T>,
 template<typename T>
 using merge_callback_t = std::function<void (WFMergeTask<T> *)>;
 
+template<typename T>
+using WFShuffleTask = WFThreadTask<algorithm::ShuffleInput<T>,
+								   algorithm::ShuffleOutput<T>>;
+template<typename T>
+using shuffle_callback_t = std::function<void (WFShuffleTask<T> *)>;
+
+template<typename T>
+using WFRemoveTask = WFThreadTask<algorithm::RemoveInput<T>,
+								  algorithm::RemoveOutput<T>>;
+template<typename T>
+using remove_callback_t = std::function<void (WFRemoveTask<T> *)>;
+
+template<typename T>
+using WFUniqueTask = WFThreadTask<algorithm::UniqueInput<T>,
+								  algorithm::UniqueOutput<T>>;
+template<typename T>
+using unique_callback_t = std::function<void (WFUniqueTask<T> *)>;
+
+template<typename T>
+using WFReverseTask = WFThreadTask<algorithm::ReverseInput<T>,
+								   algorithm::ReverseOutput<T>>;
+template<typename T>
+using reverse_callback_t = std::function<void (WFReverseTask<T> *)>;
+
+template<typename T>
+using WFRotateTask = WFThreadTask<algorithm::RotateInput<T>,
+								  algorithm::RotateOutput<T>>;
+template<typename T>
+using rotate_callback_t = std::function<void (WFRotateTask<T> *)>;
+
 template<typename KEY = std::string, typename VAL = std::string>
 using WFReduceTask = WFThreadTask<algorithm::ReduceInput<KEY, VAL>,
 								  algorithm::ReduceOutput<KEY, VAL>>;
@@ -99,6 +201,17 @@ public:
 										   CMP compare,
 										   CB callback);
 
+	template<typename T, class CB = sort_callback_t<T>>
+	static WFSortTask<T> *create_psort_task(const std::string& queue_name,
+											T *first, T *last,
+											CB callback);
+
+	template<typename T, class CMP, class CB = sort_callback_t<T>>
+	static WFSortTask<T> *create_psort_task(const std::string& queue_name,
+											T *first, T *last,
+											CMP compare,
+											CB callback);
+
 	template<typename T, class CB = merge_callback_t<T>>
 	static WFMergeTask<T> *create_merge_task(const std::string& queue_name,
 											 T *first1, T *last1,
@@ -114,16 +227,37 @@ public:
 											 CMP compare,
 											 CB callback);
 
-	template<typename T, class CB = sort_callback_t<T>>
-	static WFSortTask<T> *create_psort_task(const std::string& queue_name,
-											T *first, T *last,
-											CB callback);
+	template<typename T, class CB = shuffle_callback_t<T>>
+	static WFShuffleTask<T> *create_shuffle_task(const std::string& queue_name,
+												 T *first, T *last,
+												 CB callback);
 
-	template<typename T, class CMP, class CB = sort_callback_t<T>>
-	static WFSortTask<T> *create_psort_task(const std::string& queue_name,
-											T *first, T *last,
-											CMP compare,
-											CB callback);
+	template<typename T, class URBG, class CB = shuffle_callback_t<T>>
+	static WFShuffleTask<T> *create_shuffle_task(const std::string& queue_name,
+												 T *first, T *last,
+												 URBG generator,
+												 CB callback);
+
+	template<typename T, class CB = remove_callback_t<T>>
+	static WFRemoveTask<T> *create_remove_task(const std::string& queue_name,
+											   T *first, T *last,
+											   T value,
+											   CB callback);
+
+	template<typename T, class CB = unique_callback_t<T>>
+	static WFUniqueTask<T> *create_unique_task(const std::string& queue_name,
+											   T *first, T *last,
+											   CB callback);
+
+	template<typename T, class CB = reverse_callback_t<T>>
+	static WFReverseTask<T> *create_reverse_task(const std::string& queue_name,
+												 T *first, T *last,
+												 CB callback);
+
+	template<typename T, class CB = rotate_callback_t<T>>
+	static WFRotateTask<T> *create_rotate_task(const std::string& queue_name,
+											   T *first, T *middle, T *last,
+											   CB callback);
 
 	template<typename KEY = std::string, typename VAL = std::string,
 			 class RED = algorithm::reduce_function_t<KEY, VAL>,

src/factory/WFAlgoTaskFactory.inl

@@ -17,6 +17,8 @@
 */
 
 #include <assert.h>
+#include <stdlib.h>
+#include <random>
 #include <algorithm>
 #include <vector>
 #include <functional>
@@ -425,6 +427,231 @@ WFSortTask<T> *WFAlgoTaskFactory::create_psort_task(const std::string& name,
 										  std::move(callback));
 }
 
+/****************** Shuffle ******************/
+
+template<typename T>
+class __WFShuffleTask : public WFShuffleTask<T>
+{
+protected:
+	virtual void execute()
+	{
+		std::shuffle(this->input.first, this->input.last,
+					 std::mt19937_64(random()));
+		this->output.first = this->input.first;
+		this->output.last = this->input.last;
+	}
+
+public:
+	__WFShuffleTask(ExecQueue *queue, Executor *executor,
+					T *first, T *last,
+					shuffle_callback_t<T>&& cb) :
+		WFShuffleTask<T>(queue, executor, std::move(cb))
+	{
+		this->input.first = first;
+		this->input.last = last;
+		this->output.first = NULL;
+		this->output.last = NULL;
+	}
+};
+
+template<typename T, class URBG>
+class __WFShuffleTaskGen : public __WFShuffleTask<T>
+{
+protected:
+	virtual void execute()
+	{
+		std::shuffle(this->input.first, this->input.last,
+					 std::move(this->generator));
+		this->output.first = this->input.first;
+		this->output.last = this->input.last;
+	}
+
+protected:
+	URBG generator;
+
+public:
+	__WFShuffleTaskGen(ExecQueue *queue, Executor *executor,
+					   T *first, T *last, URBG&& gen,
+					   shuffle_callback_t<T>&& cb) :
+		__WFShuffleTask<T>(queue, executor, std::move(cb)),
+		generator(std::move(gen))
+	{
+	}
+};
+
+template<typename T, class CB>
+WFShuffleTask<T> *WFAlgoTaskFactory::create_shuffle_task(const std::string& name,
+														 T *first, T *last,
+														 CB callback)
+{
+	return new __WFShuffleTask<T>(WFGlobal::get_exec_queue(name),
+								  WFGlobal::get_compute_executor(),
+								  first, last,
+								  std::move(callback));
+}
+
+template<typename T, class URBG, class CB>
+WFShuffleTask<T> *WFAlgoTaskFactory::create_shuffle_task(const std::string& name,
+														 T *first, T *last,
+														 URBG generator,
+														 CB callback)
+{
+	return new __WFShuffleTaskGen<T, URBG>(WFGlobal::get_exec_queue(name),
+										   WFGlobal::get_compute_executor(),
+										   first, last, std::move(generator),
+										   std::move(callback));
+}
+
+/****************** Remove ******************/
+
+template<typename T>
+class __WFRemoveTask : public WFRemoveTask<T>
+{
+protected:
+	virtual void execute()
+	{
+		this->output.last = std::remove(this->input.first, this->input.last,
+										this->input.value);
+		this->output.first = this->input.first;
+	}
+
+public:
+	__WFRemoveTask(ExecQueue *queue, Executor *executor,
+				   T *first, T *last, T&& value,
+				   remove_callback_t<T>&& cb) :
+		WFRemoveTask<T>(queue, executor, std::move(cb))
+	{
+		this->input.first = first;
+		this->input.last = last;
+		this->input.value = std::move(value);
+		this->output.first = NULL;
+		this->output.last = NULL;
+	}
+};
+
+template<typename T, class CB>
+WFRemoveTask<T> *WFAlgoTaskFactory::create_remove_task(const std::string& name,
+													   T *first, T *last,
+													   T value,
+													   CB callback)
+{
+	return new __WFRemoveTask<T>(WFGlobal::get_exec_queue(name),
+								 WFGlobal::get_compute_executor(),
+								 first, last, std::move(value),
+								 std::move(callback));
+}
+
+/****************** Unique ******************/
+
+template<typename T>
+class __WFUniqueTask : public WFUniqueTask<T>
+{
+protected:
+	virtual void execute()
+	{
+		this->output.last = std::unique(this->input.first, this->input.last);
+		this->output.first = this->input.first;
+	}
+
+public:
+	__WFUniqueTask(ExecQueue *queue, Executor *executor,
+				   T *first, T *last,
+				   unique_callback_t<T>&& cb) :
+		WFUniqueTask<T>(queue, executor, std::move(cb))
+	{
+		this->input.first = first;
+		this->input.last = last;
+		this->output.first = NULL;
+		this->output.last = NULL;
+	}
+};
+
+template<typename T, class CB>
+WFUniqueTask<T> *WFAlgoTaskFactory::create_unique_task(const std::string& name,
+													   T *first, T *last,
+													   CB callback)
+{
+	return new __WFUniqueTask<T>(WFGlobal::get_exec_queue(name),
+								 WFGlobal::get_compute_executor(),
+								 first, last,
+								 std::move(callback));
+}
+
+/****************** Reverse ******************/
+
+template<typename T>
+class __WFReverseTask : public WFReverseTask<T>
+{
+protected:
+	virtual void execute()
+	{
+		std::reverse(this->input.first, this->input.last);
+		this->output.first = this->input.first;
+		this->output.last = this->input.last;
+	}
+
+public:
+	__WFReverseTask(ExecQueue *queue, Executor *executor,
+					T *first, T *last,
+					reverse_callback_t<T>&& cb) :
+		WFReverseTask<T>(queue, executor, std::move(cb))
+	{
+		this->input.first = first;
+		this->input.last = last;
+		this->output.first = NULL;
+		this->output.last = NULL;
+	}
+};
+
+template<typename T, class CB>
+WFReverseTask<T> *WFAlgoTaskFactory::create_reverse_task(const std::string& name,
+														 T *first, T *last,
+														 CB callback)
+{
+	return new __WFReverseTask<T>(WFGlobal::get_exec_queue(name),
+								  WFGlobal::get_compute_executor(),
+								  first, last,
+								  std::move(callback));
+}
+
+/****************** Rotate ******************/
+
+template<typename T>
+class __WFRotateTask : public WFRotateTask<T>
+{
+protected:
+	virtual void execute()
+	{
+		std::rotate(this->input.first, this->input.middle, this->input.last);
+		this->output.first = this->input.first;
+		this->output.last = this->input.last;
+	}
+
+public:
+	__WFRotateTask(ExecQueue *queue, Executor *executor,
+					T *first, T* middle, T *last,
+					rotate_callback_t<T>&& cb) :
+		WFRotateTask<T>(queue, executor, std::move(cb))
+	{
+		this->input.first = first;
+		this->input.middle = middle;
+		this->input.last = last;
+		this->output.first = NULL;
+		this->output.last = NULL;
+	}
+};
+
+template<typename T, class CB>
+WFRotateTask<T> *WFAlgoTaskFactory::create_rotate_task(const std::string& name,
+													   T *first, T *middle, T *last,
+													   CB callback)
+{
+	return new __WFRotateTask<T>(WFGlobal::get_exec_queue(name),
+								 WFGlobal::get_compute_executor(),
+								 first, middle, last,
+								 std::move(callback));
+}
+
 /****************** MapReduce ******************/
 
 template<typename KEY, typename VAL>

src/factory/WFTaskFactory.h

@@ -34,7 +34,6 @@
 #include "WFTask.h"
 #include "WFGraphTask.h"
 #include "EndpointParams.h"
-#include "WFAlgoTaskFactory.h"
 
 // Network Client/Server tasks
 

tutorial/tutorial-07-sort_task.cc

@@ -18,7 +18,7 @@
 
 #include <stdlib.h>
 #include <stdio.h>
-#include "workflow/WFTaskFactory.h"
+#include "workflow/WFAlgoTaskFactory.h"
 #include "workflow/WFFacilities.h"
 
 using namespace algorithm;