tbb::interface7::internal::task_arena_base Class Reference

#include <task_arena.h>

Inheritance diagram for tbb::interface7::internal::task_arena_base:

Collaboration diagram for tbb::interface7::internal::task_arena_base:

Static Public Attributes
static const int	automatic = -1
	Typedef for number of threads that is automatic. More...
static const int	not_initialized = -2

Protected Types
enum	{ default_flags, exact_exception_flag = task_group_context::exact_exception }

Protected Member Functions
	task_arena_base (int max_concurrency, unsigned reserved_for_masters)
void __TBB_EXPORTED_METHOD	internal_initialize ()
void __TBB_EXPORTED_METHOD	internal_terminate ()
void __TBB_EXPORTED_METHOD	internal_attach ()
void __TBB_EXPORTED_METHOD	internal_enqueue (task &, intptr_t) const
void __TBB_EXPORTED_METHOD	internal_execute (delegate_base &) const
void __TBB_EXPORTED_METHOD	internal_wait () const

Static Protected Member Functions
static int __TBB_EXPORTED_FUNC	internal_current_slot ()
static int __TBB_EXPORTED_FUNC	internal_max_concurrency (const task_arena *)

Protected Attributes
internal::arena *	my_arena
	NULL if not currently initialized. More...
task_group_context *	my_context
	default context of the arena More...
int	my_max_concurrency
	Concurrency level for deferred initialization. More...
unsigned	my_master_slots
	Reserved master slots. More...
intptr_t	my_version_and_traits
	Special settings. More...

Detailed Description

Definition at line 103 of file task_arena.h.

Member Enumeration Documentation

anonymous enum

anonymous enum

protected

Enumerator
default_flags
exact_exception_flag

Definition at line 122 of file task_arena.h.

         {
        default_flags = 0
#if __TBB_TASK_GROUP_CONTEXT
        | (task_group_context::default_traits & task_group_context::exact_exception)  // 0 or 1 << 16
        , exact_exception_flag = task_group_context::exact_exception // used to specify flag for context directly
#endif
    };

Constructor & Destructor Documentation

task_arena_base()

tbb::interface7::internal::task_arena_base::task_arena_base ( int  max_concurrency, unsigned  reserved_for_masters  )

inlineprotected

Definition at line 130 of file task_arena.h.

References __TBB_EXPORTED_FUNC, and __TBB_EXPORTED_METHOD.

        : my_arena(0)
#if __TBB_TASK_GROUP_CONTEXT
        , my_context(0)
#endif
        , my_max_concurrency(max_concurrency)
        , my_master_slots(reserved_for_masters)
        , my_version_and_traits(default_flags)
        {}

Member Function Documentation

internal_attach()

void tbb::interface7::internal::task_arena_base::internal_attach ( )

protected

Definition at line 780 of file arena.cpp.

References __TBB_ASSERT, tbb::internal::market::global_market(), tbb::internal::governor::local_scheduler_if_initialized(), tbb::internal::scheduler_state::my_arena, tbb::internal::arena_base::my_max_num_workers, tbb::internal::arena_base::my_num_reserved_slots, tbb::internal::arena_base::my_num_slots, tbb::internal::arena_base::my_references, tbb::internal::arena::num_arena_slots(), tbb::internal::arena::ref_external, and s.

                                       {
    __TBB_ASSERT(!my_arena, NULL);
    generic_scheduler* s = governor::local_scheduler_if_initialized();
    if( s && s->my_arena ) {
        // There is an active arena to attach to.
        // It's still used by s, so won't be destroyed right away.
        my_arena = s->my_arena;
        __TBB_ASSERT( my_arena->my_references > 0, NULL );
        my_arena->my_references += arena::ref_external;
#if __TBB_TASK_GROUP_CONTEXT
        my_context = my_arena->my_default_ctx;
        my_version_and_traits |= my_context->my_version_and_traits & exact_exception_flag;
#endif
        my_master_slots = my_arena->my_num_reserved_slots;
        my_max_concurrency = my_master_slots + my_arena->my_max_num_workers;
        __TBB_ASSERT(arena::num_arena_slots(my_max_concurrency)==my_arena->my_num_slots, NULL);
        // increases market's ref count for task_arena
        market::global_market( /*is_public=*/true );
    }
}

