tf::Executor class

#include <taskflow/core/executor.hpp>

tf::Executor::Executor(size_t N = std::thread::hardware_concurrency(), std::shared_ptr<WorkerInterface> wix = nullptr) explicit

constructs the executor with N worker threads

The constructor spawns N worker threads to run tasks in a work-stealing loop. The number of workers must be greater than zero or an exception will be thrown. By default, the number of worker threads is equal to the maximum hardware concurrency returned by std::thread::hardware_concurrency.

Users can alter the worker behavior, such as changing thread affinity, via deriving an instance from tf::WorkerInterface.

tf::Executor::~Executor()

destructs the executor

The destructor calls Executor::wait_for_all to wait for all submitted taskflows to complete and then notifies all worker threads to stop and join these threads.

tf::Future<void> tf::Executor::run(Taskflow& taskflow)

runs a taskflow once

This member function executes the given taskflow once and returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run(taskflow);
// do something else
future.wait();

This member function is thread-safe.

Attention

The executor does not own the given taskflow. It is your responsibility to ensure the taskflow remains alive during its execution.

tf::Future<void> tf::Executor::run(Taskflow&& taskflow)

runs a moved taskflow once

This member function executes a moved taskflow once and returns a tf::Future object that eventually holds the result of the execution. The executor will take care of the lifetime of the moved taskflow.

tf::Future<void> future = executor.run(std::move(taskflow));
// do something else
future.wait();

This member function is thread-safe.

template<typename C>

tf::Future<void> tf::Executor::run(Taskflow& taskflow, C&& callable)

runs a taskflow once and invoke a callback upon completion

This member function executes the given taskflow once and invokes the given callable when the execution completes. This member function returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run(taskflow, [](){ std::cout << "done"; });
// do something else
future.wait();

This member function is thread-safe.

Attention

The executor does not own the given taskflow. It is your responsibility to ensure the taskflow remains alive during its execution.

template<typename C>

tf::Future<void> tf::Executor::run(Taskflow&& taskflow, C&& callable)

runs a moved taskflow once and invoke a callback upon completion

This member function executes a moved taskflow once and invokes the given callable when the execution completes. This member function returns a tf::Future object that eventually holds the result of the execution. The executor will take care of the lifetime of the moved taskflow.

tf::Future<void> future = executor.run(
  std::move(taskflow), [](){ std::cout << "done"; }
);
// do something else
future.wait();

This member function is thread-safe.

tf::Future<void> tf::Executor::run_n(Taskflow& taskflow, size_t N)

runs a taskflow for N times

This member function executes the given taskflow N times and returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run_n(taskflow, 2);  // run taskflow 2 times
// do something else
future.wait();

This member function is thread-safe.

Attention

The executor does not own the given taskflow. It is your responsibility to ensure the taskflow remains alive during its execution.

tf::Future<void> tf::Executor::run_n(Taskflow&& taskflow, size_t N)

runs a moved taskflow for N times

This member function executes a moved taskflow N times and returns a tf::Future object that eventually holds the result of the execution. The executor will take care of the lifetime of the moved taskflow.

tf::Future<void> future = executor.run_n(
  std::move(taskflow), 2    // run the moved taskflow 2 times
);
// do something else
future.wait();

This member function is thread-safe.

template<typename C>

tf::Future<void> tf::Executor::run_n(Taskflow& taskflow, size_t N, C&& callable)

runs a taskflow for N times and then invokes a callback

This member function executes the given taskflow N times and invokes the given callable when the execution completes. This member function returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run(
  taskflow, 2, [](){ std::cout << "done"; }  // runs taskflow 2 times and invoke
                                             // the lambda to print "done"
);
// do something else
future.wait();

This member function is thread-safe.

Attention

The executor does not own the given taskflow. It is your responsibility to ensure the taskflow remains alive during its execution.

template<typename C>

tf::Future<void> tf::Executor::run_n(Taskflow&& taskflow, size_t N, C&& callable)

runs a moved taskflow for N times and then invokes a callback

