| Taskflow: A General-purpose Task-parallel Programming System

Loading...
Searching...
No Matches
runtime.hpp
1#pragma once
2
3#include "executor.hpp"
4
5 namespace tf {
6
7// ------------------------------------------------------------------------------------------------
8// class: Runtime
9// ------------------------------------------------------------------------------------------------
10
47 class Runtime {
48
49 friend class Executor;
50 friend class FlowBuilder;
51 friend class PreemptionGuard;
52 friend class Algorithm;
53
54 public:
55
71 Executor& executor();
72
76 inline Worker& worker();
77
118 void schedule(Task task);
119
120 // ----------------------------------------------------------------------------------------------
121 // async methods
122 // ----------------------------------------------------------------------------------------------
123
156 template <typename F>
157 auto async(F&& f);
158
179 template <TaskParamsLike P, typename F>
180 auto async(P&& params, F&& f);
181
182 // ----------------------------------------------------------------------------------------------
183 // silent async methods
184 // ----------------------------------------------------------------------------------------------
185
206 template <typename F>
207 void silent_async(F&& f);
208
227 template <TaskParamsLike P, typename F>
228 void silent_async(P&& params, F&& f);
229
230 // ----------------------------------------------------------------------------------------------
231 // dependent async methods
232 // ----------------------------------------------------------------------------------------------
233
268 template <typename F, AsyncTaskHandleLike... Tasks>
269 auto dependent_async(F&& func, Tasks&&... tasks);
270
309 template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
310 auto dependent_async(P&& params, F&& func, Tasks&&... tasks);
311
349 template <typename F, std::input_iterator I>
350 auto dependent_async(F&& func, I first, I last);
351
393 template <TaskParamsLike P, typename F, std::input_iterator I>
394 auto dependent_async(P&& params, F&& func, I first, I last);
395
396 // ----------------------------------------------------------------------------------------------
397 // silent dependent async methods
398 // ----------------------------------------------------------------------------------------------
399
427 template <typename F, AsyncTaskHandleLike... Tasks>
428 tf::AsyncTask silent_dependent_async(F&& func, Tasks&&... tasks);
429
462 template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
463 tf::AsyncTask silent_dependent_async(P&& params, F&& func, Tasks&&... tasks);
464
497 template <typename F, std::input_iterator I>
498 tf::AsyncTask silent_dependent_async(F&& func, I first, I last);
499
535 template <TaskParamsLike P, typename F, std::input_iterator I>
536 tf::AsyncTask silent_dependent_async(P&& params, F&& func, I first, I last);
537
538
539
540 // ----------------------------------------------------------------------------------------------
541 // cooperative execution methods
542 // ----------------------------------------------------------------------------------------------
543
574 void corun();
575
579 void corun_all();
580
584 bool is_cancelled();
585
586 private:
587
591 explicit Runtime(Executor&, Worker&, Node*);
592
596 Executor& _executor;
597
601 Worker& _worker;
602
606 Node* _node;
607};
608
609// constructor
610 inline Runtime::Runtime(Executor& executor, Worker& worker, Node* node) :
611 _executor {executor},
612 _worker {worker},
613 _node {node} {
614}
615
616// Function: executor
617 inline Executor& Runtime::executor() {
618 return _executor;
619}
620
621// Function: worker
622 inline Worker& Runtime::worker() {
623 return _worker;
624}
625
626// Procedure: schedule
627 inline void Runtime::schedule(Task task) {
628
629 auto node = task._node;
630 // need to keep the invariant: when scheduling a task, the task must have
631 // zero dependency (join counter is 0)
632 // or we can encounter bug when inserting a nested flow (e.g., module task)
633 node->_join_counter.store(0, std::memory_order_relaxed);
634
635 auto& j = node->_parent ? node->_parent->_join_counter :
636 node->_topology->_join_counter;
637 j.fetch_add(1, std::memory_order_relaxed);
638 _executor._schedule(_worker, node);
639}
640
641// Function: corun
642 inline void Runtime::corun() {
643 {
644 ExplicitAnchorGuard anchor(_node);
645 _executor._corun_until(_worker, [this] () -> bool {
646 return _node->_join_counter.load(std::memory_order_acquire) == 1;
647 });
648 }
649 _node->_rethrow_exception();
650}
651
652// Function: corun_all
653 inline void Runtime::corun_all() {
654 corun();
655}
656
657 inline bool Runtime::is_cancelled() {
658 return _node->_is_parent_cancelled();
659}
660
661// ------------------------------------------------------------------------------------------------
662// Runtime::silent_async
663// ------------------------------------------------------------------------------------------------
664
665// Function: silent_async
666 template <typename F>
667 void Runtime::silent_async(F&& f) {
668 silent_async(DefaultTaskParams{}, std::forward<F>(f));
669}
670
671// Function: silent_async
672 template <TaskParamsLike P, typename F>
673 void Runtime::silent_async(P&& params, F&& f) {
674 _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
675 _executor._silent_async(
676 std::forward<P>(params), std::forward<F>(f), _node->_topology, _node
677 );
678}
679
680// ------------------------------------------------------------------------------------------------
681// Runtime::async
682// ------------------------------------------------------------------------------------------------
683
684// Function: async
685 template <typename F>
686 auto Runtime::async(F&& f) {
687 return async(DefaultTaskParams{}, std::forward<F>(f));
688}
689
690// Function: async
691 template <TaskParamsLike P, typename F>
692 auto Runtime::async(P&& params, F&& f) {
693 _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
694 return _executor._async(
695 std::forward<P>(params), std::forward<F>(f), _node->_topology, _node
696 );
697}
698
699// ------------------------------------------------------------------------------------------------
700// silent dependent async
701// ------------------------------------------------------------------------------------------------
702
703// Function: silent_dependent_async
704 template <typename F, AsyncTaskHandleLike... Tasks>
705 tf::AsyncTask Runtime::silent_dependent_async(F&& func, Tasks&&... tasks) {
706 return silent_dependent_async(
707 DefaultTaskParams{}, std::forward<F>(func), std::forward<Tasks>(tasks)...
708 );
709}
710
711// Function: silent_dependent_async
712 template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
713 tf::AsyncTask Runtime::silent_dependent_async(
714 P&& params, F&& func, Tasks&&... tasks
715){
716 std::array<AsyncTask*, sizeof...(Tasks)> array = { (&tasks)... };
717 return silent_dependent_async(
718 std::forward<P>(params), std::forward<F>(func), array.begin(), array.end()
719 );
720}
721
722// Function: silent_dependent_async
723 template <typename F, std::input_iterator I>
724 tf::AsyncTask Runtime::silent_dependent_async(F&& func, I first, I last) {
725 return silent_dependent_async(DefaultTaskParams{}, std::forward<F>(func), first, last);
726}
727
728// Function: silent_dependent_async
729 template <TaskParamsLike P, typename F, std::input_iterator I>
730 tf::AsyncTask Runtime::silent_dependent_async(
731 P&& params, F&& func, I first, I last
732) {
733 _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
734 return _executor._silent_dependent_async(
735 std::forward<P>(params), std::forward<F>(func), first, last, _node->_topology, _node
736 );
737}
738
739// ------------------------------------------------------------------------------------------------
740// dependent async
741// ------------------------------------------------------------------------------------------------
742
743// Function: dependent_async
744 template <typename F, AsyncTaskHandleLike... Tasks>
745 auto Runtime::dependent_async(F&& func, Tasks&&... tasks) {
746 return dependent_async(DefaultTaskParams{}, std::forward<F>(func), std::forward<Tasks>(tasks)...);
747}
748
749// Function: dependent_async
750 template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
751 auto Runtime::dependent_async(P&& params, F&& func, Tasks&&... tasks) {
752 std::array<AsyncTask*, sizeof...(Tasks)> array = { (&tasks)... };
753 return dependent_async(
754 std::forward<P>(params), std::forward<F>(func), array.begin(), array.end()
755 );
756}
757
758// Function: dependent_async
759 template <typename F, std::input_iterator I>
760 auto Runtime::dependent_async(F&& func, I first, I last) {
761 return dependent_async(DefaultTaskParams{}, std::forward<F>(func), first, last);
762}
763
764// Function: dependent_async
765 template <TaskParamsLike P, typename F, std::input_iterator I>
766 auto Runtime::dependent_async(P&& params, F&& func, I first, I last) {
767 _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
768 return _executor._dependent_async(
769 std::forward<P>(params), std::forward<F>(func), first, last, _node->_topology, _node
770 );
771}
772
773// ----------------------------------------------------------------------------
774// Executor Forward Declaration
775// ----------------------------------------------------------------------------
776
777// Procedure: _invoke_runtime_task
778 inline bool Executor::_invoke_runtime_task(Worker& worker, Node* node) {
779 return _invoke_runtime_task_impl(
780 worker, node, std::get_if<Node::Runtime>(&node->_handle)->work
781 );
782}
783
784// Function: _invoke_runtime_task_impl
785 inline bool Executor::_invoke_runtime_task_impl(
786 Worker& worker, Node* node, std::function<void(Runtime&)>& work
787) {
788 // first time
789 if((node->_nstate & NSTATE::PREEMPTED) == 0) {
790
791 Runtime rt(*this, worker, node);
792
793 node->_nstate |= (NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
794
795 node->_join_counter.fetch_add(1, std::memory_order_release);
796
797 _observer_prologue(worker, node);
798 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
799 work(rt);
800 });
801 _observer_epilogue(worker, node);
802
803 // Last one to leave the runtime; no need to preempt this runtime.
804 if(node->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
805 node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
806 }
807 // There are still child tasks running; need to preempt this runtime.
808 // Here, we cannot let caller check the state from node->_nstate due to data race,
809 // but return a stateless boolean to indicate preemption.
810 // Ex: if preempted, another task may finish real quck and insert this parent task
811 // again into the scheduling queue. When running this parent task, it will jump to
812 // else branch below and modify tne nstate, thus incuring data race.
813 else {
814 return true;
815 }
816 }
817 // second time - previously preempted
818 else {
819 node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
820 }
821 return false;
822}
823
824// Function: _invoke_runtime_task_impl
825 inline bool Executor::_invoke_runtime_task_impl(
826 Worker& worker, Node* node, std::function<void(Runtime&, bool)>& work
827) {
828
829 Runtime rt(*this, worker, node);
830
831 // first time
832 if((node->_nstate & NSTATE::PREEMPTED) == 0) {
833
834 node->_nstate |= (NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
835 node->_join_counter.fetch_add(1, std::memory_order_release);
836
837 _observer_prologue(worker, node);
838 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
839 work(rt, false);
840 });
841 _observer_epilogue(worker, node);
842
843 // Last one to leave this runtime; no need to preempt this runtime
844 if(node->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
845 node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
846 }
847 // Here, we cannot let caller check the state from node->_nstate due to data race,
848 // but return a stateless boolean to indicate preemption.
849 // Ex: if preempted, another task may finish real quck and insert this parent task
850 // again into the scheduling queue. When running this parent task, it will jump to
851 // else branch below and modify tne nstate, thus incuring data race.
852 else {
853 return true;
854 }
855 }
856 // second time - previously preempted
857 else {
858 node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
859 }
860
861 // clean up outstanding work (e.g., exception)
862 work(rt, true);
863
864 return false;
865}
866
867// ------------------------------------------------------------------------------------------------
868// class: NonpreemptiveRuntime (internal use only)
869// ------------------------------------------------------------------------------------------------
870
874 class NonpreemptiveRuntime {
875
876 friend class Executor;
877
878 public:
879
883 void schedule(Task task);
884
885 private:
886
890 explicit NonpreemptiveRuntime(Executor& executor, Worker& worker) :
891 _executor {executor}, _worker {worker}{
892 }
893
897 Executor& _executor;
898
902 Worker& _worker;
903};
904
905// Procedure: schedule
906 inline void NonpreemptiveRuntime::schedule(Task task) {
907
908 auto node = task._node;
909 // need to keep the invariant: when scheduling a task, the task must have
910 // zero dependency (join counter is 0)
911 // or we can encounter bug when inserting a nested flow (e.g., module task)
912 node->_join_counter.store(0, std::memory_order_relaxed);
913
914 auto& j = node->_parent ? node->_parent->_join_counter :
915 node->_topology->_join_counter;
916 j.fetch_add(1, std::memory_order_relaxed);
917 _executor._schedule(_worker, node);
918}
919
920// ----------------------------------------------------------------------------
921// Executor Forward Declaration
922// ----------------------------------------------------------------------------
923
924// Procedure: _invoke_nonpreemptive_runtime_task
925 inline void Executor::_invoke_nonpreemptive_runtime_task(Worker& worker, Node* node) {
926 _observer_prologue(worker, node);
927 tf::NonpreemptiveRuntime nprt(*this, worker);
928 TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
929 std::get_if<Node::NonpreemptiveRuntime>(&node->_handle)->work(nprt);
930 });
931 _observer_epilogue(worker, node);
932}
933
934
935// Function: run_until
936 template <typename P, typename C>
937 tf::Future<void> Executor::run_until(Taskflow&& f, P&& p, C&& c) {
938
939 // No need to create a real topology but returns an dummy future for invariant.
940 if(f.empty() || p()) {
941 c();
942 std::promise<void> promise;
943 promise.set_value();
944 return tf::Future<void>(promise.get_future());
945 }
946
947 _increment_topology();
948
949 auto g = std::make_unique<Taskflow>(std::move(f));
950
951 // creates a topology for this run
952 auto t = std::make_shared<Topology>(*g, std::forward<P>(p), std::forward<C>(c));
953 //auto t = std::make_shared<DerivedTopology<P, C>>(*g, std::forward<P>(p), std::forward<C>(c));
954
955 // need to create future before the topology got torn down quickly
956 tf::Future<void> future(t->_promise.get_future(), t);
957
958 // creates a silent-async that holds the taskflow
959 silent_async([g=MoC{std::move(g)}, t](tf::Runtime& rt) mutable {
960 t->_parent = rt._node;
961 t->_parent->_join_counter.fetch_add(1, std::memory_order_release);
962 if(g.object->_fetch_enqueue(t) == 0) {
963 rt._executor._schedule_graph(
964 rt._worker, g.object->_graph, t.get(), t.get()
965 );
966 }
967 });
968
969 return future;
970}
971
972} // end of namespace tf -----------------------------------------------------
tf::AsyncTask
class to hold a dependent asynchronous task with shared ownership
Definition async_task.hpp:45
tf::DefaultTaskParams
class to create an empty task parameter for compile-time optimization
Definition graph.hpp:217
tf::Executor
class to create an executor
Definition executor.hpp:62
tf::Executor::silent_async
void silent_async(P &&params, F &&func)
similar to tf::Executor::async but does not return a future object
tf::Executor::run_until
tf::Future< void > run_until(Taskflow &taskflow, P &&pred)
runs a taskflow multiple times until the predicate becomes true
tf::Future
class to access the result of an execution
Definition taskflow.hpp:640
tf::Runtime
class to create a runtime task
Definition runtime.hpp:47
tf::Runtime::silent_async
void silent_async(F &&f)
runs the given function asynchronously without returning any future object
Definition runtime.hpp:667
tf::Runtime::is_cancelled
bool is_cancelled()
queries if this runtime task has been cancelled
Definition runtime.hpp:657
tf::Runtime::executor
Executor & executor()
obtains the running executor
Definition runtime.hpp:617
tf::Runtime::async
auto async(F &&f)
runs the given callable asynchronously
Definition runtime.hpp:686
tf::Runtime::silent_dependent_async
tf::AsyncTask silent_dependent_async(F &&func, Tasks &&... tasks)
runs the given function asynchronously when the given predecessors finish
Definition runtime.hpp:705
tf::Runtime::schedule
void schedule(Task task)
schedules an active task immediately to the worker's queue
Definition runtime.hpp:627
tf::Runtime::corun
void corun()
corun all tasks spawned by this runtime with other workers
Definition runtime.hpp:642
tf::Runtime::dependent_async
auto dependent_async(F &&func, Tasks &&... tasks)
runs the given function asynchronously when the given predecessors finish
Definition runtime.hpp:745
tf::Runtime::worker
Worker & worker()
acquire a reference to the underlying worker
Definition runtime.hpp:622
tf::Runtime::corun_all
void corun_all()
equivalent to tf::Runtime::corun - just an alias for legacy purpose
Definition runtime.hpp:653
tf::Task
class to create a task handle over a taskflow node
Definition task.hpp:569
tf::Taskflow
class to create a taskflow object
Definition taskflow.hpp:64
tf::Worker
class to create a worker in an executor
Definition worker.hpp:55
tf::AsyncTaskHandleLike
concept to check if a type is a tf::AsyncTask
Definition async_task.hpp:314
tf::TaskParamsLike
concept to check if a type is a task parameter
Definition graph.hpp:228
tf
taskflow namespace
Definition small_vector.hpp:20