Migrating from low-level task API¶
The low-level task API of Intel(R) Threading Building Blocks (TBB) was considered complex and hence error-prone, which was the primary reason it had been removed from oneAPI Threading Building Blocks (oneTBB). This guide helps with the migration from TBB to oneTBB for the use cases where low-level task API is used.
Spawning of individual tasks¶
For most use cases, the spawning of individual tasks can be replaced with the use of either
oneapi::tbb::task_group
or oneapi::tbb::parallel_invoke
.
For example, RootTask
, ChildTask1
, and ChildTask2
are the user-side functors that
inherit tbb::task
and implement its interface. Then spawning of ChildTask1
and
ChildTask2
tasks that can execute in parallel with each other and waiting on the RootTask
is
implemented as:
#include <tbb/task.h>
int main() {
// Assuming RootTask, ChildTask1, and ChildTask2 are defined.
RootTask& root = *new(tbb::task::allocate_root()) RootTask{};
ChildTask1& child1 = *new(root.allocate_child()) ChildTask1{/*params*/};
ChildTask2& child2 = *new(root.allocate_child()) ChildTask2{/*params*/};
root.set_ref_count(3);
tbb::task::spawn(child1);
tbb::task::spawn(child2);
root.wait_for_all();
}
Using oneapi::tbb::task_group
¶
The code above can be rewritten using oneapi::tbb::task_group
:
#include <oneapi/tbb/task_group.h>
int main() {
// Assuming ChildTask1, and ChildTask2 are defined.
oneapi::tbb::task_group tg;
tg.run(ChildTask1{/*params*/});
tg.run(ChildTask2{/*params*/});
tg.wait();
}
The code looks more concise now. It also enables lambda functions and does not require you to
implement tbb::task
interface that overrides the tbb::task* tbb::task::execute()
virtual
method. With this new approach, you work with functors in a C++-standard way by implementing void
operator() const
:
struct Functor {
// Member to be called when object of this type are passed into
// oneapi::tbb::task_group::run() method
void operator()() const {}
};
Using oneapi::tbb::parallel_invoke
¶
It is also possible to use oneapi::tbb::parallel_invoke
to rewrite the original code and make it
even more concise:
#include <oneapi/tbb/parallel_invoke.h>
int main() {
// Assuming ChildTask1, and ChildTask2 are defined.
oneapi::tbb::parallel_invoke(
ChildTask1{/*params*/},
ChildTask2{/*params*/}
);
}
Adding more work during task execution¶
oneapi::tbb::parallel_invoke
follows a blocking style of programming, which means that it
completes only when all functors passed to the parallel pattern complete their execution.
In TBB, cases when the amount of work is not known in advance and the work needs to be added during
the execution of a parallel algorithm were mostly covered by tbb::parallel_do
high-level
parallel pattern. The tbb::parallel_do
algorithm logic may be implemented using the task API as:
#include <cstddef>
#include <vector>
#include <tbb/task.h>
// Assuming RootTask and OtherWork are defined and implement tbb::task interface.
struct Task : public tbb::task {
Task(tbb::task& root, int i)
: m_root(root), m_i(i)
{}
tbb::task* execute() override {
// ... do some work for item m_i ...
if (add_more_parallel_work) {
tbb::task& child = *new(m_root.allocate_child()) OtherWork;
tbb::task::spawn(child);
}
return nullptr;
}
tbb::task& m_root;
int m_i;
};
int main() {
std::vector<int> items = { 0, 1, 2, 3, 4, 5, 6, 7 };
RootTask& root = *new(tbb::task::allocate_root()) RootTask{/*params*/};
root.set_ref_count(items.size() + 1);
for (std::size_t i = 0; i < items.size(); ++i) {
Task& task = *new(root.allocate_child()) Task(root, items[i]);
tbb::task::spawn(task);
}
root.wait_for_all();
return 0;
}
In oneTBB tbb::parallel_do
interface was removed. Instead, the functionality of adding new work
was included into the oneapi::tbb::parallel_for_each
interface.
The previous use case can be rewritten in oneTBB as follows:
#include <vector>
#include <oneapi/tbb/parallel_for_each.h>
int main() {
std::vector<int> items = { 0, 1, 2, 3, 4, 5, 6, 7 };
oneapi::tbb::parallel_for_each(
items.begin(), items.end(),
[](int& i, tbb::feeder<int>& feeder) {
// ... do some work for item i ...
if (add_more_parallel_work)
feeder.add(i);
}
);
}
Since both TBB and oneTBB support nested expressions, you can run additional functors from within an already running functor.
The previous use case can be rewritten using oneapi::tbb::task_group
as:
#include <cstddef>
#include <vector>
#include <oneapi/tbb/task_group.h>
int main() {
std::vector<int> items = { 0, 1, 2, 3, 4, 5, 6, 7 };
oneapi::tbb::task_group tg;
for (std::size_t i = 0; i < items.size(); ++i) {
tg.run([&i = items[i], &tg] {
// ... do some work for item i ...
if (add_more_parallel_work)
// Assuming OtherWork is defined.
tg.run(OtherWork{});
});
}
tg.wait();
}
Task recycling¶
You can re-run the functor by passing *this
to the oneapi::tbb::task_group::run()
method. The functor will be copied in this case. However, its state can be shared among instances:
#include <memory>
#include <oneapi/tbb/task_group.h>
struct SharedStateFunctor {
std::shared_ptr<Data> m_shared_data;
oneapi::tbb::task_group& m_task_group;
void operator()() const {
// do some work processing m_shared_data
if (has_more_work)
m_task_group.run(*this);
// Note that this might be concurrently accessing m_shared_data already
}
};
int main() {
// Assuming Data is defined.
std::shared_ptr<Data> data = std::make_shared<Data>(/*params*/);
oneapi::tbb::task_group tg;
tg.run(SharedStateFunctor{data, tg});
tg.wait();
}
Such patterns are particularly useful when the work within a functor is not completed but there is a need for the task scheduler to react to outer circumstances, such as cancellation of group execution. To avoid issues with concurrent access, it is recommended to submit it for re-execution as the last step:
#include <memory>
#include <oneapi/tbb/task_group.h>
struct SharedStateFunctor {
std::shared_ptr<Data> m_shared_data;
oneapi::tbb::task_group& m_task_group;
void operator()() const {
// do some work processing m_shared_data
if (need_to_yield) {
m_task_group.run(*this);
return;
}
}
};
int main() {
// Assuming Data is defined.
std::shared_ptr<Data> data = std::make_shared<Data>(/*params*/);
oneapi::tbb::task_group tg;
tg.run(SharedStateFunctor{data, tg});
tg.wait();
}
Recycling as child or continuation¶
In oneTBB this kind of recycling is done manually. You have to track when it is time to run the task:
#include <cstddef>
#include <vector>
#include <atomic>
#include <cassert>
#include <oneapi/tbb/task_group.h>
struct ContinuationTask {
ContinuationTask(std::vector<int>& data, int& result)
: m_data(data), m_result(result)
{}
void operator()() const {
for (const auto& item : m_data)
m_result += item;
}
std::vector<int>& m_data;
int& m_result;
};
struct ChildTask {
ChildTask(std::vector<int>& data, int& result,
std::atomic<std::size_t>& tasks_left, std::atomic<std::size_t>& tasks_done,
oneapi::tbb::task_group& tg)
: m_data(data), m_result(result), m_tasks_left(tasks_left), m_tasks_done(tasks_done), m_tg(tg)
{}
void operator()() const {
std::size_t index = --m_tasks_left;
m_data[index] = produce_item_for(index);
std::size_t done_num = ++m_tasks_done;
if (index % 2 != 0) {
// Recycling as child
m_tg.run(*this);
return;
} else if (done_num == m_data.size()) {
assert(m_tasks_left == 0);
// Spawning a continuation that does reduction
m_tg.run(ContinuationTask(m_data, m_result));
}
}
std::vector<int>& m_data;
int& m_result;
std::atomic<std::size_t>& m_tasks_left;
std::atomic<std::size_t>& m_tasks_done;
oneapi::tbb::task_group& m_tg;
};
int main() {
int result = 0;
std::vector<int> items(10, 0);
std::atomic<std::size_t> tasks_left{items.size()};
std::atomic<std::size_t> tasks_done{0};
oneapi::tbb::task_group tg;
for (std::size_t i = 0; i < items.size(); i+=2) {
tg.run(ChildTask(items, result, tasks_left, tasks_done, tg));
}
tg.wait();
}