This member function executes a moved taskflow N times and invokes the given callable when the execution completes. This member function returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run_n(
  // run the moved taskflow 2 times and invoke the lambda to print "done"
  std::move(taskflow), 2, [](){ std::cout << "done"; }
);
// do something else
future.wait();

This member function is thread-safe.

template<typename P>

tf::Future<void> tf::Executor::run_until(Taskflow& taskflow, P&& pred)

runs a taskflow multiple times until the predicate becomes true

This member function executes the given taskflow multiple times until the predicate returns true. This member function returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run_until(
  taskflow, [](){ return rand()%10 == 0 }
);
// do something else
future.wait();

This member function is thread-safe.

Attention

The executor does not own the given taskflow. It is your responsibility to ensure the taskflow remains alive during its execution.

template<typename P>

tf::Future<void> tf::Executor::run_until(Taskflow&& taskflow, P&& pred)

runs a moved taskflow and keeps running it until the predicate becomes true

This member function executes a moved taskflow multiple times until the predicate returns true. This member function returns a tf::Future object that eventually holds the result of the execution. The executor will take care of the lifetime of the moved taskflow.

tf::Future<void> future = executor.run_until(
  std::move(taskflow), [](){ return rand()%10 == 0 }
);
// do something else
future.wait();

This member function is thread-safe.

template<typename P, typename C>

tf::Future<void> tf::Executor::run_until(Taskflow& taskflow, P&& pred, C&& callable)

runs a taskflow multiple times until the predicate becomes true and then invokes the callback

This member function executes the given taskflow multiple times until the predicate returns true and then invokes the given callable when the execution completes. This member function returns a tf::Future object that eventually holds the result of the execution.

tf::Future<void> future = executor.run_until(
  taskflow, [](){ return rand()%10 == 0 }, [](){ std::cout << "done"; }
);
// do something else
future.wait();

This member function is thread-safe.

Attention

The executor does not own the given taskflow. It is your responsibility to ensure the taskflow remains alive during its execution.

template<typename P, typename C>

tf::Future<void> tf::Executor::run_until(Taskflow&& taskflow, P&& pred, C&& callable)

runs a moved taskflow and keeps running it until the predicate becomes true and then invokes the callback

This member function executes a moved taskflow multiple times until the predicate returns true and then invokes the given callable when the execution completes. This member function returns a tf::Future object that eventually holds the result of the execution. The executor will take care of the lifetime of the moved taskflow.

tf::Future<void> future = executor.run_until(
  std::move(taskflow),
  [](){ return rand()%10 == 0 }, [](){ std::cout << "done"; }
);
// do something else
future.wait();

This member function is thread-safe.

template<typename T>

void tf::Executor::corun(T& target)

runs a target graph and waits until it completes using an internal worker of this executor

The method runs a target graph which has tf::Graph& T::graph() defined and waits until the execution completes. Unlike the typical flow of calling tf::Executor::run series plus waiting on the result, this method must be called by an internal worker of this executor. The caller worker will participate in the work-stealing loop of the scheduler, thereby avoiding potential deadlock caused by blocked waiting.

tf::Executor executor(2);
tf::Taskflow taskflow;
std::array<tf::Taskflow, 1000> others;

std::atomic<size_t> counter{0};

for(size_t n=0; n<1000; n++) {
  for(size_t i=0; i<1000; i++) {
    others[n].emplace([&](){ counter++; });
  }
  taskflow.emplace([&executor, &tf=others[n]](){
    executor.corun(tf);
    //executor.run(tf).wait();  <- blocking the worker without doing anything
    //                             will introduce deadlock
  });
}
executor.run(taskflow).wait();

The method is thread-safe as long as the target is not concurrently ran by two or more threads.

Attention

You must call tf::Executor::corun from a worker of the calling executor or an exception will be thrown.

template<typename P>

void tf::Executor::corun_until(P&& predicate)

keeps running the work-stealing loop until the predicate becomes true

The method keeps the caller worker running in the work-stealing loop until the stop predicate becomes true.

taskflow.emplace([&](){
  std::future<void> fu = std::async([](){ std::sleep(100s); });
  executor.corun_until([](){
    return fu.wait_for(std::chrono::seconds(0)) == future_status::ready;
  });
});