Here is the call graph for this function:

internal_current_slot()

int tbb::interface7::internal::task_arena_base::internal_current_slot ( )

staticprotected

Definition at line 1015 of file arena.cpp.

References __TBB_ASSERT, __TBB_ASSERT_EX, d, int, tbb::interface7::internal::isolate_within_arena(), tbb::internal::governor::local_scheduler_if_initialized(), tbb::internal::governor::local_scheduler_weak(), tbb::internal::scheduler_state::my_arena_index, tbb::internal::scheduler_state::my_innermost_running_task, tbb::task::prefix(), and s.

                                             {
    generic_scheduler* s = governor::local_scheduler_if_initialized();
    return s? int(s->my_arena_index) : -1;
}

Here is the call graph for this function:

internal_enqueue()

void tbb::interface7::internal::task_arena_base::internal_enqueue ( task &  t, intptr_t  prio  ) const

protected

Definition at line 801 of file arena.cpp.

References __TBB_ASSERT, tbb::internal::arena::enqueue_task(), tbb::internal::governor::local_scheduler_if_initialized(), tbb::internal::scheduler_state::my_arena, tbb::internal::generic_scheduler::my_random, tbb::task::prefix(), and s.

                                                                     {
    __TBB_ASSERT(my_arena, NULL);
    generic_scheduler* s = governor::local_scheduler_if_initialized();
    __TBB_ASSERT(s, "Scheduler is not initialized"); // we allocated a task so can expect the scheduler
#if __TBB_TASK_GROUP_CONTEXT
    __TBB_ASSERT(my_arena->my_default_ctx == t.prefix().context, NULL);
    __TBB_ASSERT(!my_arena->my_default_ctx->is_group_execution_cancelled(), // TODO: any better idea?
                 "The task will not be executed because default task_group_context of task_arena is cancelled. Has previously enqueued task thrown an exception?");
#endif
    my_arena->enqueue_task( t, prio, s->my_random );
}

Here is the call graph for this function:

internal_execute()

void tbb::interface7::internal::task_arena_base::internal_execute ( delegate_base &  ) const

protected

Definition at line 862 of file arena.cpp.

