transwarp
 All Classes Namespaces Functions Variables Typedefs Enumerations Enumerator Pages
transwarp.h
1 /// @mainpage transwarp is a header-only C++ library for task concurrency
2 /// @details https://github.com/bloomen/transwarp
3 /// @version 1.6.1
4 /// @author Christian Blume, Guan Wang
5 /// @date 2018
6 /// @copyright MIT http://www.opensource.org/licenses/mit-license.php
7 #pragma once
8 #include <algorithm>
9 #include <atomic>
10 #include <chrono>
11 #include <cstddef>
12 #include <functional>
13 #include <future>
14 #include <memory>
15 #include <mutex>
16 #include <queue>
17 #include <stdexcept>
18 #include <string>
19 #include <thread>
20 #include <tuple>
21 #include <type_traits>
22 #include <vector>
23 
24 
25 /// The transwarp namespace
26 namespace transwarp {
27 
28 
29 /// The possible task types
30 enum class task_type {
31  root, ///< The task has no parents
32  accept, ///< The task's functor accepts all parent futures
33  accept_any, ///< The task's functor accepts the first parent future that becomes ready
34  consume, ///< The task's functor consumes all parent results
35  consume_any, ///< The task's functor consumes the first parent result that becomes ready
36  wait, ///< The task's functor takes no arguments but waits for all parents to finish
37  wait_any, ///< The task's functor takes no arguments but waits for the first parent to finish
38 };
39 
40 
41 /// Base class for exceptions
42 class transwarp_error : public std::runtime_error {
43 public:
44  explicit transwarp_error(const std::string& message)
45  : std::runtime_error(message)
46  {}
47 };
48 
49 
50 /// Exception thrown when a task is canceled
52 public:
53  explicit task_canceled(const std::string& node_repr)
54  : transwarp::transwarp_error("Task canceled: " + node_repr)
55  {}
56 };
57 
58 
59 /// Exception thrown when a task was destroyed prematurely
61 public:
62  explicit task_destroyed(const std::string& node_repr)
63  : transwarp::transwarp_error("Task destroyed: " + node_repr)
64  {}
65 };
66 
67 
68 /// Exception thrown when an invalid parameter was passed to a function
70 public:
71  explicit invalid_parameter(const std::string& parameter)
72  : transwarp::transwarp_error("Invalid parameter: " + parameter)
73  {}
74 };
75 
76 
77 /// Exception thrown when a task is used in unintended ways
79 public:
80  explicit control_error(const std::string& message)
81  : transwarp::transwarp_error("Control error: " + message)
82  {}
83 };
84 
85 
86 /// String conversion for the task_type enumeration
87 inline std::string to_string(const transwarp::task_type& type) {
88  switch (type) {
89  case transwarp::task_type::root: return "root";
90  case transwarp::task_type::accept: return "accept";
91  case transwarp::task_type::accept_any: return "accept_any";
92  case transwarp::task_type::consume: return "consume";
93  case transwarp::task_type::consume_any: return "consume_any";
94  case transwarp::task_type::wait: return "wait";
95  case transwarp::task_type::wait_any: return "wait_any";
96  default: throw transwarp::invalid_parameter("task type");
97  }
98 }
99 
100 
101 /// The root type. Used for tag dispatch
102 struct root_type : std::integral_constant<transwarp::task_type, transwarp::task_type::root> {};
103 constexpr transwarp::root_type root{}; ///< The root task tag
104 
105 /// The accept type. Used for tag dispatch
106 struct accept_type : std::integral_constant<transwarp::task_type, transwarp::task_type::accept> {};
107 constexpr transwarp::accept_type accept{}; ///< The accept task tag
108 
109 /// The accept_any type. Used for tag dispatch
110 struct accept_any_type : std::integral_constant<transwarp::task_type, transwarp::task_type::accept_any> {};
111 constexpr transwarp::accept_any_type accept_any{}; ///< The accept_any task tag
112 
113 /// The consume type. Used for tag dispatch
114 struct consume_type : std::integral_constant<transwarp::task_type, transwarp::task_type::consume> {};
115 constexpr transwarp::consume_type consume{}; ///< The consume task tag
116 
117 /// The consume_any type. Used for tag dispatch
118 struct consume_any_type : std::integral_constant<transwarp::task_type, transwarp::task_type::consume_any> {};
119 constexpr transwarp::consume_any_type consume_any{}; ///< The consume_any task tag
120 
121 /// The wait type. Used for tag dispatch
122 struct wait_type : std::integral_constant<transwarp::task_type, transwarp::task_type::wait> {};
123 constexpr transwarp::wait_type wait{}; ///< The wait task tag
124 
125 /// The wait_any type. Used for tag dispatch
126 struct wait_any_type : std::integral_constant<transwarp::task_type, transwarp::task_type::wait_any> {};
127 constexpr transwarp::wait_any_type wait_any{}; ///< The wait_any task tag
128 
129 
130 /// Detail namespace for internal functionality only
131 namespace detail {
132 
133 struct visit_depth;
134 struct unvisit;
135 struct final_visitor;
136 struct schedule_visitor;
137 struct node_manip;
138 template<bool>
140 
141 } // detail
142 
143 
144 /// A node carrying meta-data of a task
145 class node {
146 public:
147 
148  node() = default;
149 
150  // delete copy/move semantics
151  node(const node&) = delete;
152  node& operator=(const node&) = delete;
153  node(node&&) = delete;
154  node& operator=(node&&) = delete;
155 
156  /// The task ID
157  std::size_t get_id() const noexcept {
158  return id_;
159  }
160 
161  /// The task level
162  std::size_t get_level() const noexcept {
163  return level_;
164  }
165 
166  /// The task type
167  transwarp::task_type get_type() const noexcept {
168  return type_;
169  }
170 
171  /// The optional task name (may be null)
172  const std::shared_ptr<std::string>& get_name() const noexcept {
173  return name_;
174  }
175 
176  /// The optional, task-specific executor (may be null)
177  const std::shared_ptr<std::string>& get_executor() const noexcept {
178  return executor_;
179  }
180 
181  /// The task's parents (may be empty)
182  const std::vector<std::shared_ptr<node>>& get_parents() const noexcept {
183  return parents_;
184  }
185 
186  /// The task priority (defaults to 0)
187  std::size_t get_priority() const noexcept {
188  return priority_;
189  }
190 
191  /// The custom task data (may be null)
192  const std::shared_ptr<void>& get_custom_data() const noexcept {
193  return custom_data_;
194  }
195 
196  /// Returns whether the associated task is canceled
197  bool is_canceled() const noexcept {
198  return canceled_.load();
199  }
200 
201 private:
202  friend struct transwarp::detail::node_manip;
203 
204  std::size_t id_ = 0;
205  std::size_t level_ = 0;
207  std::shared_ptr<std::string> name_;
208  std::shared_ptr<std::string> executor_;
209  std::vector<std::shared_ptr<node>> parents_;
210  std::size_t priority_ = 0;
211  std::shared_ptr<void> custom_data_;
212  std::atomic_bool canceled_{false};
213 };
214 
215 /// String conversion for the node class
216 inline std::string to_string(const transwarp::node& node, const std::string& seperator="\n") {
217  std::string s;
218  s += '"';
219  const std::shared_ptr<std::string>& name = node.get_name();
220  if (name) {
221  s += "<" + *name + ">" + seperator;
222  }
223  s += transwarp::to_string(node.get_type());
224  s += " id=" + std::to_string(node.get_id());
225  s += " lev=" + std::to_string(node.get_level());
226  const std::shared_ptr<std::string>& exec = node.get_executor();
227  if (exec) {
228  s += seperator + "<" + *exec + ">";
229  }
230  s += '"';
231  return s;
232 }
233 
234 
235 /// An edge between two nodes
236 class edge {
237 public:
238  // cppcheck-suppress passedByValue
239  edge(std::shared_ptr<transwarp::node> parent, std::shared_ptr<transwarp::node> child) noexcept
240  : parent_(std::move(parent)), child_(std::move(child))
241  {}
242 
243  // default copy/move semantics
244  edge(const edge&) = default;
245  edge& operator=(const edge&) = default;
246  edge(edge&&) = default;
247  edge& operator=(edge&&) = default;
248 
249  /// Returns the parent node
250  const std::shared_ptr<transwarp::node>& get_parent() const noexcept {
251  return parent_;
252  }
253 
254  /// Returns the child node
255  const std::shared_ptr<transwarp::node>& get_child() const noexcept {
256  return child_;
257  }
258 
259 private:
260  std::shared_ptr<transwarp::node> parent_;
261  std::shared_ptr<transwarp::node> child_;
262 };
263 
264 /// String conversion for the edge class
265 inline std::string to_string(const transwarp::edge& edge, const std::string& separator="\n") {
266  return transwarp::to_string(*edge.get_parent(), separator) + " -> " + transwarp::to_string(*edge.get_child(), separator);
267 }
268 
269 
270 /// Creates a dot-style string from the given graph
271 inline std::string to_string(const std::vector<transwarp::edge>& graph, const std::string& separator="\n") {
272  std::string dot = "digraph {" + separator;
273  for (const transwarp::edge& edge : graph) {
274  dot += transwarp::to_string(edge, separator) + separator;
275  }
276  dot += "}";
277  return dot;
278 }
279 
280 
281 /// The executor interface used to perform custom task execution
282 class executor {
283 public:
284  virtual ~executor() = default;
285 
286  /// Returns the name of the executor
287  virtual std::string get_name() const = 0;
288 
289  /// Runs a task which is wrapped by the given functor. The functor only
290  /// captures one shared pointer and can hence be copied at low cost.
291  /// node represents the task that the functor belongs to.
292  /// This function is only ever called on the thread of the caller to schedule().
293  /// The implementer needs to ensure that this never throws exceptions
294  virtual void execute(const std::function<void()>& functor, const std::shared_ptr<transwarp::node>& node) = 0;
295 };
296 
297 
298 /// The task events that can be subscribed to using the listener interface
299 enum class event_type {
300  before_scheduled, ///< Just before a task is scheduled (handle_event called on thread of caller to schedule())
301  before_started, ///< Just before a task starts running (handle_event called on thread that task is run on)
302  before_invoked, ///< Just before a task's functor is invoked (handle_event called on thread that task is run on)
303  after_finished, ///< Just after a task has finished running (handle_event called on thread that task is run on)
304  after_canceled, ///< Just after a task was canceled (handle_event called on thread that task is run on)
305  count,
306 };
307 
308 
309 /// The listener interface to listen to events raised by tasks
310 class listener {
311 public:
312  virtual ~listener() = default;
313 
314  /// This may be called from arbitrary threads depending on the event type (see transwarp::event_type).
315  /// The implementer needs to ensure that this never throws exceptions
316  virtual void handle_event(transwarp::event_type event, const std::shared_ptr<transwarp::node>& node) = 0;
317 };
318 
319 
320 /// Determines in which order tasks are scheduled in the graph
321 enum class schedule_type {
322  breadth, ///< Scheduling according to a breadth-first search (default)
323  depth, ///< Scheduling according to a depth-first search
324 };
325 
326 
327 /// An interface for the task class
328 class itask {
329 public:
330  virtual ~itask() = default;
331 
332  virtual void set_executor(std::shared_ptr<transwarp::executor> executor) = 0;
333  virtual void set_executor_all(std::shared_ptr<transwarp::executor> executor) = 0;
334  virtual void remove_executor() = 0;
335  virtual void remove_executor_all() = 0;
336  virtual void set_priority(std::size_t priority) = 0;
337  virtual void set_priority_all(std::size_t priority) = 0;
338  virtual void reset_priority() = 0;
339  virtual void reset_priority_all() = 0;
340  virtual void set_custom_data(std::shared_ptr<void> custom_data) = 0;
341  virtual void set_custom_data_all(std::shared_ptr<void> custom_data) = 0;
342  virtual void remove_custom_data() = 0;
343  virtual void remove_custom_data_all() = 0;
344  virtual const std::shared_ptr<transwarp::node>& get_node() const noexcept = 0;
345  virtual void add_listener(std::shared_ptr<transwarp::listener> listener) = 0;
346  virtual void add_listener(transwarp::event_type event, std::shared_ptr<transwarp::listener> listener) = 0;
347  virtual void remove_listener(const std::shared_ptr<transwarp::listener>& listener) = 0;
348  virtual void remove_listener(transwarp::event_type event, const std::shared_ptr<transwarp::listener>& listener) = 0;
349  virtual void remove_listeners(transwarp::event_type event) = 0;
350  virtual void remove_listeners() = 0;
351  virtual void schedule() = 0;
352  virtual void schedule(transwarp::executor& executor) = 0;
353  virtual void schedule(bool reset) = 0;
354  virtual void schedule(transwarp::executor& executor, bool reset) = 0;
355  virtual void schedule_all() = 0;
356  virtual void schedule_all(transwarp::executor& executor) = 0;
357  virtual void schedule_all(bool reset_all) = 0;
358  virtual void schedule_all(transwarp::executor& executor, bool reset_all) = 0;
359  virtual void schedule_all(transwarp::schedule_type type) = 0;
360  virtual void schedule_all(transwarp::executor& executor, transwarp::schedule_type type) = 0;
361  virtual void schedule_all(transwarp::schedule_type type, bool reset_all) = 0;
362  virtual void schedule_all(transwarp::executor& executor, transwarp::schedule_type type, bool reset_all) = 0;
363  virtual void set_exception(std::exception_ptr exception) = 0;
364  virtual bool was_scheduled() const noexcept = 0;
365  virtual void wait() const = 0;
366  virtual bool is_ready() const = 0;
367  virtual bool has_result() const = 0;
368  virtual void reset() = 0;
369  virtual void reset_all() = 0;
370  virtual void cancel(bool enabled) noexcept = 0;
371  virtual void cancel_all(bool enabled) noexcept = 0;
372  virtual std::vector<transwarp::edge> get_graph() const = 0;
373 
374 protected:
375  virtual void schedule_impl(bool reset, transwarp::executor* executor=nullptr) = 0;
376 
377 private:
378  friend struct transwarp::detail::visit_depth;
379  friend struct transwarp::detail::unvisit;
380  friend struct transwarp::detail::final_visitor;
382 
383  virtual void visit_depth(const std::function<void(itask&)>& visitor) = 0;
384  virtual void unvisit() noexcept = 0;
385  virtual void set_node_id(std::size_t id) noexcept = 0;
386 };
387 
388 
389 /// Removes reference and const from a type
390 template<typename T>
391 struct decay {
392  using type = typename std::remove_const<typename std::remove_reference<T>::type>::type;
393 };
394 
395 
396 /// Returns the result type of a std::shared_future<T>
397 template<typename T>
398 struct result {
399  using type = typename std::result_of<decltype(&std::shared_future<T>::get)(std::shared_future<T>)>::type;
400 };
401 
402 
403 /// The task class
404 template<typename ResultType>
405 class task : public transwarp::itask {
406 public:
407  using result_type = ResultType;
408 
409  virtual ~task() = default;
410 
411  virtual void set_value(const typename transwarp::decay<result_type>::type& value) = 0;
412  virtual void set_value(typename transwarp::decay<result_type>::type&& value) = 0;
413  virtual const std::shared_future<result_type>& get_future() const noexcept = 0;
414  virtual typename transwarp::result<result_type>::type get() const = 0;
415 };
416 
417 /// The task class (reference result type)
418 template<typename ResultType>
419 class task<ResultType&> : public transwarp::itask {
420 public:
421  using result_type = ResultType&;
422 
423  virtual ~task() = default;
424 
425  virtual void set_value(typename transwarp::decay<result_type>::type& value) = 0;
426  virtual const std::shared_future<result_type>& get_future() const noexcept = 0;
427  virtual typename transwarp::result<result_type>::type get() const = 0;
428 };
429 
430 /// The task class (void result type)
431 template<>
432 class task<void> : public transwarp::itask {
433 public:
434  using result_type = void;
435 
436  virtual ~task() = default;
437 
438  virtual void set_value() = 0;
439  virtual const std::shared_future<result_type>& get_future() const noexcept = 0;
440  virtual result_type get() const = 0;
441 };
442 
443 
444 /// A base class for a user-defined functor that needs access to the node associated
445 /// to the task or a cancel point to stop a task while it's running
446 class functor {
447 public:
448 
449  virtual ~functor() = default;
450 
451 protected:
452 
453  /// The node associated to the task
454  const std::shared_ptr<transwarp::node>& transwarp_node() const noexcept {
455  return transwarp_node_;
456  }
457 
458  /// If the associated task is canceled then this will throw transwarp::task_canceled
459  /// which will stop the task while it's running
460  void transwarp_cancel_point() const {
461  if (transwarp_node_->is_canceled()) {
462  throw transwarp::task_canceled(std::to_string(transwarp_node_->get_id()));
463  }
464  }
465 
466 private:
467  template<bool>
469 
470  std::shared_ptr<transwarp::node> transwarp_node_;
471 };
472 
473 
474 /// Detail namespace for internal functionality only
475 namespace detail {
476 
477 
478 /// Node manipulation
479 struct node_manip {
480 
481  static void set_id(transwarp::node& node, std::size_t id) noexcept {
482  node.id_ = id;
483  }
484 
485  static void set_level(transwarp::node& node, std::size_t level) noexcept {
486  node.level_ = level;
487  }
488 
489  static void set_type(transwarp::node& node, transwarp::task_type type) noexcept {
490  node.type_ = type;
491  }
492 
493  static void set_name(transwarp::node& node, std::shared_ptr<std::string> name) noexcept {
494  node.name_ = name;
495  }
496 
497  static void set_executor(transwarp::node& node, std::shared_ptr<std::string> executor) noexcept {
498  if (executor) {
499  node.executor_ = std::move(executor);
500  } else {
501  node.executor_.reset();
502  }
503  }
504 
505  static void add_parent(transwarp::node& node, std::shared_ptr<transwarp::node> parent) {
506  node.parents_.push_back(std::move(parent));
507  }
508 
509  static void set_priority(transwarp::node& node, std::size_t priority) noexcept {
510  node.priority_ = priority;
511  }
512 
513  static void set_custom_data(transwarp::node& node, std::shared_ptr<void> custom_data) {
514  if (custom_data) {
515  node.custom_data_ = std::move(custom_data);
516  } else {
517  node.custom_data_.reset();
518  }
519  }
520 
521  static void set_canceled(transwarp::node& node, bool enabled) noexcept {
522  node.canceled_ = enabled;
523  }
524 
525 };
526 
527 
528 /// A simple thread pool used to execute tasks in parallel
529 class thread_pool {
530 public:
531 
532  explicit thread_pool(std::size_t n_threads)
533  : done_(false)
534  {
535  if (n_threads == 0) {
536  throw transwarp::invalid_parameter("number of threads");
537  }
538  const std::size_t n_target = threads_.size() + n_threads;
539  while (threads_.size() < n_target) {
540  std::thread thread;
541  try {
542  thread = std::thread(&thread_pool::worker, this);
543  } catch (...) {
544  shutdown();
545  throw;
546  }
547  try {
548  threads_.push_back(std::move(thread));
549  } catch (...) {
550  shutdown();
551  thread.join();
552  throw;
553  }
554  }
555  }
556 
557  // delete copy/move semantics
558  thread_pool(const thread_pool&) = delete;
559  thread_pool& operator=(const thread_pool&) = delete;
560  thread_pool(thread_pool&&) = delete;
561  thread_pool& operator=(thread_pool&&) = delete;
562 
563  ~thread_pool() {
564  shutdown();
565  }
566 
567  void push(const std::function<void()>& functor) {
568  {
569  std::lock_guard<std::mutex> lock(mutex_);
570  functors_.push(functor);
571  }
572  cond_var_.notify_one();
573  }
574 
575 private:
576 
577  void worker() {
578  for (;;) {
579  std::function<void()> functor;
580  {
581  std::unique_lock<std::mutex> lock(mutex_);
582  cond_var_.wait(lock, [this]{
583  return done_ || !functors_.empty();
584  });
585  if (done_ && functors_.empty()) {
586  break;
587  }
588  functor = functors_.front();
589  functors_.pop();
590  }
591  functor();
592  }
593  }
594 
595  void shutdown() {
596  {
597  std::lock_guard<std::mutex> lock(mutex_);
598  done_ = true;
599  }
600  cond_var_.notify_all();
601  for (std::thread& thread : threads_) {
602  thread.join();
603  }
604  threads_.clear();
605  }
606 
607  bool done_;
608  std::vector<std::thread> threads_;
609  std::queue<std::function<void()>> functors_;
610  std::condition_variable cond_var_;
611  std::mutex mutex_;
612 };
613 
614 
615 template<int offset, typename... ParentResults>
617  static void work(const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& source, std::tuple<std::shared_future<ParentResults>...>& target) {
618  std::get<offset>(target) = std::get<offset>(source)->get_future();
620  }
621 };
622 
623 template<typename... ParentResults>
624 struct assign_futures_impl<-1, ParentResults...> {
625  static void work(const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>&, std::tuple<std::shared_future<ParentResults>...>&) {}
626 };
627 
628 /// Returns the futures from the given tuple of tasks
629 template<typename... ParentResults>
630 std::tuple<std::shared_future<ParentResults>...> get_futures(const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& input) {
631  std::tuple<std::shared_future<ParentResults>...> result;
632  assign_futures_impl<static_cast<int>(sizeof...(ParentResults)) - 1, ParentResults...>::work(input, result);
633  return result;
634 }
635 
636 /// Returns the futures from the given vector of tasks
637 template<typename ParentResultType>
638 std::vector<std::shared_future<ParentResultType>> get_futures(const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& input) {
639  std::vector<std::shared_future<ParentResultType>> result;
640  result.reserve(input.size());
641  for (const std::shared_ptr<transwarp::task<ParentResultType>>& task : input) {
642  result.emplace_back(task->get_future());
643  }
644  return result;
645 }
646 
647 
648 /// Runs the task with the given arguments, hence, invoking the task's functor
649 template<typename Result, typename Task, typename... Args>
650 Result run_task(std::size_t node_id, const std::weak_ptr<Task>& task, Args&&... args) {
651  const std::shared_ptr<Task> t = task.lock();
652  if (!t) {
653  throw transwarp::task_destroyed(std::to_string(node_id));
654  }
655  if (t->node_->is_canceled()) {
656  throw transwarp::task_canceled(std::to_string(node_id));
657  }
659  return t->functor_(std::forward<Args>(args)...);
660 }
661 
662 
663 inline void wait_for_all() {}
664 
665 /// Waits for all parents to finish
666 template<typename ParentResult, typename... ParentResults>
667 void wait_for_all(const std::shared_ptr<transwarp::task<ParentResult>>& parent, const std::shared_ptr<transwarp::task<ParentResults>>& ...parents) {
668  parent->get_future().wait();
670 }
671 
672 
673 /// Waits for all parents to finish
674 template<typename ParentResultType>
675 void wait_for_all(const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
676  for (const std::shared_ptr<transwarp::task<ParentResultType>>& parent : parents) {
677  parent->get_future().wait();
678  }
679 }
680 
681 
682 template<typename Parent>
683 Parent wait_for_any_impl() {
684  return {};
685 }
686 
687 template<typename Parent, typename ParentResult, typename... ParentResults>
688 Parent wait_for_any_impl(const std::shared_ptr<transwarp::task<ParentResult>>& parent, const std::shared_ptr<transwarp::task<ParentResults>>& ...parents) {
689  const std::future_status status = parent->get_future().wait_for(std::chrono::microseconds(1));
690  if (status == std::future_status::ready) {
691  return parent;
692  }
693  return transwarp::detail::wait_for_any_impl<Parent>(parents...);
694 }
695 
696 /// Waits for the first parent to finish
697 template<typename Parent, typename... ParentResults>
698 Parent wait_for_any(const std::shared_ptr<transwarp::task<ParentResults>>& ...parents) {
699  for (;;) {
700  Parent parent = transwarp::detail::wait_for_any_impl<Parent>(parents...);
701  if (parent) {
702  return parent;
703  }
704  }
705 }
706 
707 
708 /// Waits for the first parent to finish
709 template<typename ParentResultType>
710 std::shared_ptr<transwarp::task<ParentResultType>> wait_for_any(const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
711  for (;;) {
712  for (const std::shared_ptr<transwarp::task<ParentResultType>>& parent : parents) {
713  const std::future_status status = parent->get_future().wait_for(std::chrono::microseconds(1));
714  if (status == std::future_status::ready) {
715  return parent;
716  }
717  }
718  }
719 }
720 
721 
722 template<typename OneResult>
723 void cancel_all_but_one(const std::shared_ptr<transwarp::task<OneResult>>&) {}
724 
725 /// Cancels all tasks but one
726 template<typename OneResult, typename ParentResult, typename... ParentResults>
727 void cancel_all_but_one(const std::shared_ptr<transwarp::task<OneResult>>& one, const std::shared_ptr<transwarp::task<ParentResult>>& parent, const std::shared_ptr<transwarp::task<ParentResults>>& ...parents) {
728  if (one != parent) {
729  parent->cancel(true);
730  }
732 }
733 
734 
735 /// Cancels all tasks but one
736 template<typename OneResult, typename ParentResultType>
737 void cancel_all_but_one(const std::shared_ptr<transwarp::task<OneResult>>& one, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
738  for (const std::shared_ptr<transwarp::task<ParentResultType>>& parent : parents) {
739  if (one != parent) {
740  parent->cancel(true);
741  }
742  }
743 }
744 
745 
746 template<typename TaskType, bool done, int total, int... n>
747 struct call_impl {
748  template<typename Result, typename Task, typename... ParentResults>
749  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
750  return call_impl<TaskType, total == 1 + static_cast<int>(sizeof...(n)), total, n..., static_cast<int>(sizeof...(n))>::template
751  work<Result>(node_id, task, parents);
752  }
753 };
754 
755 template<typename TaskType>
757 
758 template<int total, int... n>
759 struct call_impl<transwarp::root_type, true, total, n...> {
760  template<typename Result, typename Task, typename... ParentResults>
761  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>&) {
762  return transwarp::detail::run_task<Result>(node_id, task);
763  }
764 };
765 
766 template<>
767 struct call_impl_vector<transwarp::root_type> {
768  template<typename Result, typename Task, typename ParentResultType>
769  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>&) {
770  return transwarp::detail::run_task<Result>(node_id, task);
771  }
772 };
773 
774 template<int total, int... n>
775 struct call_impl<transwarp::accept_type, true, total, n...> {
776  template<typename Result, typename Task, typename... ParentResults>
777  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
778  transwarp::detail::wait_for_all(std::get<n>(parents)...);
779  const std::tuple<std::shared_future<ParentResults>...> futures = transwarp::detail::get_futures(parents);
780  return transwarp::detail::run_task<Result>(node_id, task, std::get<n>(futures)...);
781  }
782 };
783 
784 template<>
785 struct call_impl_vector<transwarp::accept_type> {
786  template<typename Result, typename Task, typename ParentResultType>
787  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
789  return transwarp::detail::run_task<Result>(node_id, task, transwarp::detail::get_futures(parents));
790  }
791 };
792 
793 template<int total, int... n>
794 struct call_impl<transwarp::accept_any_type, true, total, n...> {
795  template<typename Result, typename Task, typename... ParentResults>
796  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
797  using parent_t = typename std::remove_reference<decltype(std::get<0>(parents))>::type; // Use first type as reference
798  parent_t parent = transwarp::detail::wait_for_any<parent_t>(std::get<n>(parents)...);
799  transwarp::detail::cancel_all_but_one(parent, std::get<n>(parents)...);
800  return transwarp::detail::run_task<Result>(node_id, task, parent->get_future());
801  }
802 };
803 
804 template<>
805 struct call_impl_vector<transwarp::accept_any_type> {
806  template<typename Result, typename Task, typename ParentResultType>
807  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
808  std::shared_ptr<transwarp::task<ParentResultType>> parent = transwarp::detail::wait_for_any(parents);
810  return transwarp::detail::run_task<Result>(node_id, task, parent->get_future());
811  }
812 };
813 
814 template<int total, int... n>
815 struct call_impl<transwarp::consume_type, true, total, n...> {
816  template<typename Result, typename Task, typename... ParentResults>
817  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
818  transwarp::detail::wait_for_all(std::get<n>(parents)...);
819  return transwarp::detail::run_task<Result>(node_id, task, std::get<n>(parents)->get_future().get()...);
820  }
821 };
822 
823 template<>
824 struct call_impl_vector<transwarp::consume_type> {
825  template<typename Result, typename Task, typename ParentResultType>
826  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
828  std::vector<ParentResultType> results;
829  results.reserve(parents.size());
830  for (const std::shared_ptr<transwarp::task<ParentResultType>>& parent : parents) {
831  results.emplace_back(parent->get_future().get());
832  }
833  return transwarp::detail::run_task<Result>(node_id, task, std::move(results));
834  }
835 };
836 
837 template<int total, int... n>
838 struct call_impl<transwarp::consume_any_type, true, total, n...> {
839  template<typename Result, typename Task, typename... ParentResults>
840  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
841  using parent_t = typename std::remove_reference<decltype(std::get<0>(parents))>::type; /// Use first type as reference
842  parent_t parent = transwarp::detail::wait_for_any<parent_t>(std::get<n>(parents)...);
843  transwarp::detail::cancel_all_but_one(parent, std::get<n>(parents)...);
844  return transwarp::detail::run_task<Result>(node_id, task, parent->get_future().get());
845  }
846 };
847 
848 template<>
849 struct call_impl_vector<transwarp::consume_any_type> {
850  template<typename Result, typename Task, typename ParentResultType>
851  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
852  std::shared_ptr<transwarp::task<ParentResultType>> parent = transwarp::detail::wait_for_any(parents);
854  return transwarp::detail::run_task<Result>(node_id, task, parent->get_future().get());
855  }
856 };
857 
858 template<int total, int... n>
859 struct call_impl<transwarp::wait_type, true, total, n...> {
860  template<typename Result, typename Task, typename... ParentResults>
861  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
862  transwarp::detail::wait_for_all(std::get<n>(parents)...);
863  get_all(std::get<n>(parents)...); // Ensures that exceptions are propagated
864  return transwarp::detail::run_task<Result>(node_id, task);
865  }
866  template<typename T, typename... Args>
867  static void get_all(const T& arg, const Args& ...args) {
868  arg->get_future().get();
869  get_all(args...);
870  }
871  static void get_all() {}
872 };
873 
874 template<>
875 struct call_impl_vector<transwarp::wait_type> {
876  template<typename Result, typename Task, typename ParentResultType>
877  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
879  for (const std::shared_ptr<transwarp::task<ParentResultType>>& parent : parents) {
880  parent->get_future().get(); // Ensures that exceptions are propagated
881  }
882  return transwarp::detail::run_task<Result>(node_id, task);
883  }
884 };
885 
886 template<int total, int... n>
887 struct call_impl<transwarp::wait_any_type, true, total, n...> {
888  template<typename Result, typename Task, typename... ParentResults>
889  static Result work(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
890  using parent_t = typename std::remove_reference<decltype(std::get<0>(parents))>::type; // Use first type as reference
891  parent_t parent = transwarp::detail::wait_for_any<parent_t>(std::get<n>(parents)...);
892  transwarp::detail::cancel_all_but_one(parent, std::get<n>(parents)...);
893  parent->get_future().get(); // Ensures that exceptions are propagated
894  return transwarp::detail::run_task<Result>(node_id, task);
895  }
896 };
897 
898 template<>
899 struct call_impl_vector<transwarp::wait_any_type> {
900  template<typename Result, typename Task, typename ParentResultType>
901  static Result work(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
902  std::shared_ptr<transwarp::task<ParentResultType>> parent = transwarp::detail::wait_for_any(parents);
904  parent->get_future().get(); // Ensures that exceptions are propagated
905  return transwarp::detail::run_task<Result>(node_id, task);
906  }
907 };
908 
909 /// Calls the functor of the given task with the results from the tuple of parents.
910 /// Throws transwarp::task_canceled if the task is canceled.
911 /// Throws transwarp::task_destroyed in case the task was destroyed prematurely.
912 template<typename TaskType, typename Result, typename Task, typename... ParentResults>
913 Result call(std::size_t node_id, const Task& task, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& parents) {
914  constexpr std::size_t n = std::tuple_size<std::tuple<std::shared_future<ParentResults>...>>::value;
916  work<Result>(node_id, task, parents);
917 }
918 
919 /// Calls the functor of the given task with the results from the vector of parents.
920 /// Throws transwarp::task_canceled if the task is canceled.
921 /// Throws transwarp::task_destroyed in case the task was destroyed prematurely.
922 template<typename TaskType, typename Result, typename Task, typename ParentResultType>
923 Result call(std::size_t node_id, const Task& task, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& parents) {
925  work<Result>(node_id, task, parents);
926 }
927 
928 template<std::size_t...> struct indices {};
929 
930 template<std::size_t...> struct construct_range;
931 
932 template<std::size_t end, std::size_t idx, std::size_t... i>
933 struct construct_range<end, idx, i...> : construct_range<end, idx + 1, i..., idx> {};
934 
935 template<std::size_t end, std::size_t... i>
936 struct construct_range<end, end, i...> {
937  using type = transwarp::detail::indices<i...>;
938 };
939 
940 template<std::size_t b, std::size_t e>
941 struct index_range {
943 };
944 
945 template<typename Functor, typename... ParentResults>
946 void call_with_each_index(transwarp::detail::indices<>, const Functor&, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>&) {}
947 
948 template<std::size_t i, std::size_t... j, typename Functor, typename... ParentResults>
949 void call_with_each_index(transwarp::detail::indices<i, j...>, const Functor& f, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& t) {
950  auto ptr = std::get<i>(t);
951  if (!ptr) {
952  throw transwarp::invalid_parameter("task pointer");
953  }
954  f(*ptr);
955  transwarp::detail::call_with_each_index(transwarp::detail::indices<j...>(), f, t);
956 }
957 
958 /// Calls the functor with every element in the tuple
959 template<typename Functor, typename... ParentResults>
960 void call_with_each(const Functor& f, const std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>& t) {
961  constexpr std::size_t n = std::tuple_size<std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>>::value;
962  using index_t = typename transwarp::detail::index_range<0, n>::type;
963  transwarp::detail::call_with_each_index(index_t(), f, t);
964 }
965 
966 /// Calls the functor with every element in the vector
967 template<typename Functor, typename ParentResultType>
968 void call_with_each(const Functor& f, const std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>& v) {
969  for (const std::shared_ptr<transwarp::task<ParentResultType>>& ptr : v) {
970  if (!ptr) {
971  throw transwarp::invalid_parameter("task pointer");
972  }
973  f(*ptr);
974  }
975 }
976 
977 
978 /// Sets parents and level of the node
980  explicit parent_visitor(transwarp::node& node) noexcept
981  : node_(node) {}
982 
983  void operator()(const transwarp::itask& task) const {
984  transwarp::detail::node_manip::add_parent(node_, task.get_node());
985  if (node_.get_level() <= task.get_node()->get_level()) {
986  /// A child's level is always larger than any of its parents' levels
987  transwarp::detail::node_manip::set_level(node_, task.get_node()->get_level() + 1);
988  }
989  }
990 
991  transwarp::node& node_;
992 };
993 
994 /// Applies final bookkeeping to the task
996  final_visitor() noexcept
997  : id_(0) {}
998 
999  void operator()(transwarp::itask& task) noexcept {
1000  task.set_node_id(id_++);
1001  }
1002 
1003  std::size_t id_;
1004 };
1005 
1006 /// Generates a graph
1008  explicit graph_visitor(std::vector<transwarp::edge>& graph) noexcept
1009  : graph_(graph) {}
1010 
1011  void operator()(const transwarp::itask& task) {
1012  const std::shared_ptr<transwarp::node>& node = task.get_node();
1013  for (const std::shared_ptr<transwarp::node>& parent : node->get_parents()) {
1014  graph_.emplace_back(parent, node);
1015  }
1016  }
1017 
1018  std::vector<transwarp::edge>& graph_;
1019 };
1020 
1021 /// Schedules using the given executor
1023  schedule_visitor(bool reset, transwarp::executor* executor) noexcept
1024  : reset_(reset), executor_(executor) {}
1025 
1026  void operator()(transwarp::itask& task) {
1027  task.schedule_impl(reset_, executor_);
1028  }
1029 
1030  bool reset_;
1031  transwarp::executor* executor_;
1032 };
1033 
1034 /// Resets the given task
1036 
1037  void operator()(transwarp::itask& task) const {
1038  task.reset();
1039  }
1040 };
1041 
1042 /// Cancels or resumes the given task
1044  explicit cancel_visitor(bool enabled) noexcept
1045  : enabled_(enabled) {}
1046 
1047  void operator()(transwarp::itask& task) const noexcept {
1048  task.cancel(enabled_);
1049  }
1050 
1051  bool enabled_;
1052 };
1053 
1054 /// Assigns an executor to the given task
1056  explicit set_executor_visitor(std::shared_ptr<transwarp::executor> executor) noexcept
1057  : executor_(std::move(executor)) {}
1058 
1059  void operator()(transwarp::itask& task) const noexcept {
1060  task.set_executor(executor_);
1061  }
1062 
1063  std::shared_ptr<transwarp::executor> executor_;
1064 };
1065 
1066 /// Removes the executor from the given task
1068 
1069  void operator()(transwarp::itask& task) const noexcept {
1070  task.remove_executor();
1071  }
1072 };
1073 
1074 /// Assigns a priority to the given task
1076  explicit set_priority_visitor(std::size_t priority) noexcept
1077  : priority_(priority) {}
1078 
1079  void operator()(transwarp::itask& task) const noexcept {
1080  task.set_priority(priority_);
1081  }
1082 
1083  std::size_t priority_;
1084 };
1085 
1086 /// Resets the priority of the given task
1088 
1089  void operator()(transwarp::itask& task) const noexcept {
1090  task.reset_priority();
1091  }
1092 };
1093 
1094 /// Assigns custom data to the given task
1096  explicit set_custom_data_visitor(std::shared_ptr<void> custom_data) noexcept
1097  : custom_data_(std::move(custom_data)) {}
1098 
1099  void operator()(transwarp::itask& task) const noexcept {
1100  task.set_custom_data(custom_data_);
1101  }
1102 
1103  std::shared_ptr<void> custom_data_;
1104 };
1105 
1106 /// Removes custom data from the given task
1108 
1109  void operator()(transwarp::itask& task) const noexcept {
1110  task.remove_custom_data();
1111  }
1112 };
1113 
1114 /// Pushes the given task into the vector of tasks
1116  explicit push_task_visitor(std::vector<transwarp::itask*>& tasks)
1117  : tasks_(tasks) {}
1118 
1119  void operator()(transwarp::itask& task) {
1120  tasks_.push_back(&task);
1121  }
1122 
1123  std::vector<transwarp::itask*>& tasks_;
1124 };
1125 
1126 /// Visits the given task using the visitor given in the constructor
1127 struct visit_depth {
1128  explicit visit_depth(const std::function<void(transwarp::itask&)>& visitor) noexcept
1129  : visitor_(visitor) {}
1130 
1131  void operator()(transwarp::itask& task) const {
1132  task.visit_depth(visitor_);
1133  }
1134 
1135  const std::function<void(transwarp::itask&)>& visitor_;
1136 };
1137 
1138 /// Unvisits the given task
1139 struct unvisit {
1140 
1141  void operator()(transwarp::itask& task) const noexcept {
1142  task.unvisit();
1143  }
1144 };
1145 
1146 /// Determines the result type of the Functor dispatching on the task type
1147 template<typename TaskType, typename Functor, typename... ParentResults>
1149  static_assert(std::is_same<TaskType, transwarp::root_type>::value ||
1150  std::is_same<TaskType, transwarp::accept_type>::value ||
1151  std::is_same<TaskType, transwarp::accept_any_type>::value ||
1152  std::is_same<TaskType, transwarp::consume_type>::value ||
1153  std::is_same<TaskType, transwarp::consume_any_type>::value ||
1154  std::is_same<TaskType, transwarp::wait_type>::value ||
1155  std::is_same<TaskType, transwarp::wait_any_type>::value,
1156  "Invalid task type, must be one of: root, accept, accept_any, consume, consume_any, wait, wait_any");
1157 };
1158 
1159 template<typename Functor, typename... ParentResults>
1160 struct functor_result<transwarp::root_type, Functor, ParentResults...> {
1161  static_assert(sizeof...(ParentResults) == 0, "A root task cannot have parent tasks");
1162  using type = decltype(std::declval<Functor>()());
1163 };
1164 
1165 template<typename Functor, typename... ParentResults>
1166 struct functor_result<transwarp::accept_type, Functor, ParentResults...> {
1167  static_assert(sizeof...(ParentResults) > 0, "An accept task must have at least one parent");
1168  using type = decltype(std::declval<Functor>()(std::declval<std::shared_future<ParentResults>>()...));
1169 };
1170 
1171 template<typename Functor, typename ParentResultType>
1172 struct functor_result<transwarp::accept_type, Functor, std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1173  using type = decltype(std::declval<Functor>()(std::declval<std::vector<std::shared_future<ParentResultType>>>()));
1174 };
1175 
1176 template<typename Functor, typename... ParentResults>
1177 struct functor_result<transwarp::accept_any_type, Functor, ParentResults...> {
1178  static_assert(sizeof...(ParentResults) > 0, "An accept_any task must have at least one parent");
1179  using arg_t = typename std::tuple_element<0, std::tuple<ParentResults...>>::type; // Using first type as reference
1180  using type = decltype(std::declval<Functor>()(std::declval<std::shared_future<arg_t>>()));
1181 };
1182 
1183 template<typename Functor, typename ParentResultType>
1184 struct functor_result<transwarp::accept_any_type, Functor, std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1185  using type = decltype(std::declval<Functor>()(std::declval<std::shared_future<ParentResultType>>()));
1186 };
1187 
1188 template<typename Functor, typename... ParentResults>
1189 struct functor_result<transwarp::consume_type, Functor, ParentResults...> {
1190  static_assert(sizeof...(ParentResults) > 0, "A consume task must have at least one parent");
1191  using type = decltype(std::declval<Functor>()(std::declval<ParentResults>()...));
1192 };
1193 
1194 template<typename Functor, typename ParentResultType>
1195 struct functor_result<transwarp::consume_type, Functor, std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1196  using type = decltype(std::declval<Functor>()(std::declval<std::vector<ParentResultType>>()));
1197 };
1198 
1199 template<typename Functor, typename... ParentResults>
1200 struct functor_result<transwarp::consume_any_type, Functor, ParentResults...> {
1201  static_assert(sizeof...(ParentResults) > 0, "A consume_any task must have at least one parent");
1202  using arg_t = typename std::tuple_element<0, std::tuple<ParentResults...>>::type; // Using first type as reference
1203  using type = decltype(std::declval<Functor>()(std::declval<arg_t>()));
1204 };
1205 
1206 template<typename Functor, typename ParentResultType>
1207 struct functor_result<transwarp::consume_any_type, Functor, std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1208  using type = decltype(std::declval<Functor>()(std::declval<ParentResultType>()));
1209 };
1210 
1211 template<typename Functor, typename... ParentResults>
1212 struct functor_result<transwarp::wait_type, Functor, ParentResults...> {
1213  static_assert(sizeof...(ParentResults) > 0, "A wait task must have at least one parent");
1214  using type = decltype(std::declval<Functor>()());
1215 };
1216 
1217 template<typename Functor, typename ParentResultType>
1218 struct functor_result<transwarp::wait_type, Functor, std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1219  using type = decltype(std::declval<Functor>()());
1220 };
1221 
1222 template<typename Functor, typename... ParentResults>
1223 struct functor_result<transwarp::wait_any_type, Functor, ParentResults...> {
1224  static_assert(sizeof...(ParentResults) > 0, "A wait_any task must have at least one parent");
1225  using type = decltype(std::declval<Functor>()());
1226 };
1227 
1228 template<typename Functor, typename ParentResultType>
1229 struct functor_result<transwarp::wait_any_type, Functor, std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1230  using type = decltype(std::declval<Functor>()());
1231 };
1232 
1233 template<bool is_transwarp_functor>
1234 struct assign_node_if_impl;
1235 
1236 template<>
1237 struct assign_node_if_impl<true> {
1238  template<typename Functor>
1239  void operator()(Functor& functor, std::shared_ptr<transwarp::node> node) const noexcept {
1240  functor.transwarp_node_ = std::move(node);
1241  }
1242 };
1243 
1244 template<>
1245 struct assign_node_if_impl<false> {
1246  template<typename Functor>
1247  void operator()(Functor&, std::shared_ptr<transwarp::node>) const noexcept {}
1248 };
1249 
1250 /// Assigns the node to the given functor if the functor is a subclass of transwarp::functor
1251 template<typename Functor>
1252 void assign_node_if(Functor& functor, std::shared_ptr<transwarp::node> node) noexcept {
1254 }
1255 
1256 /// Returns a ready future with the given value as its state
1257 template<typename ResultType, typename Value>
1258 std::shared_future<ResultType> make_future_with_value(Value&& value) {
1259  std::promise<ResultType> promise;
1260  promise.set_value(std::forward<Value>(value));
1261  return promise.get_future();
1262 }
1263 
1264 /// Returns a ready future
1265 inline std::shared_future<void> make_ready_future() {
1266  std::promise<void> promise;
1267  promise.set_value();
1268  return promise.get_future();
1269 }
1270 
1271 /// Returns a ready future with the given exception as its state
1272 template<typename ResultType>
1273 std::shared_future<ResultType> make_future_with_exception(std::exception_ptr exception) {
1274  if (!exception) {
1275  throw transwarp::invalid_parameter("exception pointer");
1276  }
1277  std::promise<ResultType> promise;
1278  promise.set_exception(exception);
1279  return promise.get_future();
1280 }
1281 
1282 
1283 /// Determines the type of the parents
1284 template<typename... ParentResults>
1285 struct parents {
1286  using type = std::tuple<std::shared_ptr<transwarp::task<ParentResults>>...>;
1287 };
1288 
1289 /// Determines the type of the parents. Specialization for vector parents
1290 template<typename ParentResultType>
1291 struct parents<std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>> {
1292  using type = std::vector<std::shared_ptr<transwarp::task<ParentResultType>>>;
1293 };
1294 
1295 
1296 template<typename ResultType, typename TaskType>
1298 protected:
1299 
1300  template<typename Task, typename Parents>
1301  void call(std::size_t node_id,
1302  const std::weak_ptr<Task>& task,
1303  const Parents& parents) {
1304  promise_.set_value(transwarp::detail::call<TaskType, ResultType>(node_id, task, parents));
1305  }
1306 
1307  std::promise<ResultType> promise_;
1308 };
1309 
1310 template<typename TaskType>
1311 class base_runner<void, TaskType> {
1312 protected:
1313 
1314  template<typename Task, typename Parents>
1315  void call(std::size_t node_id,
1316  const std::weak_ptr<Task>& task,
1317  const Parents& parents) {
1318  transwarp::detail::call<TaskType, void>(node_id, task, parents);
1319  promise_.set_value();
1320  }
1321 
1322  std::promise<void> promise_;
1323 };
1324 
1325 /// A callable to run a task given its parents
1326 template<typename ResultType, typename TaskType, typename Task, typename Parents>
1327 class runner : public transwarp::detail::base_runner<ResultType, TaskType> {
1328 public:
1329 
1330  runner(std::size_t node_id,
1331  const std::weak_ptr<Task>& task,
1332  const typename transwarp::decay<Parents>::type& parents)
1333  : node_id_(node_id),
1334  task_(task),
1335  parents_(parents)
1336  {}
1337 
1338  std::future<ResultType> get_future() {
1339  return this->promise_.get_future();
1340  }
1341 
1342  void operator()() {
1343  if (const std::shared_ptr<Task> t = task_.lock()) {
1344  t->raise_event(transwarp::event_type::before_started);
1345  }
1346  try {
1347  this->call(node_id_, task_, parents_);
1348  } catch (const transwarp::task_canceled&) {
1349  this->promise_.set_exception(std::current_exception());
1350  if (const std::shared_ptr<Task> t = task_.lock()) {
1351  t->raise_event(transwarp::event_type::after_canceled);
1352  }
1353  } catch (...) {
1354  this->promise_.set_exception(std::current_exception());
1355  }
1356  if (const std::shared_ptr<Task> t = task_.lock()) {
1357  t->raise_event(transwarp::event_type::after_finished);
1358  }
1359  }
1360 
1361 private:
1362  const std::size_t node_id_;
1363  const std::weak_ptr<Task> task_;
1364  const typename transwarp::decay<Parents>::type parents_;
1365 };
1366 
1367 
1368 } // detail
1369 
1370 
1371 /// Executor for sequential execution. Runs functors sequentially on the same thread
1373 public:
1374 
1375  sequential() = default;
1376 
1377  // delete copy/move semantics
1378  sequential(const sequential&) = delete;
1379  sequential& operator=(const sequential&) = delete;
1380  sequential(sequential&&) = delete;
1381  sequential& operator=(sequential&&) = delete;
1382 
1383  /// Returns the name of the executor
1384  std::string get_name() const override {
1385  return "transwarp::sequential";
1386  }
1387 
1388  /// Runs the functor on the current thread
1389  void execute(const std::function<void()>& functor, const std::shared_ptr<transwarp::node>&) override {
1390  functor();
1391  }
1392 };
1393 
1394 
1395 /// Executor for parallel execution. Uses a simple thread pool
1397 public:
1398 
1399  explicit parallel(std::size_t n_threads)
1400  : pool_(n_threads)
1401  {}
1402 
1403  // delete copy/move semantics
1404  parallel(const parallel&) = delete;
1405  parallel& operator=(const parallel&) = delete;
1406  parallel(parallel&&) = delete;
1407  parallel& operator=(parallel&&) = delete;
1408 
1409  /// Returns the name of the executor
1410  std::string get_name() const override {
1411  return "transwarp::parallel";
1412  }
1413 
1414  /// Pushes the functor into the thread pool for asynchronous execution
1415  void execute(const std::function<void()>& functor, const std::shared_ptr<transwarp::node>&) override {
1416  pool_.push(functor);
1417  }
1418 
1419 private:
1421 };
1422 
1423 
1424 /// Detail namespace for internal functionality only
1425 namespace detail {
1426 
1427 /// The base task class that contains the functionality that can be used
1428 /// with all result types (void and non-void).
1429 template<typename ResultType, typename TaskType, typename Functor, typename... ParentResults>
1430 class task_impl_base : public transwarp::task<ResultType>,
1431  public std::enable_shared_from_this<task_impl_base<ResultType, TaskType, Functor, ParentResults...>> {
1432 public:
1433  /// The task type
1434  using task_type = TaskType;
1435 
1436  /// The result type of this task
1437  using result_type = ResultType;
1438 
1439  /// Assigns an executor to this task which takes precedence over
1440  /// the executor provided in schedule() or schedule_all()
1441  void set_executor(std::shared_ptr<transwarp::executor> executor) override {
1443  if (!executor) {
1444  throw transwarp::invalid_parameter("executor pointer");
1445  }
1446  executor_ = std::move(executor);
1447  transwarp::detail::node_manip::set_executor(*node_, std::shared_ptr<std::string>(new std::string(executor_->get_name())));
1448  }
1449 
1450  /// Assigns an executor to all tasks which takes precedence over
1451  /// the executor provided in schedule() or schedule_all()
1452  void set_executor_all(std::shared_ptr<transwarp::executor> executor) override {
1454  transwarp::detail::set_executor_visitor visitor(std::move(executor));
1455  visit_depth_all(visitor);
1456  }
1457 
1458  /// Removes the executor from this task
1459  void remove_executor() override {
1461  executor_.reset();
1462  transwarp::detail::node_manip::set_executor(*node_, nullptr);
1463  }
1464 
1465  /// Removes the executor from all tasks
1466  void remove_executor_all() override {
1469  visit_depth_all(visitor);
1470  }
1471 
1472  /// Sets a task priority (defaults to 0). transwarp will not directly use this.
1473  /// This is only useful if something else is using the priority (e.g. a custom executor)
1474  void set_priority(std::size_t priority) override {
1476  transwarp::detail::node_manip::set_priority(*node_, priority);
1477  }
1478 
1479  /// Sets a priority to all tasks (defaults to 0). transwarp will not directly use this.
1480  /// This is only useful if something else is using the priority (e.g. a custom executor)
1481  void set_priority_all(std::size_t priority) override {
1483  transwarp::detail::set_priority_visitor visitor(priority);
1484  visit_depth_all(visitor);
1485  }
1486 
1487  /// Resets the task priority to 0
1488  void reset_priority() override {
1490  transwarp::detail::node_manip::set_priority(*node_, 0);
1491  }
1492 
1493  /// Resets the priority of all tasks to 0
1494  void reset_priority_all() override {
1497  visit_depth_all(visitor);
1498  }
1499 
1500  /// Assigns custom data to this task. transwarp will not directly use this.
1501  /// This is only useful if something else is using this custom data (e.g. a custom executor)
1502  void set_custom_data(std::shared_ptr<void> custom_data) override {
1504  if (!custom_data) {
1505  throw transwarp::invalid_parameter("custom data pointer");
1506  }
1507  transwarp::detail::node_manip::set_custom_data(*node_, std::move(custom_data));
1508  }
1509 
1510  /// Assigns custom data to all tasks. transwarp will not directly use this.
1511  /// This is only useful if something else is using this custom data (e.g. a custom executor)
1512  void set_custom_data_all(std::shared_ptr<void> custom_data) override {
1514  transwarp::detail::set_custom_data_visitor visitor(std::move(custom_data));
1515  visit_depth_all(visitor);
1516  }
1517 
1518  /// Removes custom data from this task
1519  void remove_custom_data() override {
1521  transwarp::detail::node_manip::set_custom_data(*node_, nullptr);
1522  }
1523 
1524  /// Removes custom data from all tasks
1525  void remove_custom_data_all() override {
1528  visit_depth_all(visitor);
1529  }
1530 
1531  /// Returns the future associated to the underlying execution
1532  const std::shared_future<result_type>& get_future() const noexcept override {
1533  return future_;
1534  }
1535 
1536  /// Returns the associated node
1537  const std::shared_ptr<transwarp::node>& get_node() const noexcept override {
1538  return node_;
1539  }
1540 
1541  /// Adds a new listener for all event types
1542  void add_listener(std::shared_ptr<transwarp::listener> listener) override {
1544  check_listener(listener);
1545  for (std::vector<std::shared_ptr<transwarp::listener>>& l : listeners_) {
1546  l.push_back(listener);
1547  }
1548  }
1549 
1550  /// Adds a new listener for the given event type only
1551  void add_listener(transwarp::event_type event, std::shared_ptr<transwarp::listener> listener) override {
1553  check_listener(listener);
1554  listeners_[get_event_index(event)].push_back(std::move(listener));
1555  }
1556 
1557  /// Removes the listener for all event types
1558  void remove_listener(const std::shared_ptr<transwarp::listener>& listener) override {
1560  check_listener(listener);
1561  for (std::vector<std::shared_ptr<transwarp::listener>>& l : listeners_) {
1562  l.erase(std::remove(l.begin(), l.end(), listener), l.end());
1563  }
1564  }
1565 
1566  /// Removes the listener for the given event type only
1567  void remove_listener(transwarp::event_type event, const std::shared_ptr<transwarp::listener>& listener) override {
1569  check_listener(listener);
1570  std::vector<std::shared_ptr<transwarp::listener>>& l = listeners_[get_event_index(event)];
1571  l.erase(std::remove(l.begin(), l.end(), listener), l.end());
1572  }
1573 
1574  /// Removes all listeners for the given event type
1577  listeners_[get_event_index(event)].clear();
1578  }
1579 
1580  /// Removes all listeners
1581  void remove_listeners() override {
1583  for (std::vector<std::shared_ptr<transwarp::listener>>& l : listeners_) {
1584  l.clear();
1585  }
1586  }
1587 
1588  /// Schedules this task for execution on the caller thread.
1589  /// The task-specific executor gets precedence if it exists.
1590  /// This overload will reset the underlying future.
1591  void schedule() override {
1593  this->schedule_impl(true);
1594  }
1595 
1596  /// Schedules this task for execution on the caller thread.
1597  /// The task-specific executor gets precedence if it exists.
1598  /// reset denotes whether schedule should reset the underlying
1599  /// future and schedule even if the future is already valid.
1600  void schedule(bool reset) override {
1602  this->schedule_impl(reset);
1603  }
1604 
1605  /// Schedules this task for execution using the provided executor.
1606  /// The task-specific executor gets precedence if it exists.
1607  /// This overload will reset the underlying future.
1610  this->schedule_impl(true, &executor);
1611  }
1612 
1613  /// Schedules this task for execution using the provided executor.
1614  /// The task-specific executor gets precedence if it exists.
1615  /// reset denotes whether schedule should reset the underlying
1616  /// future and schedule even if the future is already valid.
1617  void schedule(transwarp::executor& executor, bool reset) override {
1619  this->schedule_impl(reset, &executor);
1620  }
1621 
1622  /// Schedules all tasks in the graph for execution on the caller thread.
1623  /// The task-specific executors get precedence if they exist.
1624  /// This overload will reset the underlying futures.
1625  void schedule_all() override {
1627  schedule_all_impl(true, transwarp::schedule_type::breadth);
1628  }
1629 
1630  /// Schedules all tasks in the graph for execution using the provided executor.
1631  /// The task-specific executors get precedence if they exist.
1632  /// This overload will reset the underlying futures.
1635  schedule_all_impl(true, transwarp::schedule_type::breadth, &executor);
1636  }
1637 
1638  /// Schedules all tasks in the graph for execution on the caller thread.
1639  /// The task-specific executors get precedence if they exist.
1640  /// reset_all denotes whether schedule_all should reset the underlying
1641  /// futures and schedule even if the futures are already present.
1642  void schedule_all(bool reset_all) override {
1644  schedule_all_impl(reset_all, transwarp::schedule_type::breadth);
1645  }
1646 
1647  /// Schedules all tasks in the graph for execution using the provided executor.
1648  /// The task-specific executors get precedence if they exist.
1649  /// reset_all denotes whether schedule_all should reset the underlying
1650  /// futures and schedule even if the futures are already present.
1653  schedule_all_impl(reset_all, transwarp::schedule_type::breadth, &executor);
1654  }
1655 
1656  /// Schedules all tasks in the graph for execution on the caller thread.
1657  /// The task-specific executors get precedence if they exist.
1658  /// This overload will reset the underlying futures.
1661  schedule_all_impl(true, type);
1662  }
1663 
1664  /// Schedules all tasks in the graph for execution using the provided executor.
1665  /// The task-specific executors get precedence if they exist.
1666  /// This overload will reset the underlying futures.
1669  schedule_all_impl(true, type, &executor);
1670  }
1671 
1672  /// Schedules all tasks in the graph for execution on the caller thread.
1673  /// The task-specific executors get precedence if they exist.
1674  /// reset_all denotes whether schedule_all should reset the underlying
1675  /// futures and schedule even if the futures are already present.
1678  schedule_all_impl(reset_all, type);
1679  }
1680 
1681  /// Schedules all tasks in the graph for execution using the provided executor.
1682  /// The task-specific executors get precedence if they exist.
1683  /// reset_all denotes whether schedule_all should reset the underlying
1684  /// futures and schedule even if the futures are already present.
1687  schedule_all_impl(reset_all, type, &executor);
1688  }
1689 
1690  /// Assigns an exception to this task. Scheduling will have no effect after an exception
1691  /// has been set. Calling reset() will remove the exception and re-enable scheduling.
1692  void set_exception(std::exception_ptr exception) override {
1694  future_ = transwarp::detail::make_future_with_exception<result_type>(exception);
1695  schedule_mode_ = false;
1696  }
1697 
1698  /// Returns whether the task was scheduled and not reset afterwards.
1699  /// This means that the underlying future is valid
1700  bool was_scheduled() const noexcept override {
1701  return future_.valid();
1702  }
1703 
1704  /// Waits for the task to complete. Should only be called if was_scheduled()
1705  /// is true, throws transwarp::control_error otherwise
1706  void wait() const override {
1708  future_.wait();
1709  }
1710 
1711  /// Returns whether the task has finished processing. Should only be called
1712  /// if was_scheduled() is true, throws transwarp::control_error otherwise
1713  bool is_ready() const override {
1715  return future_.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
1716  }
1717 
1718  /// Returns whether this task contains a result
1719  bool has_result() const noexcept override {
1720  return was_scheduled() && is_ready();
1721  }
1722 
1723  /// Resets this task
1724  void reset() override {
1726  future_ = std::shared_future<result_type>();
1727  transwarp::detail::node_manip::set_canceled(*node_, false);
1728  schedule_mode_ = true;
1729  }
1730 
1731  /// Resets all tasks in the graph
1732  void reset_all() override {
1735  visit_depth_all(visitor);
1736  }
1737 
1738  /// If enabled then this task is canceled which will
1739  /// throw transwarp::task_canceled when retrieving the task result.
1740  /// Passing false is equivalent to resume.
1741  void cancel(bool enabled) noexcept override {
1742  transwarp::detail::node_manip::set_canceled(*node_, enabled);
1743  }
1744 
1745  /// If enabled then all pending tasks in the graph are canceled which will
1746  /// throw transwarp::task_canceled when retrieving the task result.
1747  /// Passing false is equivalent to resume.
1748  void cancel_all(bool enabled) noexcept override {
1749  transwarp::detail::cancel_visitor visitor(enabled);
1750  visit_depth_all(visitor);
1751  }
1752 
1753  /// Returns the graph of the task structure. This is mainly for visualizing
1754  /// the tasks and their interdependencies. Pass the result into transwarp::to_string
1755  /// to retrieve a dot-style graph representation for easy viewing.
1756  std::vector<transwarp::edge> get_graph() const override {
1757  std::vector<transwarp::edge> graph;
1758  transwarp::detail::graph_visitor visitor(graph);
1759  const_cast<task_impl_base*>(this)->visit_depth_all(visitor);
1760  return graph;
1761  }
1762 
1763 protected:
1764 
1765  template<typename F>
1766  // cppcheck-suppress passedByValue
1767  task_impl_base(bool has_name, std::string name, F&& functor, std::shared_ptr<transwarp::task<ParentResults>>... parents)
1768  : node_(new transwarp::node),
1769  functor_(std::forward<F>(functor)),
1770  parents_(std::move(parents)...)
1771  {
1772  init(has_name, std::move(name));
1773  }
1774 
1775  template<typename F, typename P>
1776  // cppcheck-suppress passedByValue
1777  task_impl_base(bool has_name, std::string name, F&& functor, std::vector<std::shared_ptr<transwarp::task<P>>> parents)
1778  : node_(new transwarp::node),
1779  functor_(std::forward<F>(functor)),
1780  parents_(std::move(parents))
1781  {
1782  if (parents_.empty()) {
1783  throw transwarp::invalid_parameter("parents are empty");
1784  }
1785  init(has_name, std::move(name));
1786  }
1787 
1788  void init(bool has_name, std::string name) {
1789  transwarp::detail::node_manip::set_type(*node_, task_type::value);
1790  transwarp::detail::node_manip::set_name(*node_, (has_name ? std::shared_ptr<std::string>(new std::string(std::move(name))) : nullptr));
1791  transwarp::detail::assign_node_if(functor_, node_);
1794  visit_depth(visitor);
1795  unvisit();
1796  }
1797 
1798  /// Checks if the task is currently running and throws transwarp::control_error if it is
1800  if (future_.valid() && future_.wait_for(std::chrono::seconds(0)) != std::future_status::ready) {
1801  throw transwarp::control_error("task currently running: " + transwarp::to_string(*node_, " "));
1802  }
1803  }
1804 
1805  /// Checks if the task was scheduled and throws transwarp::control_error if it's not
1807  if (!future_.valid()) {
1808  throw transwarp::control_error("task was not scheduled: " + transwarp::to_string(*node_, " "));
1809  }
1810  }
1811 
1812  bool schedule_mode_ = true;
1813  std::shared_future<result_type> future_;
1814 
1815 private:
1816 
1817  template<typename R, typename Y, typename T, typename P>
1818  friend class transwarp::detail::runner;
1819 
1820  template<typename R, typename T, typename... A>
1821  friend R transwarp::detail::run_task(std::size_t, const std::weak_ptr<T>&, A&&...);
1822 
1823  /// Assigns the given id to the node
1824  void set_node_id(std::size_t id) noexcept override {
1825  transwarp::detail::node_manip::set_id(*node_, id);
1826  }
1827 
1828  /// Schedules this task for execution using the provided executor.
1829  /// The task-specific executor gets precedence if it exists.
1830  /// Runs the task on the same thread as the caller if neither the global
1831  /// nor the task-specific executor is found.
1832  void schedule_impl(bool reset, transwarp::executor* executor=nullptr) override {
1833  if (schedule_mode_ && (reset || !future_.valid())) {
1834  if (reset) {
1835  transwarp::detail::node_manip::set_canceled(*node_, false);
1836  }
1837  std::weak_ptr<task_impl_base> self = this->shared_from_this();
1839  std::shared_ptr<runner_t> runner = std::shared_ptr<runner_t>(new runner_t(node_->get_id(), self, parents_));
1841  future_ = runner->get_future();
1842  if (executor_) {
1843  executor_->execute([runner]{ (*runner)(); }, node_);
1844  } else if (executor) {
1845  executor->execute([runner]{ (*runner)(); }, node_);
1846  } else {
1847  (*runner)();
1848  }
1849  }
1850  }
1851 
1852  /// Schedules all tasks in the graph for execution using the provided executor.
1853  /// The task-specific executors get precedence if they exist.
1854  /// Runs tasks on the same thread as the caller if neither the global
1855  /// nor a task-specific executor is found.
1856  void schedule_all_impl(bool reset_all, transwarp::schedule_type type, transwarp::executor* executor=nullptr) {
1857  transwarp::detail::schedule_visitor visitor(reset_all, executor);
1858  switch (type) {
1860  visit_breadth_all(visitor);
1861  break;
1863  visit_depth_all(visitor);
1864  break;
1865  default:
1866  throw transwarp::invalid_parameter("schedule type");
1867  }
1868  }
1869 
1870  /// Visits all tasks in a breadth-first traversal.
1871  template<typename Visitor>
1872  void visit_breadth_all(Visitor& visitor) {
1873  if (breadth_tasks_.empty()) {
1874  breadth_tasks_.reserve(node_->get_id() + 1);
1875  visit_depth(transwarp::detail::push_task_visitor(breadth_tasks_));
1876  unvisit();
1877  auto compare = [](const transwarp::itask* const l, const transwarp::itask* const r) {
1878  const std::size_t l_level = l->get_node()->get_level();
1879  const std::size_t l_id = l->get_node()->get_id();
1880  const std::size_t r_level = r->get_node()->get_level();
1881  const std::size_t r_id = r->get_node()->get_id();
1882  return std::tie(l_level, l_id) < std::tie(r_level, r_id);
1883  };
1884  std::sort(breadth_tasks_.begin(), breadth_tasks_.end(), compare);
1885  }
1886  for (transwarp::itask* task : breadth_tasks_) {
1887  visitor(*task);
1888  }
1889  }
1890 
1891  /// Visits all tasks in a depth-first traversal.
1892  template<typename Visitor>
1893  void visit_depth_all(Visitor& visitor) {
1894  if (depth_tasks_.empty()) {
1895  depth_tasks_.reserve(node_->get_id() + 1);
1896  visit_depth(transwarp::detail::push_task_visitor(depth_tasks_));
1897  unvisit();
1898  }
1899  for (transwarp::itask* task : depth_tasks_) {
1900  visitor(*task);
1901  }
1902  }
1903 
1904  /// Visits each task in a depth-first traversal.
1905  void visit_depth(const std::function<void(transwarp::itask&)>& visitor) override {
1906  if (!visited_) {
1908  visitor(*this);
1909  visited_ = true;
1910  }
1911  }
1912 
1913  /// Traverses through each task and marks them as not visited.
1914  void unvisit() noexcept override {
1915  if (visited_) {
1916  visited_ = false;
1918  }
1919  }
1920 
1921  /// Returns the index for a given event type
1922  std::size_t get_event_index(transwarp::event_type event) const {
1923  const std::size_t index = static_cast<std::size_t>(event);
1924  if (index >= static_cast<std::size_t>(transwarp::event_type::count)) {
1925  throw transwarp::invalid_parameter("event type");
1926  }
1927  return index;
1928  }
1929 
1930  /// Raises the given event to all listeners
1931  void raise_event(transwarp::event_type event) const {
1932  for (const std::shared_ptr<transwarp::listener>& listener : listeners_[static_cast<std::size_t>(event)]) {
1933  listener->handle_event(event, node_);
1934  }
1935  }
1936 
1937  /// Check for non-null listener pointer
1938  void check_listener(const std::shared_ptr<transwarp::listener>& listener) const {
1939  if (!listener) {
1940  throw transwarp::invalid_parameter("listener pointer");
1941  }
1942  }
1943 
1944  std::shared_ptr<transwarp::node> node_;
1945  Functor functor_;
1946  typename transwarp::detail::parents<ParentResults...>::type parents_;
1947  bool visited_ = false;
1948  std::shared_ptr<transwarp::executor> executor_;
1949  std::vector<std::shared_ptr<transwarp::listener>> listeners_[static_cast<std::size_t>(transwarp::event_type::count)];
1950  std::vector<transwarp::itask*> depth_tasks_;
1951  std::vector<transwarp::itask*> breadth_tasks_;
1952 };
1953 
1954 
1955 /// A task proxy
1956 template<typename ResultType, typename TaskType, typename Functor, typename... ParentResults>
1957 class task_impl_proxy : public transwarp::detail::task_impl_base<ResultType, TaskType, Functor, ParentResults...> {
1958 public:
1959  /// The task type
1960  using task_type = TaskType;
1961 
1962  /// The result type of this task
1963  using result_type = ResultType;
1964 
1965  /// Assigns a value to this task. Scheduling will have no effect after a value
1966  /// has been set. Calling reset() will remove the value and re-enable scheduling.
1967  void set_value(const typename transwarp::decay<result_type>::type& value) override {
1968  this->ensure_task_not_running();
1969  this->future_ = transwarp::detail::make_future_with_value<result_type>(value);
1970  this->schedule_mode_ = false;
1971  }
1972 
1973  /// Assigns a value to this task. Scheduling will have no effect after a value
1974  /// has been set. Calling reset() will remove the value and re-enable scheduling.
1975  void set_value(typename transwarp::decay<result_type>::type&& value) override {
1976  this->ensure_task_not_running();
1977  this->future_ = transwarp::detail::make_future_with_value<result_type>(std::move(value));
1978  this->schedule_mode_ = false;
1979  }
1980 
1981  /// Returns the result of this task. Throws any exceptions that the underlying
1982  /// functor throws. Should only be called if was_scheduled() is true,
1983  /// throws transwarp::control_error otherwise
1984  typename transwarp::result<result_type>::type get() const override {
1985  this->ensure_task_was_scheduled();
1986  return this->future_.get();
1987  }
1988 
1989 protected:
1990 
1991  template<typename F>
1992  // cppcheck-suppress passedByValue
1993  // cppcheck-suppress uninitMemberVar
1994  task_impl_proxy(bool has_name, std::string name, F&& functor, std::shared_ptr<transwarp::task<ParentResults>>... parents)
1995  : transwarp::detail::task_impl_base<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents)...)
1996  {}
1997 
1998  template<typename F, typename P>
1999  // cppcheck-suppress passedByValue
2000  // cppcheck-suppress uninitMemberVar
2001  task_impl_proxy(bool has_name, std::string name, F&& functor, std::vector<std::shared_ptr<transwarp::task<P>>> parents)
2002  : transwarp::detail::task_impl_base<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents))
2003  {}
2004 
2005 };
2006 
2007 /// A task proxy for reference result type.
2008 template<typename ResultType, typename TaskType, typename Functor, typename... ParentResults>
2009 class task_impl_proxy<ResultType&, TaskType, Functor, ParentResults...> : public transwarp::detail::task_impl_base<ResultType&, TaskType, Functor, ParentResults...> {
2010 public:
2011  /// The task type
2012  using task_type = TaskType;
2013 
2014  /// The result type of this task
2015  using result_type = ResultType&;
2016 
2017  /// Assigns a value to this task. Scheduling will have no effect after a value
2018  /// has been set. Calling reset() will remove the value and re-enable scheduling.
2019  void set_value(typename transwarp::decay<result_type>::type& value) override {
2020  this->ensure_task_not_running();
2021  this->future_ = transwarp::detail::make_future_with_value<result_type>(value);
2022  this->schedule_mode_ = false;
2023  }
2024 
2025  /// Returns the result of this task. Throws any exceptions that the underlying
2026  /// functor throws. Should only be called if was_scheduled() is true,
2027  /// throws transwarp::control_error otherwise
2028  typename transwarp::result<result_type>::type get() const override {
2029  this->ensure_task_was_scheduled();
2030  return this->future_.get();
2031  }
2032 
2033 protected:
2034 
2035  template<typename F>
2036  // cppcheck-suppress passedByValue
2037  // cppcheck-suppress uninitMemberVar
2038  task_impl_proxy(bool has_name, std::string name, F&& functor, std::shared_ptr<transwarp::task<ParentResults>>... parents)
2039  : transwarp::detail::task_impl_base<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents)...)
2040  {}
2041 
2042  template<typename F, typename P>
2043  // cppcheck-suppress passedByValue
2044  // cppcheck-suppress uninitMemberVar
2045  task_impl_proxy(bool has_name, std::string name, F&& functor, std::vector<std::shared_ptr<transwarp::task<P>>> parents)
2046  : transwarp::detail::task_impl_base<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents))
2047  {}
2048 
2049 };
2050 
2051 /// A task proxy for void result type.
2052 template<typename TaskType, typename Functor, typename... ParentResults>
2053 class task_impl_proxy<void, TaskType, Functor, ParentResults...> : public transwarp::detail::task_impl_base<void, TaskType, Functor, ParentResults...> {
2054 public:
2055  /// The task type
2056  using task_type = TaskType;
2057 
2058  /// The result type of this task
2059  using result_type = void;
2060 
2061  /// Assigns a value to this task. Scheduling will have no effect after a call
2062  /// to this. Calling reset() will reset this and re-enable scheduling.
2063  void set_value() override {
2064  this->ensure_task_not_running();
2065  this->future_ = transwarp::detail::make_ready_future();
2066  this->schedule_mode_ = false;
2067  }
2068 
2069  /// Blocks until the task finishes. Throws any exceptions that the underlying
2070  /// functor throws. Should only be called if was_scheduled() is true,
2071  /// throws transwarp::control_error otherwise
2072  void get() const override {
2073  this->ensure_task_was_scheduled();
2074  this->future_.get();
2075  }
2076 
2077 protected:
2078 
2079  template<typename F>
2080  // cppcheck-suppress passedByValue
2081  // cppcheck-suppress uninitMemberVar
2082  task_impl_proxy(bool has_name, std::string name, F&& functor, std::shared_ptr<transwarp::task<ParentResults>>... parents)
2083  : transwarp::detail::task_impl_base<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents)...)
2084  {}
2085 
2086  template<typename F, typename P>
2087  // cppcheck-suppress passedByValue
2088  // cppcheck-suppress uninitMemberVar
2089  task_impl_proxy(bool has_name, std::string name, F&& functor, std::vector<std::shared_ptr<transwarp::task<P>>> parents)
2090  : transwarp::detail::task_impl_base<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents))
2091  {}
2092 
2093 };
2094 
2095 } // detail
2096 
2097 
2098 /// A task representing a piece of work given by functor and parent tasks.
2099 /// By connecting tasks a directed acyclic graph is built.
2100 /// Tasks should be created using the make_task factory functions.
2101 template<typename TaskType, typename Functor, typename... ParentResults>
2102 class task_impl : public transwarp::detail::task_impl_proxy<typename transwarp::detail::functor_result<TaskType, Functor, ParentResults...>::type, TaskType, Functor, ParentResults...> {
2103 public:
2104  /// The task type
2105  using task_type = TaskType;
2106 
2107  /// The result type of this task
2108  using result_type = typename transwarp::detail::functor_result<TaskType, Functor, ParentResults...>::type;
2109 
2110  /// A task is defined by name, functor, and parent tasks.
2111  /// Note: Don't use this constructor directly, use transwarp::make_task
2112  template<typename F>
2113  // cppcheck-suppress passedByValue
2114  // cppcheck-suppress uninitMemberVar
2115  task_impl(bool has_name, std::string name, F&& functor, std::shared_ptr<transwarp::task<ParentResults>>... parents)
2116  : transwarp::detail::task_impl_proxy<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents)...)
2117  {}
2118 
2119  /// A task is defined by name, functor, and parent tasks.
2120  /// Note: Don't use this constructor directly, use transwarp::make_task
2121  template<typename F, typename P>
2122  // cppcheck-suppress uninitMemberVar
2123  task_impl(bool has_name, std::string name, F&& functor, std::vector<std::shared_ptr<transwarp::task<P>>> parents)
2124  : transwarp::detail::task_impl_proxy<result_type, task_type, Functor, ParentResults...>(has_name, std::move(name), std::forward<F>(functor), std::move(parents))
2125  {}
2126 
2127  // delete copy/move semantics
2128  task_impl(const task_impl&) = delete;
2129  task_impl& operator=(const task_impl&) = delete;
2130  task_impl(task_impl&&) = delete;
2131  task_impl& operator=(task_impl&&) = delete;
2132 
2133  /// Creates a continuation to this task
2134  template<typename TaskType_, typename Functor_>
2135  std::shared_ptr<transwarp::task_impl<TaskType_, typename std::decay<Functor_>::type, result_type>>
2136  then(TaskType_, std::string name, Functor_&& functor) const {
2138  return std::shared_ptr<task_t>(new task_t(true, std::move(name), std::forward<Functor_>(functor), std::dynamic_pointer_cast<transwarp::task<result_type>>(const_cast<task_impl*>(this)->shared_from_this())));
2139  }
2140 
2141  /// Creates a continuation to this task. Overload for omitting for task name
2142  template<typename TaskType_, typename Functor_>
2143  std::shared_ptr<transwarp::task_impl<TaskType_, typename std::decay<Functor_>::type, result_type>>
2144  then(TaskType_, Functor_&& functor) const {
2146  return std::shared_ptr<task_t>(new task_t(false, "", std::forward<Functor_>(functor), std::dynamic_pointer_cast<transwarp::task<result_type>>(const_cast<task_impl*>(this)->shared_from_this())));
2147  }
2148 
2149 };
2150 
2151 
2152 /// A value task that stores a single value and doesn't require scheduling.
2153 /// Value tasks should be created using the make_value_task factory functions.
2154 template<typename ResultType>
2155 class value_task : public transwarp::task<ResultType>,
2156  public std::enable_shared_from_this<value_task<ResultType>> {
2157 public:
2158  /// The task type
2160 
2161  /// The result type of this task
2162  using result_type = ResultType;
2163 
2164  /// A value task is defined by name and value.
2165  /// Note: Don't use this constructor directly, use transwarp::make_value_task
2166  template<typename T>
2167  // cppcheck-suppress passedByValue
2168  // cppcheck-suppress uninitMemberVar
2169  value_task(bool has_name, std::string name, T&& value)
2170  : node_(new transwarp::node),
2171  future_(transwarp::detail::make_future_with_value<result_type>(std::forward<T>(value)))
2172  {
2173  transwarp::detail::node_manip::set_type(*node_, task_type::value);
2174  transwarp::detail::node_manip::set_name(*node_, (has_name ? std::shared_ptr<std::string>(new std::string(std::move(name))) : nullptr));
2175  }
2176 
2177  // delete copy/move semantics
2178  value_task(const value_task&) = delete;
2179  value_task& operator=(const value_task&) = delete;
2180  value_task(value_task&&) = delete;
2181  value_task& operator=(value_task&&) = delete;
2182 
2183  /// Creates a continuation to this task
2184  template<typename TaskType_, typename Functor_>
2185  std::shared_ptr<transwarp::task_impl<TaskType_, typename std::decay<Functor_>::type, result_type>>
2186  then(TaskType_, std::string name, Functor_&& functor) const {
2188  return std::shared_ptr<task_t>(new task_t(true, std::move(name), std::forward<Functor_>(functor), std::dynamic_pointer_cast<transwarp::task<result_type>>(const_cast<value_task*>(this)->shared_from_this())));
2189  }
2190 
2191  /// Creates a continuation to this task. Overload for omitting the task name
2192  template<typename TaskType_, typename Functor_>
2193  std::shared_ptr<transwarp::task_impl<TaskType_, typename std::decay<Functor_>::type, result_type>>
2194  then(TaskType_, Functor_&& functor) const {
2196  return std::shared_ptr<task_t>(new task_t(false, "", std::forward<Functor_>(functor), std::dynamic_pointer_cast<transwarp::task<result_type>>(const_cast<value_task*>(this)->shared_from_this())));
2197  }
2198 
2199  /// No-op because a value task never runs
2200  void set_executor(std::shared_ptr<transwarp::executor>) override {}
2201 
2202  /// No-op because a value task never runs and doesn't have parents
2203  void set_executor_all(std::shared_ptr<transwarp::executor>) override {}
2204 
2205  /// No-op because a value task never runs
2206  void remove_executor() override {}
2207 
2208  /// No-op because a value task never runs and doesn't have parents
2209  void remove_executor_all() override {}
2210 
2211  /// Sets a task priority (defaults to 0). transwarp will not directly use this.
2212  /// This is only useful if something else is using the priority
2213  void set_priority(std::size_t priority) override {
2214  transwarp::detail::node_manip::set_priority(*node_, priority);
2215  }
2216 
2217  /// Sets a priority to all tasks (defaults to 0). transwarp will not directly use this.
2218  /// This is only useful if something else is using the priority
2219  void set_priority_all(std::size_t priority) override {
2220  set_priority(priority);
2221  }
2222 
2223  /// Resets the task priority to 0
2224  void reset_priority() override {
2225  transwarp::detail::node_manip::set_priority(*node_, 0);
2226  }
2227 
2228  /// Resets the priority of all tasks to 0
2229  void reset_priority_all() override {
2230  reset_priority();
2231  }
2232 
2233  /// Assigns custom data to this task. transwarp will not directly use this.
2234  /// This is only useful if something else is using this custom data
2235  void set_custom_data(std::shared_ptr<void> custom_data) override {
2236  if (!custom_data) {
2237  throw transwarp::invalid_parameter("custom data pointer");
2238  }
2239  transwarp::detail::node_manip::set_custom_data(*node_, std::move(custom_data));
2240  }
2241 
2242  /// Assigns custom data to all tasks. transwarp will not directly use this.
2243  /// This is only useful if something else is using this custom data
2244  void set_custom_data_all(std::shared_ptr<void> custom_data) override {
2245  set_custom_data(std::move(custom_data));
2246  }
2247 
2248  /// Removes custom data from this task
2249  void remove_custom_data() override {
2250  transwarp::detail::node_manip::set_custom_data(*node_, nullptr);
2251  }
2252 
2253  /// Removes custom data from all tasks
2254  void remove_custom_data_all() override {
2256  }
2257 
2258  /// Returns the future associated to the underlying execution
2259  const std::shared_future<result_type>& get_future() const noexcept override {
2260  return future_;
2261  }
2262 
2263  /// Returns the associated node
2264  const std::shared_ptr<transwarp::node>& get_node() const noexcept override {
2265  return node_;
2266  }
2267 
2268  /// No-op because a value task doesn't raise events
2269  void add_listener(std::shared_ptr<transwarp::listener>) override {}
2270 
2271  /// No-op because a value task doesn't raise events
2272  void add_listener(transwarp::event_type, std::shared_ptr<transwarp::listener>) override {}
2273 
2274  /// No-op because a value task doesn't raise events
2275  void remove_listener(const std::shared_ptr<transwarp::listener>&) override {}
2276 
2277  /// No-op because a value task doesn't raise events
2278  void remove_listener(transwarp::event_type, const std::shared_ptr<transwarp::listener>&) override {}
2279 
2280  /// No-op because a value task doesn't raise events
2282 
2283  /// No-op because a value task doesn't raise events
2284  void remove_listeners() override {}
2285 
2286  /// No-op because a value task never runs
2287  void schedule() override {}
2288 
2289  /// No-op because a value task never runs
2290  void schedule(transwarp::executor&) override {}
2291 
2292  /// No-op because a value task never runs
2293  void schedule(bool) override {}
2294 
2295  /// No-op because a value task never runs
2296  void schedule(transwarp::executor&, bool) override {}
2297 
2298  /// No-op because a value task never runs and doesn't have parents
2299  void schedule_all() override {}
2300 
2301  /// No-op because a value task never runs and doesn't have parents
2303 
2304  /// No-op because a value task never runs and doesn't have parents
2305  void schedule_all(bool) override {}
2306 
2307  /// No-op because a value task never runs and doesn't have parents
2308  void schedule_all(transwarp::executor&, bool) override {}
2309 
2310  /// No-op because a value task never runs and doesn't have parents
2312 
2313  /// No-op because a value task never runs and doesn't have parents
2315 
2316  /// No-op because a value task never runs and doesn't have parents
2317  void schedule_all(transwarp::schedule_type, bool) override {}
2318 
2319  /// No-op because a value task never runs and doesn't have parents
2321 
2322  /// Assigns a value to this task
2323  void set_value(const typename transwarp::decay<result_type>::type& value) override {
2324  future_ = transwarp::detail::make_future_with_value<result_type>(value);
2325  }
2326 
2327  /// Assigns a value to this task
2328  void set_value(typename transwarp::decay<result_type>::type&& value) override {
2329  future_ = transwarp::detail::make_future_with_value<result_type>(std::move(value));
2330  };
2331 
2332  /// Assigns an exception to this task
2333  void set_exception(std::exception_ptr exception) override {
2334  future_ = transwarp::detail::make_future_with_exception<result_type>(exception);
2335  }
2336 
2337  /// Returns true because a value task is scheduled once on construction
2338  bool was_scheduled() const noexcept override {
2339  return true;
2340  }
2341 
2342  /// No-op because a value task never runs
2343  void wait() const override {}
2344 
2345  /// Returns true because a value task is always ready
2346  bool is_ready() const override {
2347  return true;
2348  }
2349 
2350  /// Returns true because a value task always contains a result
2351  bool has_result() const noexcept override {
2352  return true;
2353  }
2354 
2355  /// Returns the result of this task
2356  typename transwarp::result<result_type>::type get() const override {
2357  return future_.get();
2358  }
2359 
2360  /// No-op because a value task never runs
2361  void reset() override {}
2362 
2363  /// No-op because a value task never runs and doesn't have parents
2364  void reset_all() override {}
2365 
2366  /// No-op because a value task never runs
2367  void cancel(bool) noexcept override {}
2368 
2369  /// No-op because a value task never runs and doesn't have parents
2370  void cancel_all(bool) noexcept override {}
2371 
2372  /// Returns an empty graph because a value task doesn't have parents
2373  std::vector<transwarp::edge> get_graph() const override {
2374  return {};
2375  }
2376 
2377 private:
2378 
2379  /// Assigns the given id to the node
2380  void set_node_id(std::size_t id) noexcept override {
2381  transwarp::detail::node_manip::set_id(*node_, id);
2382  }
2383 
2384  /// No-op because a value task never runs
2385  void schedule_impl(bool, transwarp::executor*) override {}
2386 
2387  /// Visits this task
2388  void visit_depth(const std::function<void(transwarp::itask&)>& visitor) override {
2389  if (!visited_) {
2390  visitor(*this);
2391  visited_ = true;
2392  }
2393  }
2394 
2395  /// Marks this task as not visited
2396  void unvisit() noexcept override {
2397  visited_ = false;
2398  }
2399 
2400  std::shared_ptr<transwarp::node> node_;
2401  std::shared_future<result_type> future_;
2402  bool visited_ = false;
2403 };
2404 
2405 
2406 /// A factory function to create a new task
2407 template<typename TaskType, typename Functor, typename... Parents>
2408 std::shared_ptr<transwarp::task_impl<TaskType, typename std::decay<Functor>::type, typename Parents::result_type...>>
2409 make_task(TaskType, std::string name, Functor&& functor, std::shared_ptr<Parents>... parents) {
2410  using task_t = transwarp::task_impl<TaskType, typename std::decay<Functor>::type, typename Parents::result_type...>;
2411  return std::shared_ptr<task_t>(new task_t(true, std::move(name), std::forward<Functor>(functor), std::move(parents)...));
2412 }
2413 
2414 /// A factory function to create a new task. Overload for omitting the task name
2415 template<typename TaskType, typename Functor, typename... Parents>
2416 std::shared_ptr<transwarp::task_impl<TaskType, typename std::decay<Functor>::type, typename Parents::result_type...>>
2417 make_task(TaskType, Functor&& functor, std::shared_ptr<Parents>... parents) {
2418  using task_t = transwarp::task_impl<TaskType, typename std::decay<Functor>::type, typename Parents::result_type...>;
2419  return std::shared_ptr<task_t>(new task_t(false, "", std::forward<Functor>(functor), std::move(parents)...));
2420 }
2421 
2422 
2423 /// A factory function to create a new task with vector parents
2424 template<typename TaskType, typename Functor, typename ParentType>
2425 std::shared_ptr<transwarp::task_impl<TaskType, typename std::decay<Functor>::type, std::vector<ParentType>>>
2426 make_task(TaskType, std::string name, Functor&& functor, std::vector<ParentType> parents) {
2427  using task_t = transwarp::task_impl<TaskType, typename std::decay<Functor>::type, std::vector<ParentType>>;
2428  return std::shared_ptr<task_t>(new task_t(true, std::move(name), std::forward<Functor>(functor), std::move(parents)));
2429 }
2430 
2431 /// A factory function to create a new task with vector parents. Overload for omitting the task name
2432 template<typename TaskType, typename Functor, typename ParentType>
2433 std::shared_ptr<transwarp::task_impl<TaskType, typename std::decay<Functor>::type, std::vector<ParentType>>>
2434 make_task(TaskType, Functor&& functor, std::vector<ParentType> parents) {
2435  using task_t = transwarp::task_impl<TaskType, typename std::decay<Functor>::type, std::vector<ParentType>>;
2436  return std::shared_ptr<task_t>(new task_t(false, "", std::forward<Functor>(functor), std::move(parents)));
2437 }
2438 
2439 
2440 /// A factory function to create a new value task
2441 template<typename Value>
2442 std::shared_ptr<transwarp::value_task<typename transwarp::decay<Value>::type>>
2443 make_value_task(std::string name, Value&& value) {
2445  return std::shared_ptr<task_t>(new task_t(true, std::move(name), std::forward<Value>(value)));
2446 }
2447 
2448 /// A factory function to create a new value task. Overload for omitting the task name
2449 template<typename Value>
2450 std::shared_ptr<transwarp::value_task<typename transwarp::decay<Value>::type>>
2451 make_value_task(Value&& value) {
2453  return std::shared_ptr<task_t>(new task_t(false, "", std::forward<Value>(value)));
2454 }
2455 
2456 
2457 } // transwarp
void set_custom_data_all(std::shared_ptr< void > custom_data) override
Assigns custom data to all tasks. transwarp will not directly use this. This is only useful if someth...
Definition: transwarp.h:1512
constexpr transwarp::wait_any_type wait_any
The wait_any task tag.
Definition: transwarp.h:127
void remove_custom_data_all() override
Removes custom data from all tasks.
Definition: transwarp.h:2254
The executor interface used to perform custom task execution.
Definition: transwarp.h:282
Exception thrown when a task was destroyed prematurely.
Definition: transwarp.h:60
A simple thread pool used to execute tasks in parallel.
Definition: transwarp.h:529
void cancel_all(bool) noexceptoverride
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2370
void operator()(const transwarp::itask &task) const
Definition: transwarp.h:983
Node manipulation.
Definition: transwarp.h:479
void remove_listeners(transwarp::event_type) override
No-op because a value task doesn&#39;t raise events.
Definition: transwarp.h:2281
Generates a graph.
Definition: transwarp.h:1007
The consume type. Used for tag dispatch.
Definition: transwarp.h:114
Determines the result type of the Functor dispatching on the task type.
Definition: transwarp.h:1148
void schedule_all(transwarp::executor &) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2302
Removes the executor from the given task.
Definition: transwarp.h:1067
Definition: transwarp.h:930
void reset_priority_all() override
Resets the priority of all tasks to 0.
Definition: transwarp.h:1494
Definition: transwarp.h:139
The task has no parents.
Just after a task was canceled (handle_event called on thread that task is run on) ...
TaskType task_type
The task type.
Definition: transwarp.h:2056
A callable to run a task given its parents.
Definition: transwarp.h:1327
void set_exception(std::exception_ptr exception) override
Assigns an exception to this task.
Definition: transwarp.h:2333
void schedule_all(bool reset_all) override
Schedules all tasks in the graph for execution on the caller thread. The task-specific executors get ...
Definition: transwarp.h:1642
void ensure_task_was_scheduled() const
Checks if the task was scheduled and throws transwarp::control_error if it&#39;s not. ...
Definition: transwarp.h:1806
Sets parents and level of the node.
Definition: transwarp.h:979
std::shared_ptr< transwarp::task_impl< TaskType_, typename std::decay< Functor_ >::type, result_type > > then(TaskType_, std::string name, Functor_ &&functor) const
Creates a continuation to this task.
Definition: transwarp.h:2136
std::shared_ptr< transwarp::value_task< typename transwarp::decay< Value >::type > > make_value_task(std::string name, Value &&value)
A factory function to create a new value task.
Definition: transwarp.h:2443
Definition: transwarp.h:1297
void set_executor(std::shared_ptr< transwarp::executor >) override
No-op because a value task never runs.
Definition: transwarp.h:2200
void set_executor_all(std::shared_ptr< transwarp::executor > executor) override
Assigns an executor to all tasks which takes precedence over the executor provided in schedule() or s...
Definition: transwarp.h:1452
bool was_scheduled() const noexceptoverride
Returns whether the task was scheduled and not reset afterwards. This means that the underlying futur...
Definition: transwarp.h:1700
void cancel(bool) noexceptoverride
No-op because a value task never runs.
Definition: transwarp.h:2367
Just before a task is scheduled (handle_event called on thread of caller to schedule()) ...
void schedule_all() override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2299
void schedule_all(transwarp::executor &executor, bool reset_all) override
Schedules all tasks in the graph for execution using the provided executor. The task-specific executo...
Definition: transwarp.h:1651
std::string get_name() const override
Returns the name of the executor.
Definition: transwarp.h:1410
Assigns a priority to the given task.
Definition: transwarp.h:1075
void set_value(const typename transwarp::decay< result_type >::type &value) override
Assigns a value to this task.
Definition: transwarp.h:2323
void schedule(transwarp::executor &executor) override
Schedules this task for execution using the provided executor. The task-specific executor gets preced...
Definition: transwarp.h:1608
Assigns an executor to the given task.
Definition: transwarp.h:1055
Unvisits the given task.
Definition: transwarp.h:1139
void set_priority(std::size_t priority) override
Sets a task priority (defaults to 0). transwarp will not directly use this. This is only useful if so...
Definition: transwarp.h:2213
void cancel_all(bool enabled) noexceptoverride
If enabled then all pending tasks in the graph are canceled which will throw transwarp::task_canceled...
Definition: transwarp.h:1748
void remove_custom_data() override
Removes custom data from this task.
Definition: transwarp.h:1519
void set_custom_data_all(std::shared_ptr< void > custom_data) override
Assigns custom data to all tasks. transwarp will not directly use this. This is only useful if someth...
Definition: transwarp.h:2244
std::tuple< std::shared_future< ParentResults >...> get_futures(const std::tuple< std::shared_ptr< transwarp::task< ParentResults >>...> &input)
Returns the futures from the given tuple of tasks.
Definition: transwarp.h:630
void set_value(typename transwarp::decay< result_type >::type &&value) override
Assigns a value to this task.
Definition: transwarp.h:2328
void schedule_all(transwarp::schedule_type type, bool reset_all) override
Schedules all tasks in the graph for execution on the caller thread. The task-specific executors get ...
Definition: transwarp.h:1676
Scheduling according to a breadth-first search (default)
void schedule_all() override
Schedules all tasks in the graph for execution on the caller thread. The task-specific executors get ...
Definition: transwarp.h:1625
void set_executor(std::shared_ptr< transwarp::executor > executor) override
Assigns an executor to this task which takes precedence over the executor provided in schedule() or s...
Definition: transwarp.h:1441
value_task(bool has_name, std::string name, T &&value)
A value task is defined by name and value. Note: Don&#39;t use this constructor directly, use transwarp::make_value_task.
Definition: transwarp.h:2169
void set_custom_data(std::shared_ptr< void > custom_data) override
Assigns custom data to this task. transwarp will not directly use this. This is only useful if someth...
Definition: transwarp.h:2235
The task class.
Definition: transwarp.h:405
Definition: transwarp.h:941
void schedule_all(transwarp::executor &executor, transwarp::schedule_type type, bool reset_all) override
Schedules all tasks in the graph for execution using the provided executor. The task-specific executo...
Definition: transwarp.h:1685
void reset_priority_all() override
Resets the priority of all tasks to 0.
Definition: transwarp.h:2229
void set_value(const typename transwarp::decay< result_type >::type &value) override
Assigns a value to this task. Scheduling will have no effect after a value has been set...
Definition: transwarp.h:1967
void set_priority(std::size_t priority) override
Sets a task priority (defaults to 0). transwarp will not directly use this. This is only useful if so...
Definition: transwarp.h:1474
const std::shared_ptr< std::string > & get_executor() const noexcept
The optional, task-specific executor (may be null)
Definition: transwarp.h:177
void assign_node_if(Functor &functor, std::shared_ptr< transwarp::node > node) noexcept
Assigns the node to the given functor if the functor is a subclass of transwarp::functor.
Definition: transwarp.h:1252
void reset() override
Resets this task.
Definition: transwarp.h:1724
const std::vector< std::shared_ptr< node > > & get_parents() const noexcept
The task&#39;s parents (may be empty)
Definition: transwarp.h:182
Just before a task&#39;s functor is invoked (handle_event called on thread that task is run on) ...
Definition: transwarp.h:928
The task&#39;s functor accepts the first parent future that becomes ready.
std::shared_ptr< transwarp::task_impl< TaskType_, typename std::decay< Functor_ >::type, result_type > > then(TaskType_, Functor_ &&functor) const
Creates a continuation to this task. Overload for omitting for task name.
Definition: transwarp.h:2144
The consume_any type. Used for tag dispatch.
Definition: transwarp.h:118
ResultType result_type
The result type of this task.
Definition: transwarp.h:1963
void set_custom_data(std::shared_ptr< void > custom_data) override
Assigns custom data to this task. transwarp will not directly use this. This is only useful if someth...
Definition: transwarp.h:1502
bool was_scheduled() const noexceptoverride
Returns true because a value task is scheduled once on construction.
Definition: transwarp.h:2338
void execute(const std::function< void()> &functor, const std::shared_ptr< transwarp::node > &) override
Pushes the functor into the thread pool for asynchronous execution.
Definition: transwarp.h:1415
void wait_for_all(const std::vector< std::shared_ptr< transwarp::task< ParentResultType >>> &parents)
Waits for all parents to finish.
Definition: transwarp.h:675
The accept type. Used for tag dispatch.
Definition: transwarp.h:106
const std::shared_future< result_type > & get_future() const noexceptoverride
Returns the future associated to the underlying execution.
Definition: transwarp.h:1532
constexpr transwarp::accept_type accept
The accept task tag.
Definition: transwarp.h:107
void schedule_all(transwarp::executor &, transwarp::schedule_type, bool) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2320
void remove_listeners() override
No-op because a value task doesn&#39;t raise events.
Definition: transwarp.h:2284
bool has_result() const noexceptoverride
Returns whether this task contains a result.
Definition: transwarp.h:1719
The listener interface to listen to events raised by tasks.
Definition: transwarp.h:310
Exception thrown when a task is canceled.
Definition: transwarp.h:51
TaskType task_type
The task type.
Definition: transwarp.h:2105
void wait() const override
Waits for the task to complete. Should only be called if was_scheduled() is true, throws transwarp::c...
Definition: transwarp.h:1706
Scheduling according to a depth-first search.
void add_listener(transwarp::event_type event, std::shared_ptr< transwarp::listener > listener) override
Adds a new listener for the given event type only.
Definition: transwarp.h:1551
std::size_t get_level() const noexcept
The task level.
Definition: transwarp.h:162
std::shared_future< ResultType > make_future_with_value(Value &&value)
Returns a ready future with the given value as its state.
Definition: transwarp.h:1258
std::string get_name() const override
Returns the name of the executor.
Definition: transwarp.h:1384
void schedule() override
Schedules this task for execution on the caller thread. The task-specific executor gets precedence if...
Definition: transwarp.h:1591
void add_listener(std::shared_ptr< transwarp::listener >) override
No-op because a value task doesn&#39;t raise events.
Definition: transwarp.h:2269
void add_listener(std::shared_ptr< transwarp::listener > listener) override
Adds a new listener for all event types.
Definition: transwarp.h:1542
void ensure_task_not_running() const
Checks if the task is currently running and throws transwarp::control_error if it is...
Definition: transwarp.h:1799
Just before a task starts running (handle_event called on thread that task is run on) ...
Removes custom data from the given task.
Definition: transwarp.h:1107
void schedule_all(transwarp::executor &, bool) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2308
Resets the given task.
Definition: transwarp.h:1035
const std::shared_ptr< std::string > & get_name() const noexcept
The optional task name (may be null)
Definition: transwarp.h:172
A base class for a user-defined functor that needs access to the node associated to the task or a can...
Definition: transwarp.h:446
constexpr transwarp::consume_any_type consume_any
The consume_any task tag.
Definition: transwarp.h:119
bool is_canceled() const noexcept
Returns whether the associated task is canceled.
Definition: transwarp.h:197
A node carrying meta-data of a task.
Definition: transwarp.h:145
Exception thrown when a task is used in unintended ways.
Definition: transwarp.h:78
Resets the priority of the given task.
Definition: transwarp.h:1087
void schedule() override
No-op because a value task never runs.
Definition: transwarp.h:2287
Returns the result type of a std::shared_future&lt;T&gt;
Definition: transwarp.h:398
void add_listener(transwarp::event_type, std::shared_ptr< transwarp::listener >) override
No-op because a value task doesn&#39;t raise events.
Definition: transwarp.h:2272
void wait() const override
No-op because a value task never runs.
Definition: transwarp.h:2343
void schedule_all(transwarp::executor &executor, transwarp::schedule_type type) override
Schedules all tasks in the graph for execution using the provided executor. The task-specific executo...
Definition: transwarp.h:1667
std::size_t get_priority() const noexcept
The task priority (defaults to 0)
Definition: transwarp.h:187
const std::shared_ptr< transwarp::node > & get_parent() const noexcept
Returns the parent node.
Definition: transwarp.h:250
A task proxy.
Definition: transwarp.h:1957
void transwarp_cancel_point() const
If the associated task is canceled then this will throw transwarp::task_canceled which will stop the ...
Definition: transwarp.h:460
The base task class that contains the functionality that can be used with all result types (void and ...
Definition: transwarp.h:1430
Definition: transwarp.h:616
The task&#39;s functor consumes all parent results.
Just after a task has finished running (handle_event called on thread that task is run on) ...
void remove_listener(const std::shared_ptr< transwarp::listener > &) override
No-op because a value task doesn&#39;t raise events.
Definition: transwarp.h:2275
void set_priority_all(std::size_t priority) override
Sets a priority to all tasks (defaults to 0). transwarp will not directly use this. This is only useful if something else is using the priority.
Definition: transwarp.h:2219
Base class for exceptions.
Definition: transwarp.h:42
void schedule_all(transwarp::schedule_type) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2311
constexpr transwarp::consume_type consume
The consume task tag.
Definition: transwarp.h:115
std::size_t get_id() const noexcept
The task ID.
Definition: transwarp.h:157
void schedule(transwarp::executor &, bool) override
No-op because a value task never runs.
Definition: transwarp.h:2296
A value task that stores a single value and doesn&#39;t require scheduling. Value tasks should be created...
Definition: transwarp.h:2155
void reset_priority() override
Resets the task priority to 0.
Definition: transwarp.h:2224
void remove_listeners(transwarp::event_type event) override
Removes all listeners for the given event type.
Definition: transwarp.h:1575
void remove_listener(const std::shared_ptr< transwarp::listener > &listener) override
Removes the listener for all event types.
Definition: transwarp.h:1558
Determines the type of the parents.
Definition: transwarp.h:1285
const std::shared_ptr< transwarp::node > & transwarp_node() const noexcept
The node associated to the task.
Definition: transwarp.h:454
Parent wait_for_any(const std::shared_ptr< transwarp::task< ParentResults >> &...parents)
Waits for the first parent to finish.
Definition: transwarp.h:698
void reset_all() override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2364
void remove_listener(transwarp::event_type event, const std::shared_ptr< transwarp::listener > &listener) override
Removes the listener for the given event type only.
Definition: transwarp.h:1567
Assigns custom data to the given task.
Definition: transwarp.h:1095
The task&#39;s functor takes no arguments but waits for all parents to finish.
void set_exception(std::exception_ptr exception) override
Assigns an exception to this task. Scheduling will have no effect after an exception has been set...
Definition: transwarp.h:1692
void cancel_all_but_one(const std::shared_ptr< transwarp::task< OneResult >> &one, const std::vector< std::shared_ptr< transwarp::task< ParentResultType >>> &parents)
Cancels all tasks but one.
Definition: transwarp.h:737
Definition: transwarp.h:756
Removes reference and const from a type.
Definition: transwarp.h:391
void remove_listeners() override
Removes all listeners.
Definition: transwarp.h:1581
const std::shared_ptr< void > & get_custom_data() const noexcept
The custom task data (may be null)
Definition: transwarp.h:192
std::shared_ptr< transwarp::task_impl< TaskType_, typename std::decay< Functor_ >::type, result_type > > then(TaskType_, std::string name, Functor_ &&functor) const
Creates a continuation to this task.
Definition: transwarp.h:2186
void call_with_each(const Functor &f, const std::tuple< std::shared_ptr< transwarp::task< ParentResults >>...> &t)
Calls the functor with every element in the tuple.
Definition: transwarp.h:960
void reset_priority() override
Resets the task priority to 0.
Definition: transwarp.h:1488
void schedule_all(transwarp::executor &, transwarp::schedule_type) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2314
void schedule_all(transwarp::schedule_type, bool) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2317
An edge between two nodes.
Definition: transwarp.h:236
const std::shared_ptr< transwarp::node > & get_child() const noexcept
Returns the child node.
Definition: transwarp.h:255
void schedule_all(transwarp::schedule_type type) override
Schedules all tasks in the graph for execution on the caller thread. The task-specific executors get ...
Definition: transwarp.h:1659
std::vector< transwarp::edge > get_graph() const override
Returns the graph of the task structure. This is mainly for visualizing the tasks and their interdepe...
Definition: transwarp.h:1756
void set_value(typename transwarp::decay< result_type >::type &value) override
Assigns a value to this task. Scheduling will have no effect after a value has been set...
Definition: transwarp.h:2019
void schedule(transwarp::executor &) override
No-op because a value task never runs.
Definition: transwarp.h:2290
void reset_all() override
Resets all tasks in the graph.
Definition: transwarp.h:1732
Definition: transwarp.h:747
ResultType result_type
The result type of this task.
Definition: transwarp.h:2162
Executor for parallel execution. Uses a simple thread pool.
Definition: transwarp.h:1396
The wait_any type. Used for tag dispatch.
Definition: transwarp.h:126
void set_priority_all(std::size_t priority) override
Sets a priority to all tasks (defaults to 0). transwarp will not directly use this. This is only useful if something else is using the priority (e.g. a custom executor)
Definition: transwarp.h:1481
void schedule(bool reset) override
Schedules this task for execution on the caller thread. The task-specific executor gets precedence if...
Definition: transwarp.h:1600
virtual std::string get_name() const =0
Returns the name of the executor.
std::string to_string(const transwarp::task_type &type)
String conversion for the task_type enumeration.
Definition: transwarp.h:87
virtual void execute(const std::function< void()> &functor, const std::shared_ptr< transwarp::node > &node)=0
Runs a task which is wrapped by the given functor. The functor only captures one shared pointer and c...
void execute(const std::function< void()> &functor, const std::shared_ptr< transwarp::node > &) override
Runs the functor on the current thread.
Definition: transwarp.h:1389
void remove_executor() override
Removes the executor from this task.
Definition: transwarp.h:1459
The accept_any type. Used for tag dispatch.
Definition: transwarp.h:110
transwarp::task_type get_type() const noexcept
The task type.
Definition: transwarp.h:167
std::shared_ptr< transwarp::task_impl< TaskType, typename std::decay< Functor >::type, typename Parents::result_type...> > make_task(TaskType, std::string name, Functor &&functor, std::shared_ptr< Parents >...parents)
A factory function to create a new task.
Definition: transwarp.h:2409
virtual void handle_event(transwarp::event_type event, const std::shared_ptr< transwarp::node > &node)=0
This may be called from arbitrary threads depending on the event type (see transwarp::event_type). The implementer needs to ensure that this never throws exceptions.
bool is_ready() const override
Returns true because a value task is always ready.
Definition: transwarp.h:2346
constexpr transwarp::wait_type wait
The wait task tag.
Definition: transwarp.h:123
The root type. Used for tag dispatch.
Definition: transwarp.h:102
The task&#39;s functor consumes the first parent result that becomes ready.
A task representing a piece of work given by functor and parent tasks. By connecting tasks a directed...
Definition: transwarp.h:2102
void remove_listener(transwarp::event_type, const std::shared_ptr< transwarp::listener > &) override
No-op because a value task doesn&#39;t raise events.
Definition: transwarp.h:2278
The wait type. Used for tag dispatch.
Definition: transwarp.h:122
void cancel(bool enabled) noexceptoverride
If enabled then this task is canceled which will throw transwarp::task_canceled when retrieving the t...
Definition: transwarp.h:1741
Schedules using the given executor.
Definition: transwarp.h:1022
Result call(std::size_t node_id, const Task &task, const std::tuple< std::shared_ptr< transwarp::task< ParentResults >>...> &parents)
Calls the functor of the given task with the results from the tuple of parents. Throws transwarp::tas...
Definition: transwarp.h:913
constexpr transwarp::root_type root
The root task tag.
Definition: transwarp.h:103
void remove_custom_data() override
Removes custom data from this task.
Definition: transwarp.h:2249
void set_value(typename transwarp::decay< result_type >::type &&value) override
Assigns a value to this task. Scheduling will have no effect after a value has been set...
Definition: transwarp.h:1975
void schedule_all(transwarp::executor &executor) override
Schedules all tasks in the graph for execution using the provided executor. The task-specific executo...
Definition: transwarp.h:1633
static Result work(std::size_t node_id, const Task &task, const std::tuple< std::shared_ptr< transwarp::task< ParentResults >>...> &parents)
Definition: transwarp.h:840
schedule_type
Determines in which order tasks are scheduled in the graph.
Definition: transwarp.h:321
The task&#39;s functor accepts all parent futures.
Applies final bookkeeping to the task.
Definition: transwarp.h:995
void remove_executor_all() override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2209
typename transwarp::detail::functor_result< TaskType, Functor, ParentResults...>::type result_type
The result type of this task.
Definition: transwarp.h:2108
void schedule(transwarp::executor &executor, bool reset) override
Schedules this task for execution using the provided executor. The task-specific executor gets preced...
Definition: transwarp.h:1617
std::vector< transwarp::edge > get_graph() const override
Returns an empty graph because a value task doesn&#39;t have parents.
Definition: transwarp.h:2373
const std::shared_ptr< transwarp::node > & get_node() const noexceptoverride
Returns the associated node.
Definition: transwarp.h:1537
task_impl(bool has_name, std::string name, F &&functor, std::vector< std::shared_ptr< transwarp::task< P >>> parents)
A task is defined by name, functor, and parent tasks. Note: Don&#39;t use this constructor directly...
Definition: transwarp.h:2123
task_type
The possible task types.
Definition: transwarp.h:30
void set_value() override
Assigns a value to this task. Scheduling will have no effect after a call to this. Calling reset() will reset this and re-enable scheduling.
Definition: transwarp.h:2063
std::shared_ptr< transwarp::task_impl< TaskType_, typename std::decay< Functor_ >::type, result_type > > then(TaskType_, Functor_ &&functor) const
Creates a continuation to this task. Overload for omitting the task name.
Definition: transwarp.h:2194
event_type
The task events that can be subscribed to using the listener interface.
Definition: transwarp.h:299
The task&#39;s functor takes no arguments but waits for the first parent to finish.
Exception thrown when an invalid parameter was passed to a function.
Definition: transwarp.h:69
void remove_custom_data_all() override
Removes custom data from all tasks.
Definition: transwarp.h:1525
void reset() override
No-op because a value task never runs.
Definition: transwarp.h:2361
std::shared_future< void > make_ready_future()
Returns a ready future.
Definition: transwarp.h:1265
void set_executor_all(std::shared_ptr< transwarp::executor >) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2203
constexpr transwarp::accept_any_type accept_any
The accept_any task tag.
Definition: transwarp.h:111
const std::shared_future< result_type > & get_future() const noexceptoverride
Returns the future associated to the underlying execution.
Definition: transwarp.h:2259
Cancels or resumes the given task.
Definition: transwarp.h:1043
void remove_executor() override
No-op because a value task never runs.
Definition: transwarp.h:2206
void schedule(bool) override
No-op because a value task never runs.
Definition: transwarp.h:2293
void schedule_all(bool) override
No-op because a value task never runs and doesn&#39;t have parents.
Definition: transwarp.h:2305
Visits the given task using the visitor given in the constructor.
Definition: transwarp.h:1127
Pushes the given task into the vector of tasks.
Definition: transwarp.h:1115
Executor for sequential execution. Runs functors sequentially on the same thread. ...
Definition: transwarp.h:1372
bool has_result() const noexceptoverride
Returns true because a value task always contains a result.
Definition: transwarp.h:2351
task_impl(bool has_name, std::string name, F &&functor, std::shared_ptr< transwarp::task< ParentResults >>...parents)
A task is defined by name, functor, and parent tasks. Note: Don&#39;t use this constructor directly...
Definition: transwarp.h:2115
std::shared_future< ResultType > make_future_with_exception(std::exception_ptr exception)
Returns a ready future with the given exception as its state.
Definition: transwarp.h:1273
void remove_executor_all() override
Removes the executor from all tasks.
Definition: transwarp.h:1466
const std::shared_ptr< transwarp::node > & get_node() const noexceptoverride
Returns the associated node.
Definition: transwarp.h:2264
bool is_ready() const override
Returns whether the task has finished processing. Should only be called if was_scheduled() is true...
Definition: transwarp.h:1713
An interface for the task class.
Definition: transwarp.h:328
Result run_task(std::size_t node_id, const std::weak_ptr< Task > &task, Args &&...args)
Runs the task with the given arguments, hence, invoking the task&#39;s functor.
Definition: transwarp.h:650