Attention

You must call tf::Executor::corun_until from a worker of the calling executor or an exception will be thrown.

void tf::Executor::wait_for_all()

waits for all tasks to complete

This member function waits until all submitted tasks (e.g., taskflows, asynchronous tasks) to finish.

executor.run(taskflow1);
executor.run_n(taskflow2, 10);
executor.run_n(taskflow3, 100);
executor.wait_for_all();  // wait until the above submitted taskflows finish

size_t tf::Executor::num_workers() const noexcept

queries the number of worker threads

Each worker represents one unique thread spawned by an executor upon its construction time.

tf::Executor executor(4);
std::cout << executor.num_workers();    // 4

size_t tf::Executor::num_topologies() const

queries the number of running topologies at the time of this call

When a taskflow is submitted to an executor, a topology is created to store runtime metadata of the running taskflow. When the execution of the submitted taskflow finishes, its corresponding topology will be removed from the executor.

executor.run(taskflow);
std::cout << executor.num_topologies();  // 0 or 1 (taskflow still running)

size_t tf::Executor::num_taskflows() const

queries the number of running taskflows with moved ownership

executor.run(std::move(taskflow));
std::cout << executor.num_taskflows();  // 0 or 1 (taskflow still running)

int tf::Executor::this_worker_id() const

queries the id of the caller thread within this executor

Each worker has an unique id in the range of 0 to N-1 associated with its parent executor. If the caller thread does not belong to the executor, -1 is returned.

tf::Executor executor(4);   // 4 workers in the executor
executor.this_worker_id();  // -1 (main thread is not a worker)

taskflow.emplace([&](){
  std::cout << executor.this_worker_id();  // 0, 1, 2, or 3
});
executor.run(taskflow);

template<typename Observer, typename... ArgsT>

std::shared_ptr<Observer> tf::Executor::make_observer(ArgsT && ... args)

constructs an observer to inspect the activities of worker threads

Each executor manages a list of observers with shared ownership with callers. For each of these observers, the two member functions, tf::ObserverInterface::on_entry and tf::ObserverInterface::on_exit will be called before and after the execution of a task.

This member function is not thread-safe.

template<typename Observer>

void tf::Executor::remove_observer(std::shared_ptr<Observer> observer)

removes an observer from the executor

This member function is not thread-safe.

template<typename P, typename F>

auto tf::Executor::async(P&& params, F&& func)

creates a parameterized asynchronous task to run the given function

The method creates a parameterized asynchronous task to run the given function and return a std::future object that eventually will hold the result of the execution.

std::future<int> future = executor.async("name", [](){
  std::cout << "create an asynchronous task with a name and returns 1\n";
  return 1;
});
future.get();

This member function is thread-safe.

template<typename F>

auto tf::Executor::async(F&& func)

runs a given function asynchronously

The method creates an asynchronous task to run the given function and return a std::future object that eventually will hold the result of the return value.

std::future<int> future = executor.async([](){
  std::cout << "create an asynchronous task and returns 1\n";
  return 1;
});
future.get();

This member function is thread-safe.

template<typename P, typename F>

void tf::Executor::silent_async(P&& params, F&& func)

similar to tf::Executor::async but does not return a future object

The method creates a parameterized asynchronous task to run the given function without returning any std::future object. This member function is more efficient than tf::Executor::async and is encouraged to use when applications do not need a std::future to acquire the result or synchronize the execution.

executor.silent_async("name", [](){
  std::cout << "create an asynchronous task with a name and no return\n";
});
executor.wait_for_all();

This member function is thread-safe.

template<typename F>

void tf::Executor::silent_async(F&& func)

similar to tf::Executor::async but does not return a future object

The method creates an asynchronous task to run the given function without returning any std::future object. This member function is more efficient than tf::Executor::async and is encouraged to use when applications do not need a std::future to acquire the result or synchronize the execution.

executor.silent_async([](){
  std::cout << "create an asynchronous task with no return\n";
});
executor.wait_for_all();

This member function is thread-safe.

