Edit: /usr/include/oneapi/tbb/task_group.h (27329B)
/*
Copyright (c) 2005-2021 Intel Corporation
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
#ifndef __TBB_task_group_H
#define __TBB_task_group_H
#include "detail/_config.h"
#include "detail/_namespace_injection.h"
#include "detail/_assert.h"
#include "detail/_utils.h"
#include "detail/_template_helpers.h"
#include "detail/_exception.h"
#include "detail/_task.h"
#include "detail/_small_object_pool.h"
#include "detail/_intrusive_list_node.h"
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
#include "detail/_task_handle.h"
#endif
#include "profiling.h"
#include
#if _MSC_VER && !defined(__INTEL_COMPILER)
// Suppress warning: structure was padded due to alignment specifier
#pragma warning(push)
#pragma warning(disable:4324)
#endif
namespace tbb {
namespace detail {
namespace d1 {
class delegate_base;
class task_arena_base;
class task_group_context;
class task_group_base;
}
namespace r1 {
// Forward declarations
class tbb_exception_ptr;
class market;
class thread_data;
class task_dispatcher;
template
class context_guard_helper;
struct task_arena_impl;
class context_list;
TBB_EXPORT void __TBB_EXPORTED_FUNC execute(d1::task_arena_base&, d1::delegate_base&);
TBB_EXPORT void __TBB_EXPORTED_FUNC isolate_within_arena(d1::delegate_base&, std::intptr_t);
TBB_EXPORT void __TBB_EXPORTED_FUNC initialize(d1::task_group_context&);
TBB_EXPORT void __TBB_EXPORTED_FUNC destroy(d1::task_group_context&);
TBB_EXPORT void __TBB_EXPORTED_FUNC reset(d1::task_group_context&);
TBB_EXPORT bool __TBB_EXPORTED_FUNC cancel_group_execution(d1::task_group_context&);
TBB_EXPORT bool __TBB_EXPORTED_FUNC is_group_execution_cancelled(d1::task_group_context&);
TBB_EXPORT void __TBB_EXPORTED_FUNC capture_fp_settings(d1::task_group_context&);
struct task_group_context_impl;
}
namespace d2 {
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
namespace {
template
d1::task* task_ptr_or_nullptr(F&& f);
}
template
class function_task : public task_handle_task {
//TODO: apply empty base optimization here
const F m_func;
private:
d1::task* execute(d1::execution_data& ed) override {
__TBB_ASSERT(ed.context == &this->ctx(), "The task group context should be used for all tasks");
task* res = task_ptr_or_nullptr(m_func);
finalize(&ed);
return res;
}
d1::task* cancel(d1::execution_data& ed) override {
finalize(&ed);
return nullptr;
}
public:
template
function_task(FF&& f, d1::wait_context& wo, d1::task_group_context& ctx, d1::small_object_allocator& alloc)
: task_handle_task{wo, ctx, alloc},
m_func(std::forward(f)) {}
};
namespace {
template
d1::task* task_ptr_or_nullptr_impl(std::false_type, F&& f){
task_handle th = std::forward(f)();
return task_handle_accessor::release(th);
}
template
d1::task* task_ptr_or_nullptr_impl(std::true_type, F&& f){
std::forward(f)();
return nullptr;
}
template
d1::task* task_ptr_or_nullptr(F&& f){
using is_void_t = std::is_void<
decltype(std::forward(f)())
>;
return task_ptr_or_nullptr_impl(is_void_t{}, std::forward(f));
}
}
#else
namespace {
template
d1::task* task_ptr_or_nullptr(F&& f){
std::forward(f)();
return nullptr;
}
} // namespace
#endif // __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
} // namespace d2
namespace d1 {
// This structure is left here for backward compatibility check
struct context_list_node {
std::atomic prev{};
std::atomic next{};
};
//! Used to form groups of tasks
/** @ingroup task_scheduling
The context services explicit cancellation requests from user code, and unhandled
exceptions intercepted during tasks execution. Intercepting an exception results
in generating internal cancellation requests (which is processed in exactly the
same way as external ones).
The context is associated with one or more root tasks and defines the cancellation
group that includes all the descendants of the corresponding root task(s). Association
is established when a context object is passed as an argument to the task::allocate_root()
method. See task_group_context::task_group_context for more details.
The context can be bound to another one, and other contexts can be bound to it,
forming a tree-like structure: parent -> this -> children. Arrows here designate
cancellation propagation direction. If a task in a cancellation group is cancelled
all the other tasks in this group and groups bound to it (as children) get cancelled too.
**/
class task_group_context : no_copy {
public:
enum traits_type {
fp_settings = 1 << 1,
concurrent_wait = 1 << 2,
default_traits = 0
};
enum kind_type {
isolated,
bound
};
private:
//! Space for platform-specific FPU settings.
/** Must only be accessed inside TBB binaries, and never directly in user
code or inline methods. */
std::uint64_t my_cpu_ctl_env;
//! Specifies whether cancellation was requested for this task group.
std::atomic my_cancellation_requested;
//! Versioning for run-time checks and behavioral traits of the context.
enum class task_group_context_version : std::uint8_t {
gold_2021U1 = 0, // version of task_group_context released in oneTBB 2021.1 GOLD
proxy_support = 1 // backward compatible support for 'this' context to act as a proxy
};
task_group_context_version my_version;
//! The context traits.
struct context_traits {
bool fp_settings : 1;
bool concurrent_wait : 1;
bool bound : 1;
bool proxy : 1; // true if 'this' acts as a proxy for user-specified context
bool reserved1 : 1;
bool reserved2 : 1;
bool reserved3 : 1;
bool reserved4 : 1;
} my_traits;
static_assert(sizeof(context_traits) == 1, "Traits shall fit into one byte.");
static constexpr std::uint8_t may_have_children = 1;
//! The context internal state (currently only may_have_children).
std::atomic my_state;
enum class lifetime_state : std::uint8_t {
created,
locked,
isolated,
bound,
dead
};
//! The synchronization machine state to manage lifetime.
std::atomic my_lifetime_state;
union {
//! Pointer to the context of the parent cancellation group. NULL for isolated contexts.
task_group_context* my_parent;
//! Pointer to the actual context 'this' context represents a proxy of.
task_group_context* my_actual_context;
};
//! Thread data instance that registered this context in its list.
r1::context_list* my_context_list;
static_assert(sizeof(std::atomic) == sizeof(r1::context_list*), "To preserve backward compatibility these types should have the same size");
//! Used to form the thread specific list of contexts without additional memory allocation.
/** A context is included into the list of the current thread when its binding to
its parent happens. Any context can be present in the list of one thread only. **/
intrusive_list_node my_node;
static_assert(sizeof(intrusive_list_node) == sizeof(context_list_node), "To preserve backward compatibility these types should have the same size");
//! Pointer to the container storing exception being propagated across this task group.
std::atomic my_exception;
static_assert(sizeof(std::atomic) == sizeof(r1::tbb_exception_ptr*),
"backward compatibility check");
//! Used to set and maintain stack stitching point for Intel Performance Tools.
void* my_itt_caller;
//! Description of algorithm for scheduler based instrumentation.
string_resource_index my_name;
char padding[max_nfs_size
- sizeof(std::uint64_t) // my_cpu_ctl_env
- sizeof(std::atomic) // my_cancellation_requested
- sizeof(std::uint8_t) // my_version
- sizeof(context_traits) // my_traits
- sizeof(std::atomic) // my_state
- sizeof(std::atomic) // my_lifetime_state
- sizeof(task_group_context*) // my_parent
- sizeof(r1::context_list*) // my_context_list
- sizeof(intrusive_list_node) // my_node
- sizeof(std::atomic) // my_exception
- sizeof(void*) // my_itt_caller
- sizeof(string_resource_index) // my_name
];
task_group_context(context_traits t, string_resource_index name)
: my_version{task_group_context_version::proxy_support}, my_name{name}
{
my_traits = t; // GCC4.8 issues warning list initialization for bitset (missing-field-initializers)
r1::initialize(*this);
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
task_group_context(task_group_context* actual_context)
: my_version{task_group_context_version::proxy_support}
, my_actual_context{actual_context}
{
__TBB_ASSERT(my_actual_context, "Passed pointer value points to nothing.");
my_traits.proxy = true;
my_name = actual_context->my_name;
// no need to initialize 'this' context as it acts as a proxy for my_actual_context, which
// initialization is a user-side responsibility.
}
#endif
static context_traits make_traits(kind_type relation_with_parent, std::uintptr_t user_traits) {
context_traits ct;
ct.fp_settings = (user_traits & fp_settings) == fp_settings;
ct.concurrent_wait = (user_traits & concurrent_wait) == concurrent_wait;
ct.bound = relation_with_parent == bound;
ct.proxy = false;
ct.reserved1 = ct.reserved2 = ct.reserved3 = ct.reserved4 = false;
return ct;
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
bool is_proxy() const {
return my_version >= task_group_context_version::proxy_support && my_traits.proxy;
}
#endif
task_group_context& actual_context() noexcept {
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
if (is_proxy()) {
__TBB_ASSERT(my_actual_context, "Actual task_group_context is not set.");
return *my_actual_context;
}
#endif
return *this;
}
const task_group_context& actual_context() const noexcept {
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
if (is_proxy()) {
__TBB_ASSERT(my_actual_context, "Actual task_group_context is not set.");
return *my_actual_context;
}
#endif
return *this;
}
public:
//! Default & binding constructor.
/** By default a bound context is created. That is this context will be bound
(as child) to the context of the currently executing task . Cancellation
requests passed to the parent context are propagated to all the contexts
bound to it. Similarly priority change is propagated from the parent context
to its children.
If task_group_context::isolated is used as the argument, then the tasks associated
with this context will never be affected by events in any other context.
Creating isolated contexts involve much less overhead, but they have limited
utility. Normally when an exception occurs in an algorithm that has nested
ones running, it is desirably to have all the nested algorithms cancelled
as well. Such a behavior requires nested algorithms to use bound contexts.
There is one good place where using isolated algorithms is beneficial. It is
an external thread. That is if a particular algorithm is invoked directly from
the external thread (not from a TBB task), supplying it with explicitly
created isolated context will result in a faster algorithm startup.
VERSIONING NOTE:
Implementation(s) of task_group_context constructor(s) cannot be made
entirely out-of-line because the run-time version must be set by the user
code. This will become critically important for binary compatibility, if
we ever have to change the size of the context object. **/
task_group_context(kind_type relation_with_parent = bound,
std::uintptr_t t = default_traits)
: task_group_context(make_traits(relation_with_parent, t), CUSTOM_CTX) {}
// Custom constructor for instrumentation of oneTBB algorithm
task_group_context(string_resource_index name )
: task_group_context(make_traits(bound, default_traits), name) {}
// Do not introduce any logic on user side since it might break state propagation assumptions
~task_group_context() {
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
// When 'this' serves as a proxy, the initialization does not happen - nor should the
// destruction.
if (!is_proxy())
#endif
{
r1::destroy(*this);
}
}
//! Forcefully reinitializes the context after the task tree it was associated with is completed.
/** Because the method assumes that all the tasks that used to be associated with
this context have already finished, calling it while the context is still
in use somewhere in the task hierarchy leads to undefined behavior.
IMPORTANT: This method is not thread safe!
The method does not change the context's parent if it is set. **/
void reset() {
r1::reset(actual_context());
}
//! Initiates cancellation of all tasks in this cancellation group and its subordinate groups.
/** \return false if cancellation has already been requested, true otherwise.
Note that canceling never fails. When false is returned, it just means that
another thread (or this one) has already sent cancellation request to this
context or to one of its ancestors (if this context is bound). It is guaranteed
that when this method is concurrently called on the same not yet cancelled
context, true will be returned by one and only one invocation. **/
bool cancel_group_execution() {
return r1::cancel_group_execution(actual_context());
}
//! Returns true if the context received cancellation request.
bool is_group_execution_cancelled() {
return r1::is_group_execution_cancelled(actual_context());
}
#if __TBB_FP_CONTEXT
//! Captures the current FPU control settings to the context.
/** Because the method assumes that all the tasks that used to be associated with
this context have already finished, calling it while the context is still
in use somewhere in the task hierarchy leads to undefined behavior.
IMPORTANT: This method is not thread safe!
The method does not change the FPU control settings of the context's parent. **/
void capture_fp_settings() {
r1::capture_fp_settings(actual_context());
}
#endif
//! Returns the user visible context trait
std::uintptr_t traits() const {
std::uintptr_t t{};
const task_group_context& ctx = actual_context();
t |= ctx.my_traits.fp_settings ? fp_settings : 0;
t |= ctx.my_traits.concurrent_wait ? concurrent_wait : 0;
return t;
}
private:
//// TODO: cleanup friends
friend class r1::market;
friend class r1::thread_data;
friend class r1::task_dispatcher;
template
friend class r1::context_guard_helper;
friend struct r1::task_arena_impl;
friend struct r1::task_group_context_impl;
friend class task_group_base;
}; // class task_group_context
static_assert(sizeof(task_group_context) == 128, "Wrong size of task_group_context");
enum task_group_status {
not_complete,
complete,
canceled
};
class task_group;
class structured_task_group;
#if TBB_PREVIEW_ISOLATED_TASK_GROUP
class isolated_task_group;
#endif
template
class function_task : public task {
const F m_func;
wait_context& m_wait_ctx;
small_object_allocator m_allocator;
void finalize(const execution_data& ed) {
// Make a local reference not to access this after destruction.
wait_context& wo = m_wait_ctx;
// Copy allocator to the stack
auto allocator = m_allocator;
// Destroy user functor before release wait.
this->~function_task();
wo.release();
allocator.deallocate(this, ed);
}
task* execute(execution_data& ed) override {
task* res = d2::task_ptr_or_nullptr(m_func);
finalize(ed);
return res;
}
task* cancel(execution_data& ed) override {
finalize(ed);
return nullptr;
}
public:
function_task(const F& f, wait_context& wo, small_object_allocator& alloc)
: m_func(f)
, m_wait_ctx(wo)
, m_allocator(alloc) {}
function_task(F&& f, wait_context& wo, small_object_allocator& alloc)
: m_func(std::move(f))
, m_wait_ctx(wo)
, m_allocator(alloc) {}
};
template
class function_stack_task : public task {
const F& m_func;
wait_context& m_wait_ctx;
void finalize() {
m_wait_ctx.release();
}
task* execute(execution_data&) override {
task* res = d2::task_ptr_or_nullptr(m_func);
finalize();
return res;
}
task* cancel(execution_data&) override {
finalize();
return nullptr;
}
public:
function_stack_task(const F& f, wait_context& wo) : m_func(f), m_wait_ctx(wo) {}
};
class task_group_base : no_copy {
protected:
wait_context m_wait_ctx;
task_group_context m_context;
template
task_group_status internal_run_and_wait(const F& f) {
function_stack_task t{ f, m_wait_ctx };
m_wait_ctx.reserve();
bool cancellation_status = false;
try_call([&] {
execute_and_wait(t, context(), m_wait_ctx, context());
}).on_completion([&] {
// TODO: the reset method is not thread-safe. Ensure the correct behavior.
cancellation_status = context().is_group_execution_cancelled();
context().reset();
});
return cancellation_status ? canceled : complete;
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
task_group_status internal_run_and_wait(d2::task_handle&& h) {
if (h == nullptr) {
throw_exception(exception_id::bad_task_handle);
}
using acs = d2::task_handle_accessor;
if (&acs::ctx_of(h) != &context()) {
throw_exception(exception_id::bad_task_handle_wrong_task_group);
}
bool cancellation_status = false;
try_call([&] {
execute_and_wait(*acs::release(h), context(), m_wait_ctx, context());
}).on_completion([&] {
// TODO: the reset method is not thread-safe. Ensure the correct behavior.
cancellation_status = context().is_group_execution_cancelled();
context().reset();
});
return cancellation_status ? canceled : complete;
}
#endif
template
task* prepare_task(F&& f) {
m_wait_ctx.reserve();
small_object_allocator alloc{};
return alloc.new_object::type>>(std::forward(f), m_wait_ctx, alloc);
}
task_group_context& context() noexcept {
return m_context.actual_context();
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
template
d2::task_handle prepare_task_handle(F&& f) {
m_wait_ctx.reserve();
small_object_allocator alloc{};
using function_task_t = d2::function_task::type>;
d2::task_handle_task* function_task_p = alloc.new_object(std::forward(f), m_wait_ctx, context(), alloc);
return d2::task_handle_accessor::construct(function_task_p);
}
#endif
public:
task_group_base(uintptr_t traits = 0)
: m_wait_ctx(0)
, m_context(task_group_context::bound, task_group_context::default_traits | traits)
{}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
task_group_base(task_group_context& ctx)
: m_wait_ctx(0)
, m_context(&ctx)
{}
#endif
~task_group_base() noexcept(false) {
if (m_wait_ctx.continue_execution()) {
#if __TBB_CPP17_UNCAUGHT_EXCEPTIONS_PRESENT
bool stack_unwinding_in_progress = std::uncaught_exceptions() > 0;
#else
bool stack_unwinding_in_progress = std::uncaught_exception();
#endif
// Always attempt to do proper cleanup to avoid inevitable memory corruption
// in case of missing wait (for the sake of better testability & debuggability)
if (!context().is_group_execution_cancelled())
cancel();
d1::wait(m_wait_ctx, context());
if (!stack_unwinding_in_progress)
throw_exception(exception_id::missing_wait);
}
}
task_group_status wait() {
bool cancellation_status = false;
try_call([&] {
d1::wait(m_wait_ctx, context());
}).on_completion([&] {
// TODO: the reset method is not thread-safe. Ensure the correct behavior.
cancellation_status = m_context.is_group_execution_cancelled();
context().reset();
});
return cancellation_status ? canceled : complete;
}
void cancel() {
context().cancel_group_execution();
}
}; // class task_group_base
class task_group : public task_group_base {
public:
task_group() : task_group_base(task_group_context::concurrent_wait) {}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
task_group(task_group_context& ctx) : task_group_base(ctx) {}
#endif
template
void run(F&& f) {
spawn(*prepare_task(std::forward(f)), context());
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
void run(d2::task_handle&& h) {
if (h == nullptr) {
throw_exception(exception_id::bad_task_handle);
}
using acs = d2::task_handle_accessor;
if (&acs::ctx_of(h) != &context()) {
throw_exception(exception_id::bad_task_handle_wrong_task_group);
}
spawn(*acs::release(h), context());
}
template
d2::task_handle defer(F&& f) {
return prepare_task_handle(std::forward(f));
}
#endif //__TBB_PREVIEW_TASK_GROUP_EXTENSIONS
template
task_group_status run_and_wait(const F& f) {
return internal_run_and_wait(f);
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
task_group_status run_and_wait(d2::task_handle&& h) {
return internal_run_and_wait(std::move(h));
}
#endif
}; // class task_group
#if TBB_PREVIEW_ISOLATED_TASK_GROUP
class spawn_delegate : public delegate_base {
task* task_to_spawn;
task_group_context& context;
bool operator()() const override {
spawn(*task_to_spawn, context);
return true;
}
public:
spawn_delegate(task* a_task, task_group_context& ctx)
: task_to_spawn(a_task), context(ctx)
{}
};
class wait_delegate : public delegate_base {
bool operator()() const override {
status = tg.wait();
return true;
}
protected:
task_group& tg;
task_group_status& status;
public:
wait_delegate(task_group& a_group, task_group_status& tgs)
: tg(a_group), status(tgs) {}
};
template
class run_wait_delegate : public wait_delegate {
F& func;
bool operator()() const override {
status = tg.run_and_wait(func);
return true;
}
public:
run_wait_delegate(task_group& a_group, F& a_func, task_group_status& tgs)
: wait_delegate(a_group, tgs), func(a_func) {}
};
class isolated_task_group : public task_group {
intptr_t this_isolation() {
return reinterpret_cast(this);
}
public:
isolated_task_group() : task_group() {}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
isolated_task_group(task_group_context& ctx) : task_group(ctx) {}
#endif
template
void run(F&& f) {
spawn_delegate sd(prepare_task(std::forward(f)), context());
r1::isolate_within_arena(sd, this_isolation());
}
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
void run(d2::task_handle&& h) {
if (h == nullptr) {
throw_exception(exception_id::bad_task_handle);
}
using acs = d2::task_handle_accessor;
if (&acs::ctx_of(h) != &context()) {
throw_exception(exception_id::bad_task_handle_wrong_task_group);
}
spawn_delegate sd(acs::release(h), context());
r1::isolate_within_arena(sd, this_isolation());
}
#endif //__TBB_PREVIEW_TASK_GROUP_EXTENSIONS
template
task_group_status run_and_wait( const F& f ) {
task_group_status result = not_complete;
run_wait_delegate rwd(*this, f, result);
r1::isolate_within_arena(rwd, this_isolation());
__TBB_ASSERT(result != not_complete, "premature exit from wait?");
return result;
}
task_group_status wait() {
task_group_status result = not_complete;
wait_delegate wd(*this, result);
r1::isolate_within_arena(wd, this_isolation());
__TBB_ASSERT(result != not_complete, "premature exit from wait?");
return result;
}
}; // class isolated_task_group
#endif // TBB_PREVIEW_ISOLATED_TASK_GROUP
inline bool is_current_task_group_canceling() {
task_group_context* ctx = current_context();
return ctx ? ctx->is_group_execution_cancelled() : false;
}
} // namespace d1
} // namespace detail
inline namespace v1 {
using detail::d1::task_group_context;
using detail::d1::task_group;
#if TBB_PREVIEW_ISOLATED_TASK_GROUP
using detail::d1::isolated_task_group;
#endif
using detail::d1::task_group_status;
using detail::d1::not_complete;
using detail::d1::complete;
using detail::d1::canceled;
using detail::d1::is_current_task_group_canceling;
using detail::r1::missing_wait;
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
using detail::d2::task_handle;
#endif
}
} // namespace tbb
#if _MSC_VER && !defined(__INTEL_COMPILER)
#pragma warning(pop) // 4324 warning
#endif
#endif // __TBB_task_group_H