From ca8812f6a5d583717afaff3ac3108e0e0f4066a4 Mon Sep 17 00:00:00 2001
From: Xie Han <63350856@qq.com>
Date: Sat, 3 Jun 2023 00:20:44 +0800
Subject: [PATCH 1/4] Add WFShuffleTask.

---
 src/factory/WFAlgoTaskFactory.h   | 49 ++++++++++++++++----
 src/factory/WFAlgoTaskFactory.inl | 77 +++++++++++++++++++++++++++++++
 2 files changed, 117 insertions(+), 9 deletions(-)

diff --git a/src/factory/WFAlgoTaskFactory.h b/src/factory/WFAlgoTaskFactory.h
index 88722c30..af8e0936 100644
--- a/src/factory/WFAlgoTaskFactory.h
+++ b/src/factory/WFAlgoTaskFactory.h
@@ -59,6 +59,20 @@ struct MergeOutput
 	T *last;
 };
 
+template<typename T>
+struct ShuffleInput
+{
+	T *first;
+	T *last;
+};
+
+template<typename T>
+struct ShuffleOutput
+{
+	T *first;
+	T *last;
+};
+
 template<typename KEY = std::string, typename VAL = std::string>
 using ReduceInput = std::vector<std::pair<KEY, VAL>>;
 
@@ -79,6 +93,12 @@ 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 KEY = std::string, typename VAL = std::string>
 using WFReduceTask = WFThreadTask<algorithm::ReduceInput<KEY, VAL>,
 								  algorithm::ReduceOutput<KEY, VAL>>;
@@ -99,6 +119,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 +145,16 @@ 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 KEY = std::string, typename VAL = std::string,
 			 class RED = algorithm::reduce_function_t<KEY, VAL>,
diff --git a/src/factory/WFAlgoTaskFactory.inl b/src/factory/WFAlgoTaskFactory.inl
index d0901230..05c4a4a8 100644
--- a/src/factory/WFAlgoTaskFactory.inl
+++ b/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,81 @@ 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));
+}
+
 /****************** MapReduce ******************/
 
 template<typename KEY, typename VAL>

From 6fbcbe7fcf154d049b3e9e3f03ddbc8b55cf117d Mon Sep 17 00:00:00 2001
From: XieHan <XieHan@sogou-inc.com>
Date: Sun, 4 Jun 2023 02:35:26 +0800
Subject: [PATCH 2/4] Add WFRmoveTask and WFUniqueTask.

---
 src/factory/WFAlgoTaskFactory.h   | 52 +++++++++++++++++++++
 src/factory/WFAlgoTaskFactory.inl | 75 +++++++++++++++++++++++++++++++
 2 files changed, 127 insertions(+)

diff --git a/src/factory/WFAlgoTaskFactory.h b/src/factory/WFAlgoTaskFactory.h
index af8e0936..bbeda0af 100644
--- a/src/factory/WFAlgoTaskFactory.h
+++ b/src/factory/WFAlgoTaskFactory.h
@@ -73,6 +73,35 @@ struct ShuffleOutput
 	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 KEY = std::string, typename VAL = std::string>
 using ReduceInput = std::vector<std::pair<KEY, VAL>>;
 
@@ -99,6 +128,18 @@ using WFShuffleTask = WFThreadTask<algorithm::ShuffleInput<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 KEY = std::string, typename VAL = std::string>
 using WFReduceTask = WFThreadTask<algorithm::ReduceInput<KEY, VAL>,
 								  algorithm::ReduceOutput<KEY, VAL>>;
@@ -156,6 +197,17 @@ public:
 												 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 KEY = std::string, typename VAL = std::string,
 			 class RED = algorithm::reduce_function_t<KEY, VAL>,
 			 class CB = reduce_callback_t<KEY, VAL>>
diff --git a/src/factory/WFAlgoTaskFactory.inl b/src/factory/WFAlgoTaskFactory.inl
index 05c4a4a8..fec80d47 100644
--- a/src/factory/WFAlgoTaskFactory.inl
+++ b/src/factory/WFAlgoTaskFactory.inl
@@ -502,6 +502,81 @@ WFShuffleTask<T> *WFAlgoTaskFactory::create_shuffle_task(const std::string& name
 										   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));
+}
+
 /****************** MapReduce ******************/
 
 template<typename KEY, typename VAL>

From 036e00feb94a4c9dcb805cbd07d6245a79219aa0 Mon Sep 17 00:00:00 2001
From: XieHan <XieHan@sogou-inc.com>
Date: Sun, 4 Jun 2023 02:35:42 +0800
Subject: [PATCH 3/4] Remove include WFAlgoTaskFactory.h from WFTaskFactory.h.

---
 src/factory/WFTaskFactory.h       | 1 -
 tutorial/tutorial-07-sort_task.cc | 2 +-
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h
index c3047453..df2a45d4 100644
--- a/src/factory/WFTaskFactory.h
+++ b/src/factory/WFTaskFactory.h
@@ -34,7 +34,6 @@
 #include "WFTask.h"
 #include "WFGraphTask.h"
 #include "EndpointParams.h"
-#include "WFAlgoTaskFactory.h"
 
 // Network Client/Server tasks
 
diff --git a/tutorial/tutorial-07-sort_task.cc b/tutorial/tutorial-07-sort_task.cc
index ee965ad5..f471e1bc 100644
--- a/tutorial/tutorial-07-sort_task.cc
+++ b/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;

From 47e2910eef8740faba8ff7dc221d1a0ad831e281 Mon Sep 17 00:00:00 2001
From: Xie Han <63350856@qq.com>
Date: Mon, 5 Jun 2023 22:54:01 +0800
Subject: [PATCH 4/4] Add WFReverseTask, WFRotateTask.

---
 src/factory/WFAlgoTaskFactory.h   | 51 +++++++++++++++++++++
 src/factory/WFAlgoTaskFactory.inl | 75 +++++++++++++++++++++++++++++++
 2 files changed, 126 insertions(+)

diff --git a/src/factory/WFAlgoTaskFactory.h b/src/factory/WFAlgoTaskFactory.h
index bbeda0af..e85c0d3c 100644
--- a/src/factory/WFAlgoTaskFactory.h
+++ b/src/factory/WFAlgoTaskFactory.h
@@ -102,6 +102,35 @@ struct UniqueOutput
 	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>>;
 
@@ -140,6 +169,18 @@ using WFUniqueTask = WFThreadTask<algorithm::UniqueInput<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>>;
@@ -208,6 +249,16 @@ public:
 											   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>,
 			 class CB = reduce_callback_t<KEY, VAL>>
diff --git a/src/factory/WFAlgoTaskFactory.inl b/src/factory/WFAlgoTaskFactory.inl
index fec80d47..8184777a 100644
--- a/src/factory/WFAlgoTaskFactory.inl
+++ b/src/factory/WFAlgoTaskFactory.inl
@@ -577,6 +577,81 @@ WFUniqueTask<T> *WFAlgoTaskFactory::create_unique_task(const std::string& name,
 								 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>