template<typename F, typename... Tasks, std::enable_if_t<all_same_v<AsyncTask, std::decay_t<Tasks>...>, void>* = nullptr>

tf::AsyncTask tf::Executor::silent_dependent_async(F&& func, Tasks && ... tasks)

runs the given function asynchronously when the given predecessors finish

This member function is more efficient than tf::Executor::dependent_async and is encouraged to use when you do not want a std::future to acquire the result or synchronize the execution. The example below creates three asynchronous tasks, A, B, and C, in which task C runs after task A and task B.

tf::AsyncTask A = executor.silent_dependent_async([](){ printf("A\n"); });
tf::AsyncTask B = executor.silent_dependent_async([](){ printf("B\n"); });
executor.silent_dependent_async([](){ printf("C runs after A and B\n"); }, A, B);
executor.wait_for_all();

This member function is thread-safe.

template<typename P, typename F, typename... Tasks, std::enable_if_t<is_task_params_v<P> && all_same_v<AsyncTask, std::decay_t<Tasks>...>, void>* = nullptr>

tf::AsyncTask tf::Executor::silent_dependent_async(P&& params, F&& func, Tasks && ... tasks)

runs the given function asynchronously when the given predecessors finish

This member function is more efficient than tf::Executor::dependent_async and is encouraged to use when you do not want a std::future to acquire the result or synchronize the execution. The example below creates three asynchronous tasks, A, B, and C, in which task C runs after task A and task B. Assigned task names will appear in the observers of the executor.

tf::AsyncTask A = executor.silent_dependent_async("A", [](){ printf("A\n"); });
tf::AsyncTask B = executor.silent_dependent_async("B", [](){ printf("B\n"); });
executor.silent_dependent_async(
  "C", [](){ printf("C runs after A and B\n"); }, A, B
);
executor.wait_for_all();

This member function is thread-safe.

template<typename F, typename I, std::enable_if_t<!std::is_same_v<std::decay_t<I>, AsyncTask>, void>* = nullptr>

tf::AsyncTask tf::Executor::silent_dependent_async(F&& func, I first, I last)

runs the given function asynchronously when the given range of predecessors finish

This member function is more efficient than tf::Executor::dependent_async and is encouraged to use when you do not want a std::future to acquire the result or synchronize the execution. The example below creates three asynchronous tasks, A, B, and C, in which task C runs after task A and task B.

std::array<tf::AsyncTask, 2> array {
  executor.silent_dependent_async([](){ printf("A\n"); }),
  executor.silent_dependent_async([](){ printf("B\n"); })
};
executor.silent_dependent_async(
  [](){ printf("C runs after A and B\n"); }, array.begin(), array.end()
);
executor.wait_for_all();

This member function is thread-safe.

template<typename P, typename F, typename I, std::enable_if_t<is_task_params_v<P> && !std::is_same_v<std::decay_t<I>, AsyncTask>, void>* = nullptr>

tf::AsyncTask tf::Executor::silent_dependent_async(P&& params, F&& func, I first, I last)

runs the given function asynchronously when the given range of predecessors finish

This member function is more efficient than tf::Executor::dependent_async and is encouraged to use when you do not want a std::future to acquire the result or synchronize the execution. The example below creates three asynchronous tasks, A, B, and C, in which task C runs after task A and task B. Assigned task names will appear in the observers of the executor.

std::array<tf::AsyncTask, 2> array {
  executor.silent_dependent_async("A", [](){ printf("A\n"); }),
  executor.silent_dependent_async("B", [](){ printf("B\n"); })
};
executor.silent_dependent_async(
  "C", [](){ printf("C runs after A and B\n"); }, array.begin(), array.end()
);
executor.wait_for_all();

This member function is thread-safe.

template<typename F, typename... Tasks, std::enable_if_t<all_same_v<AsyncTask, std::decay_t<Tasks>...>, void>* = nullptr>

auto tf::Executor::dependent_async(F&& func, Tasks && ... tasks)

runs the given function asynchronously when the given predecessors finish

The example below creates three asynchronous tasks, A, B, and C, in which task C runs after task A and task B. Task C returns a pair of its tf::AsyncTask handle and a std::future<int> that eventually will hold the result of the execution.