References __TBB_ASSERT, __TBB_CONTEXT_ARG, __TBB_CONTEXT_ARG1, tbb::internal::__TBB_load_with_acquire(), tbb::task::allocate_root(), tbb::internal::concurrent_monitor::cancel_wait(), tbb::internal::concurrent_monitor::commit_wait(), tbb::task_group_context::copy_fp_settings(), d, tbb::task_group_context::default_traits, tbb::internal::arena::enqueue_task(), tbb::task_group_context::exact_exception, tbb::task_group_context::isolated, tbb::internal::governor::local_scheduler_weak(), tbb::internal::generic_scheduler::local_wait_for_all(), tbb::internal::scheduler_properties::master, tbb::internal::scheduler_state::my_arena, tbb::internal::scheduler_state::my_arena_index, tbb::task_group_context::my_exception, tbb::internal::arena_base::my_exit_monitors, tbb::internal::generic_scheduler::my_random, tbb::internal::concurrent_monitor::notify_one(), tbb::internal::arena::occupy_free_slot(), tbb::internal::arena::out_of_arena, tbb::internal::auto_empty_task::prefix(), tbb::internal::concurrent_monitor::prepare_wait(), tbb::internal::task_prefix::ref_count, tbb::task_group_context::register_pending_exception(), tbb::internal::context_guard_helper< T >::restore_default(), s, scope, and tbb::internal::context_guard_helper< T >::set_ctx().

                                                                     {
    __TBB_ASSERT(my_arena, NULL);
    generic_scheduler* s = governor::local_scheduler_weak();
    __TBB_ASSERT(s, "Scheduler is not initialized");

    bool same_arena = s->my_arena == my_arena;
    size_t index1 = s->my_arena_index;
    if (!same_arena) {
        index1 = my_arena->occupy_free_slot</* as_worker*/false>(*s);
        if (index1 == arena::out_of_arena) {

#if __TBB_USE_OPTIONAL_RTTI
            // Workaround for the bug inside graph. If the thread can not occupy arena slot during task_arena::execute()
            // and all aggregator operations depend on this task completion (all other threads are inside arena already)
            // deadlock appears, because enqueued task will never enter arena.
            // Workaround: check if the task came from graph via RTTI (casting to graph::spawn_functor)
            // and enqueue this task with non-blocking internal_enqueue method.
            // TODO: have to change behaviour later in next GOLD release (maybe to add new library entry point - try_execute)
            typedef tbb::flow::interface10::graph::spawn_functor graph_funct;
            internal::delegated_function< graph_funct, void >* deleg_funct =
                    dynamic_cast< internal::delegated_function< graph_funct, void>* >(&d);

            if (deleg_funct) {
                internal_enqueue(*new(task::allocate_root(*my_context))
                    internal::function_task< internal::strip< graph_funct >::type >
                        (internal::forward< graph_funct >(deleg_funct->my_func)), 0);
                return;
            } else {
#endif /*  __TBB_USE_OPTIONAL_RTTI */
                concurrent_monitor::thread_context waiter;
#if __TBB_TASK_GROUP_CONTEXT
                task_group_context exec_context(task_group_context::isolated, my_version_and_traits & exact_exception_flag);
#if __TBB_FP_CONTEXT
                exec_context.copy_fp_settings(*my_context);
#endif
#endif
                auto_empty_task root(__TBB_CONTEXT_ARG(s, &exec_context));
                root.prefix().ref_count = 2;
                my_arena->enqueue_task(*new(task::allocate_root(__TBB_CONTEXT_ARG1(exec_context)))
                    delegated_task(d, my_arena->my_exit_monitors, &root),
                    0, s->my_random); // TODO: priority?
                size_t index2 = arena::out_of_arena;
                do {
                    my_arena->my_exit_monitors.prepare_wait(waiter, (uintptr_t)&d);
                    if (__TBB_load_with_acquire(root.prefix().ref_count) < 2) {
                        my_arena->my_exit_monitors.cancel_wait(waiter);
                        break;
                    }
                    index2 = my_arena->occupy_free_slot</*as_worker*/false>(*s);
                    if (index2 != arena::out_of_arena) {
                        my_arena->my_exit_monitors.cancel_wait(waiter);
                        nested_arena_context scope(s, my_arena, index2, scheduler_properties::master, same_arena);
                        s->local_wait_for_all(root, NULL);
#if TBB_USE_EXCEPTIONS
                        __TBB_ASSERT(!exec_context.my_exception, NULL); // exception can be thrown above, not deferred
#endif
                        __TBB_ASSERT(root.prefix().ref_count == 0, NULL);
                        break;
                    }
                    my_arena->my_exit_monitors.commit_wait(waiter);
                } while (__TBB_load_with_acquire(root.prefix().ref_count) == 2);
                if (index2 == arena::out_of_arena) {
                    // notify a waiting thread even if this thread did not enter arena,
                    // in case it was woken by a leaving thread but did not need to enter
                    my_arena->my_exit_monitors.notify_one(); // do not relax!
                }
#if TBB_USE_EXCEPTIONS
                // process possible exception
                if (task_group_context::exception_container_type *pe = exec_context.my_exception)
                    TbbRethrowException(pe);
#endif
                return;
#if __TBB_USE_OPTIONAL_RTTI
            } // if task came from graph
#endif
        } // if (index1 == arena::out_of_arena)
    } // if (!same_arena)

    context_guard_helper</*report_tasks=*/false> context_guard;
    context_guard.set_ctx(__TBB_CONTEXT_ARG1(my_context));
#if TBB_USE_EXCEPTIONS
    try {
#endif
        //TODO: replace dummy tasks for workers as well to avoid using of the_dummy_context
        nested_arena_context scope(s, my_arena, index1, scheduler_properties::master, same_arena);
        d();
#if TBB_USE_EXCEPTIONS
    }
    catch (...) {
        context_guard.restore_default(); // TODO: is it needed on Windows?
        if (my_version_and_traits & exact_exception_flag) throw;
        else {
            task_group_context exception_container(task_group_context::isolated,
                task_group_context::default_traits & ~task_group_context::exact_exception);
            exception_container.register_pending_exception();
            __TBB_ASSERT(exception_container.my_exception, NULL);
            TbbRethrowException(exception_container.my_exception);
        }
    }
#endif
}

Here is the call graph for this function:

internal_initialize()

void tbb::interface7::internal::task_arena_base::internal_initialize ( )

