diff --git a/src/factory/WFAlgoTaskFactory.h b/src/factory/WFAlgoTaskFactory.h index 88722c30..e85c0d3c 100644 --- a/src/factory/WFAlgoTaskFactory.h +++ b/src/factory/WFAlgoTaskFactory.h @@ -59,6 +59,78 @@ struct MergeOutput T *last; }; +template +struct ShuffleInput +{ + T *first; + T *last; +}; + +template +struct ShuffleOutput +{ + T *first; + T *last; +}; + +template +struct RemoveInput +{ + T *first; + T *last; + T value; +}; + +template +struct RemoveOutput +{ + T *first; + T *last; +}; + +template +struct UniqueInput +{ + T *first; + T *last; +}; + +template +struct UniqueOutput +{ + T *first; + T *last; +}; + +template +struct ReverseInput +{ + T *first; + T *last; +}; + +template +struct ReverseOutput +{ + T *first; + T *last; +}; + +template +struct RotateInput +{ + T *first; + T *middle; + T *last; +}; + +template +struct RotateOutput +{ + T *first; + T *last; +}; + template using ReduceInput = std::vector>; @@ -79,6 +151,36 @@ using WFMergeTask = WFThreadTask, template using merge_callback_t = std::function *)>; +template +using WFShuffleTask = WFThreadTask, + algorithm::ShuffleOutput>; +template +using shuffle_callback_t = std::function *)>; + +template +using WFRemoveTask = WFThreadTask, + algorithm::RemoveOutput>; +template +using remove_callback_t = std::function *)>; + +template +using WFUniqueTask = WFThreadTask, + algorithm::UniqueOutput>; +template +using unique_callback_t = std::function *)>; + +template +using WFReverseTask = WFThreadTask, + algorithm::ReverseOutput>; +template +using reverse_callback_t = std::function *)>; + +template +using WFRotateTask = WFThreadTask, + algorithm::RotateOutput>; +template +using rotate_callback_t = std::function *)>; + template using WFReduceTask = WFThreadTask, algorithm::ReduceOutput>; @@ -99,6 +201,17 @@ public: CMP compare, CB callback); + template> + static WFSortTask *create_psort_task(const std::string& queue_name, + T *first, T *last, + CB callback); + + template> + static WFSortTask *create_psort_task(const std::string& queue_name, + T *first, T *last, + CMP compare, + CB callback); + template> static WFMergeTask *create_merge_task(const std::string& queue_name, T *first1, T *last1, @@ -114,16 +227,37 @@ public: CMP compare, CB callback); - template> - static WFSortTask *create_psort_task(const std::string& queue_name, - T *first, T *last, - CB callback); + template> + static WFShuffleTask *create_shuffle_task(const std::string& queue_name, + T *first, T *last, + CB callback); - template> - static WFSortTask *create_psort_task(const std::string& queue_name, - T *first, T *last, - CMP compare, - CB callback); + template> + static WFShuffleTask *create_shuffle_task(const std::string& queue_name, + T *first, T *last, + URBG generator, + CB callback); + + template> + static WFRemoveTask *create_remove_task(const std::string& queue_name, + T *first, T *last, + T value, + CB callback); + + template> + static WFUniqueTask *create_unique_task(const std::string& queue_name, + T *first, T *last, + CB callback); + + template> + static WFReverseTask *create_reverse_task(const std::string& queue_name, + T *first, T *last, + CB callback); + + template> + static WFRotateTask *create_rotate_task(const std::string& queue_name, + T *first, T *middle, T *last, + CB callback); template, diff --git a/src/factory/WFAlgoTaskFactory.inl b/src/factory/WFAlgoTaskFactory.inl index d0901230..8184777a 100644 --- a/src/factory/WFAlgoTaskFactory.inl +++ b/src/factory/WFAlgoTaskFactory.inl @@ -17,6 +17,8 @@ */ #include +#include +#include #include #include #include @@ -425,6 +427,231 @@ WFSortTask *WFAlgoTaskFactory::create_psort_task(const std::string& name, std::move(callback)); } +/****************** Shuffle ******************/ + +template +class __WFShuffleTask : public WFShuffleTask +{ +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&& cb) : + WFShuffleTask(queue, executor, std::move(cb)) + { + this->input.first = first; + this->input.last = last; + this->output.first = NULL; + this->output.last = NULL; + } +}; + +template +class __WFShuffleTaskGen : public __WFShuffleTask +{ +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&& cb) : + __WFShuffleTask(queue, executor, std::move(cb)), + generator(std::move(gen)) + { + } +}; + +template +WFShuffleTask *WFAlgoTaskFactory::create_shuffle_task(const std::string& name, + T *first, T *last, + CB callback) +{ + return new __WFShuffleTask(WFGlobal::get_exec_queue(name), + WFGlobal::get_compute_executor(), + first, last, + std::move(callback)); +} + +template +WFShuffleTask *WFAlgoTaskFactory::create_shuffle_task(const std::string& name, + T *first, T *last, + URBG generator, + CB callback) +{ + return new __WFShuffleTaskGen(WFGlobal::get_exec_queue(name), + WFGlobal::get_compute_executor(), + first, last, std::move(generator), + std::move(callback)); +} + +/****************** Remove ******************/ + +template +class __WFRemoveTask : public WFRemoveTask +{ +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&& cb) : + WFRemoveTask(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 +WFRemoveTask *WFAlgoTaskFactory::create_remove_task(const std::string& name, + T *first, T *last, + T value, + CB callback) +{ + return new __WFRemoveTask(WFGlobal::get_exec_queue(name), + WFGlobal::get_compute_executor(), + first, last, std::move(value), + std::move(callback)); +} + +/****************** Unique ******************/ + +template +class __WFUniqueTask : public WFUniqueTask +{ +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&& cb) : + WFUniqueTask(queue, executor, std::move(cb)) + { + this->input.first = first; + this->input.last = last; + this->output.first = NULL; + this->output.last = NULL; + } +}; + +template +WFUniqueTask *WFAlgoTaskFactory::create_unique_task(const std::string& name, + T *first, T *last, + CB callback) +{ + return new __WFUniqueTask(WFGlobal::get_exec_queue(name), + WFGlobal::get_compute_executor(), + first, last, + std::move(callback)); +} + +/****************** Reverse ******************/ + +template +class __WFReverseTask : public WFReverseTask +{ +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&& cb) : + WFReverseTask(queue, executor, std::move(cb)) + { + this->input.first = first; + this->input.last = last; + this->output.first = NULL; + this->output.last = NULL; + } +}; + +template +WFReverseTask *WFAlgoTaskFactory::create_reverse_task(const std::string& name, + T *first, T *last, + CB callback) +{ + return new __WFReverseTask(WFGlobal::get_exec_queue(name), + WFGlobal::get_compute_executor(), + first, last, + std::move(callback)); +} + +/****************** Rotate ******************/ + +template +class __WFRotateTask : public WFRotateTask +{ +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&& cb) : + WFRotateTask(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 +WFRotateTask *WFAlgoTaskFactory::create_rotate_task(const std::string& name, + T *first, T *middle, T *last, + CB callback) +{ + return new __WFRotateTask(WFGlobal::get_exec_queue(name), + WFGlobal::get_compute_executor(), + first, middle, last, + std::move(callback)); +} + /****************** MapReduce ******************/ template 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 #include -#include "workflow/WFTaskFactory.h" +#include "workflow/WFAlgoTaskFactory.h" #include "workflow/WFFacilities.h" using namespace algorithm;