tf::AsyncTask A = executor.silent_dependent_async([](){ printf("A\n"); });
tf::AsyncTask B = executor.silent_dependent_async([](){ printf("B\n"); });
auto [C, fuC] = executor.dependent_async(
  [](){ 
    printf("C runs after A and B\n"); 
    return 1;
  }, 
  A, B
);
fuC.get();  // C finishes, which in turns means both A and B finish

You can mixed the use of tf::AsyncTask handles returned by Executor::dependent_async and Executor::silent_dependent_async when specifying task dependencies.

This member function is thread-safe.

template<typename P, typename F, typename... Tasks, std::enable_if_t<is_task_params_v<P> && all_same_v<AsyncTask, std::decay_t<Tasks>...>, void>* = nullptr>

auto tf::Executor::dependent_async(P&& params, F&& func, Tasks && ... tasks)

runs the given function asynchronously when the given predecessors finish

The example below creates three named asynchronous tasks, A, B, and C, in which task C runs after task A and task B. Task C returns a pair of its tf::AsyncTask handle and a std::future<int> that eventually will hold the result of the execution. Assigned task names will appear in the observers of the executor.

tf::AsyncTask A = executor.silent_dependent_async("A", [](){ printf("A\n"); });
tf::AsyncTask B = executor.silent_dependent_async("B", [](){ printf("B\n"); });
auto [C, fuC] = executor.dependent_async(
  "C",
  [](){ 
    printf("C runs after A and B\n"); 
    return 1;
  }, 
  A, B
);
assert(fuC.get()==1);  // C finishes, which in turns means both A and B finish

You can mixed the use of tf::AsyncTask handles returned by Executor::dependent_async and Executor::silent_dependent_async when specifying task dependencies.

This member function is thread-safe.

template<typename F, typename I, std::enable_if_t<!std::is_same_v<std::decay_t<I>, AsyncTask>, void>* = nullptr>

auto tf::Executor::dependent_async(F&& func, I first, I last)

runs the given function asynchronously when the given range of predecessors finish

The example below creates three asynchronous tasks, A, B, and C, in which task C runs after task A and task B. Task C returns a pair of its tf::AsyncTask handle and a std::future<int> that eventually will hold the result of the execution.

std::array<tf::AsyncTask, 2> array {
  executor.silent_dependent_async([](){ printf("A\n"); }),
  executor.silent_dependent_async([](){ printf("B\n"); })
};
auto [C, fuC] = executor.dependent_async(
  [](){ 
    printf("C runs after A and B\n"); 
    return 1;
  }, 
  array.begin(), array.end()
);
assert(fuC.get()==1);  // C finishes, which in turns means both A and B finish

You can mixed the use of tf::AsyncTask handles returned by Executor::dependent_async and Executor::silent_dependent_async when specifying task dependencies.

This member function is thread-safe.

template<typename P, typename F, typename I, std::enable_if_t<is_task_params_v<P> && !std::is_same_v<std::decay_t<I>, AsyncTask>, void>* = nullptr>

auto tf::Executor::dependent_async(P&& params, F&& func, I first, I last)

runs the given function asynchronously when the given range of predecessors finish

The example below creates three named asynchronous tasks, A, B, and C, in which task C runs after task A and task B. Task C returns a pair of its tf::AsyncTask handle and a std::future<int> that eventually will hold the result of the execution. Assigned task names will appear in the observers of the executor.

std::array<tf::AsyncTask, 2> array {
  executor.silent_dependent_async("A", [](){ printf("A\n"); }),
  executor.silent_dependent_async("B", [](){ printf("B\n"); })
};
auto [C, fuC] = executor.dependent_async(
  "C",
  [](){ 
    printf("C runs after A and B\n"); 
    return 1;
  }, 
  array.begin(), array.end()
);
assert(fuC.get()==1);  // C finishes, which in turns means both A and B finish

You can mixed the use of tf::AsyncTask handles returned by Executor::dependent_async and Executor::silent_dependent_async when specifying task dependencies.

This member function is thread-safe.