protected

Definition at line 736 of file arena.cpp.

References __TBB_ASSERT, tbb::internal::as_atomic(), tbb::internal::market::create_arena(), tbb::internal::governor::default_num_threads(), tbb::task_group_context::default_traits, tbb::internal::market::global_market(), int, tbb::task_group_context::isolated, tbb::internal::governor::local_scheduler_weak(), tbb::internal::scheduler_state::my_arena, tbb::internal::NFS_Allocate(), tbb::internal::arena::on_thread_leaving(), tbb::internal::governor::one_time_init(), tbb::internal::arena::ref_external, tbb::internal::market::release(), and tbb::internal::spin_wait_while_eq().

                                           {
    governor::one_time_init();
    if( my_max_concurrency < 1 )
        my_max_concurrency = (int)governor::default_num_threads();
    __TBB_ASSERT( my_master_slots <= (unsigned)my_max_concurrency, "Number of slots reserved for master should not exceed arena concurrency");
    arena* new_arena = market::create_arena( my_max_concurrency, my_master_slots, 0 );
    // add an internal market reference; a public reference was added in create_arena
    market &m = market::global_market( /*is_public=*/false );
    // allocate default context for task_arena
#if __TBB_TASK_GROUP_CONTEXT
    new_arena->my_default_ctx = new ( NFS_Allocate(1, sizeof(task_group_context), NULL) )
            task_group_context( task_group_context::isolated, task_group_context::default_traits );
#if __TBB_FP_CONTEXT
    new_arena->my_default_ctx->capture_fp_settings();
#endif
#endif /* __TBB_TASK_GROUP_CONTEXT */
    // threads might race to initialize the arena
    if(as_atomic(my_arena).compare_and_swap(new_arena, NULL) != NULL) {
        __TBB_ASSERT(my_arena, NULL); // another thread won the race
        // release public market reference
        m.release( /*is_public=*/true, /*blocking_terminate=*/false );
        new_arena->on_thread_leaving<arena::ref_external>(); // destroy unneeded arena
#if __TBB_TASK_GROUP_CONTEXT
        spin_wait_while_eq(my_context, (task_group_context*)NULL);
    } else {
        new_arena->my_default_ctx->my_version_and_traits |= my_version_and_traits & exact_exception_flag;
        as_atomic(my_context) = new_arena->my_default_ctx;
#endif
    }
    // TODO: should it trigger automatic initialization of this thread?
    governor::local_scheduler_weak();
}

Here is the call graph for this function:

internal_max_concurrency()

int tbb::interface7::internal::task_arena_base::internal_max_concurrency ( const task_arena *  ta )

staticprotected

Definition at line 1046 of file arena.cpp.

References __TBB_ASSERT, tbb::internal::governor::default_num_threads(), int, tbb::internal::governor::local_scheduler_if_initialized(), my_arena, my_max_concurrency, tbb::internal::arena_base::my_max_num_workers, tbb::internal::arena_base::my_num_reserved_slots, and s.

