From 0d151c1a7dca90eb001c38b74dddaa322a29d893 Mon Sep 17 00:00:00 2001 From: XieHan Date: Sun, 20 Jun 2021 17:38:22 +0800 Subject: [PATCH] Add mailbox task's factory functions --- src/factory/WFTask.h | 8 ++++++++ src/factory/WFTaskFactory.cc | 27 +++++++++++++++++++++++++++ src/factory/WFTaskFactory.h | 4 ++++ 3 files changed, 39 insertions(+) diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 421fa946..110a01a0 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -620,6 +620,14 @@ public: this->mailbox = mailbox; } + WFMailboxTask(std::function&& cb) : + next(&this->user_data), + value(2), + callback(std::move(cb)) + { + this->mailbox = &this->user_data; + } + protected: virtual ~WFMailboxTask() { } }; diff --git a/src/factory/WFTaskFactory.cc b/src/factory/WFTaskFactory.cc index ce42b3c0..0a9aca68 100644 --- a/src/factory/WFTaskFactory.cc +++ b/src/factory/WFTaskFactory.cc @@ -293,6 +293,33 @@ void WFTaskFactory::count_by_name(const std::string& counter_name, unsigned int __CounterMap::get_instance()->count_n(counter_name, n); } +/********MailboxTask*************/ + +class __WFMailboxTask : public WFMailboxTask +{ +public: + __WFMailboxTask(size_t size, mailbox_callback_t&& cb) : + WFMailboxTask(new void *[size], size, std::move(cb)) + { + } + + virtual ~__WFMailboxTask() + { + delete []this->mailbox; + } +}; + +WFMailboxTask *WFTaskFactory::create_mailbox_task(size_t size, + mailbox_callback_t callback) +{ + return new __WFMailboxTask(size, std::move(callback)); +} + +WFMailboxTask *WFTaskFactory::create_mailbox_task(mailbox_callback_t callback) +{ + return new WFMailboxTask(std::move(callback)); +} + /********FileIOTask*************/ class WFFilepreadTask : public WFFileIOTask diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index cc88f2e7..c17200ff 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -90,6 +90,7 @@ using fsync_callback_t = std::function; using timer_callback_t = std::function; using counter_callback_t = std::function; +// Mailbox is like counter with data passing using mailbox_callback_t = std::function; // Graph (DAG) task. @@ -229,6 +230,9 @@ public: static WFMailboxTask *create_mailbox_task(size_t size, mailbox_callback_t callback); + /* Use 'user_data' as mailbox. Store only one message. */ + static WFMailboxTask *create_mailbox_task(mailbox_callback_t callback); + public: template static WFGoTask *create_go_task(const std::string& queue_name,