Referenced by tbb::this_task_arena::max_concurrency().

                                                                  {
    arena* a = NULL;
    if( ta ) // for special cases of ta->max_concurrency()
        a = ta->my_arena;
    else if( generic_scheduler* s = governor::local_scheduler_if_initialized() )
        a = s->my_arena; // the current arena if any

    if( a ) { // Get parameters from the arena
        __TBB_ASSERT( !ta || ta->my_max_concurrency==1, NULL );
        return a->my_num_reserved_slots + a->my_max_num_workers;
    } else {
        __TBB_ASSERT( !ta || ta->my_max_concurrency==automatic, NULL );
        return int(governor::default_num_threads());
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

internal_terminate()

void tbb::interface7::internal::task_arena_base::internal_terminate ( )

protected

Definition at line 769 of file arena.cpp.

References tbb::internal::scheduler_state::my_arena, tbb::internal::arena_base::my_market, tbb::internal::arena::on_thread_leaving(), tbb::internal::arena::ref_external, and tbb::internal::market::release().

                                          {
    if( my_arena ) {// task_arena was initialized
        my_arena->my_market->release( /*is_public=*/true, /*blocking_terminate=*/false );
        my_arena->on_thread_leaving<arena::ref_external>();
        my_arena = 0;
#if __TBB_TASK_GROUP_CONTEXT
        my_context = 0;
#endif
    }
}

Here is the call graph for this function:

internal_wait()

void tbb::interface7::internal::task_arena_base::internal_wait ( ) const

protected

Definition at line 986 of file arena.cpp.

References __TBB_ASSERT, __TBB_CONTEXT_ARG1, tbb::internal::__TBB_load_with_acquire(), __TBB_Yield, tbb::task::allocate_root(), tbb::internal::as_atomic(), tbb::internal::governor::local_scheduler_weak(), tbb::internal::generic_scheduler::master_outermost_level(), tbb::internal::scheduler_state::my_arena, tbb::internal::scheduler_state::my_arena_index, tbb::internal::arena_base::my_pool_state, tbb::internal::arena_slot_line1::my_scheduler, tbb::internal::arena::my_slots, tbb::internal::arena::num_workers_active(), tbb::internal::binary_semaphore::P(), s, tbb::internal::arena::SNAPSHOT_EMPTY, tbb::internal::generic_scheduler::wait_until_empty(), and tbb::internal::scheduler_properties::worker.

                                          {
    __TBB_ASSERT(my_arena, NULL);
    generic_scheduler* s = governor::local_scheduler_weak();
    __TBB_ASSERT(s, "Scheduler is not initialized");
    __TBB_ASSERT(s->my_arena != my_arena || s->my_arena_index == 0, "task_arena::wait_until_empty() is not supported within a worker context" );
    if( s->my_arena == my_arena ) {
        //unsupported, but try do something for outermost master
        __TBB_ASSERT(s->master_outermost_level(), "unsupported");
        if( !s->my_arena_index )
            while( my_arena->num_workers_active() )
                s->wait_until_empty();
    } else for(;;) {
        while( my_arena->my_pool_state != arena::SNAPSHOT_EMPTY ) {
            if( !__TBB_load_with_acquire(my_arena->my_slots[0].my_scheduler) // TODO TEMP: one master, make more masters
                && as_atomic(my_arena->my_slots[0].my_scheduler).compare_and_swap(s, NULL) == NULL ) {
                nested_arena_context a(s, my_arena, 0, scheduler_properties::worker, false);
                s->wait_until_empty();
            } else {
                binary_semaphore waiter; // TODO: replace by a single event notification from is_out_of_work
                internal_enqueue( *new( task::allocate_root(__TBB_CONTEXT_ARG1(*my_context)) ) wait_task(waiter), 0 ); // TODO: priority?
                waiter.P(); // TODO: concurrent_monitor
            }
        }
        if( !my_arena->num_workers_active() && !my_arena->my_slots[0].my_scheduler) // no activity
            break; // spin until workers active but avoid spinning in a worker
        __TBB_Yield(); // wait until workers and master leave
    }
}

Here is the call graph for this function:

Member Data Documentation

automatic

const int tbb::interface7::internal::task_arena_base::automatic = -1

static

Typedef for number of threads that is automatic.

Definition at line 150 of file task_arena.h.

my_arena

internal::arena* tbb::interface7::internal::task_arena_base::my_arena

protected

NULL if not currently initialized.

Definition at line 106 of file task_arena.h.

Referenced by internal_max_concurrency().

my_context

task_group_context* tbb::interface7::internal::task_arena_base::my_context

protected

default context of the arena

Definition at line 110 of file task_arena.h.

my_master_slots

unsigned tbb::interface7::internal::task_arena_base::my_master_slots

protected

Reserved master slots.

Definition at line 117 of file task_arena.h.

my_max_concurrency

int tbb::interface7::internal::task_arena_base::my_max_concurrency

protected

Concurrency level for deferred initialization.

Definition at line 114 of file task_arena.h.

Referenced by internal_max_concurrency().

my_version_and_traits

intptr_t tbb::interface7::internal::task_arena_base::my_version_and_traits

protected

Special settings.

Definition at line 120 of file task_arena.h.

not_initialized

const int tbb::interface7::internal::task_arena_base::not_initialized = -2

static

Definition at line 151 of file task_arena.h.

Referenced by tbb::this_task_arena::current_thread_index(), and tbb::interface9::internal::get_initial_partition_head().

---

The documentation for this class was generated from the following files:

• task_arena.h
• arena.cpp