_flow_graph_async_msg_impl.h
Go to the documentation of this file.
/*
Copyright (c) 2005-2020 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__flow_graph_async_msg_impl_H
#define __TBB__flow_graph_async_msg_impl_H
#ifndef __TBB_flow_graph_H
#error Do not #include this internal file directly; use public TBB headers instead.
#endif
namespace internal {
template <typename T>
class async_storage {
public:
typedef receiver<T> async_storage_client;
async_storage() : my_graph(nullptr) {
my_data_ready.store<tbb::relaxed>(false);
}
~async_storage() {
// Release reference to the graph if async_storage
// was destructed before set() call
if (my_graph) {
my_graph->release_wait();
my_graph = nullptr;
}
}
template<typename C>
async_storage(C&& data) : my_graph(nullptr), my_data( std::forward<C>(data) ) {
using namespace tbb::internal;
__TBB_STATIC_ASSERT( (is_same_type<typename strip<C>::type, typename strip<T>::type>::value), "incoming type must be T" );
my_data_ready.store<tbb::relaxed>(true);
}
template<typename C>
bool set(C&& data) {
using namespace tbb::internal;
__TBB_STATIC_ASSERT( (is_same_type<typename strip<C>::type, typename strip<T>::type>::value), "incoming type must be T" );
{
tbb::spin_mutex::scoped_lock locker(my_mutex);
if (my_data_ready.load<tbb::relaxed>()) {
__TBB_ASSERT(false, "double set() call");
return false;
}
my_data = std::forward<C>(data);
my_data_ready.store<tbb::release>(true);
}
// Thread sync is on my_data_ready flag
for (typename subscriber_list_type::iterator it = my_clients.begin(); it != my_clients.end(); ++it) {
(*it)->try_put(my_data);
}
// Data was sent, release reference to the graph
if (my_graph) {
my_graph->release_wait();
my_graph = nullptr;
}
return true;
}
task* subscribe(async_storage_client& client, graph& g) {
if (! my_data_ready.load<tbb::acquire>())
{
tbb::spin_mutex::scoped_lock locker(my_mutex);
if (! my_data_ready.load<tbb::relaxed>()) {
#if TBB_USE_ASSERT
for (typename subscriber_list_type::iterator it = my_clients.begin(); it != my_clients.end(); ++it) {
__TBB_ASSERT(*it != &client, "unexpected double subscription");
}
#endif // TBB_USE_ASSERT
// Increase graph lifetime
my_graph = &g;
my_graph->reserve_wait();
// Subscribe
my_clients.push_back(&client);
return SUCCESSFULLY_ENQUEUED;
}
}
__TBB_ASSERT(my_data_ready.load<tbb::relaxed>(), "data is NOT ready");
return client.try_put_task(my_data);
}
private:
graph* my_graph;
tbb::spin_mutex my_mutex;
tbb::atomic<bool> my_data_ready;
T my_data;
typedef std::vector<async_storage_client*> subscriber_list_type;
subscriber_list_type my_clients;
};
} // namespace internal
template <typename T>
class async_msg {
template< typename > friend class receiver;
template< typename, typename > friend struct internal::async_helpers;
public:
typedef T async_msg_data_type;
async_msg() : my_storage(std::make_shared< internal::async_storage<T> >()) {}
async_msg(const T& t) : my_storage(std::make_shared< internal::async_storage<T> >(t)) {}
async_msg(T&& t) : my_storage(std::make_shared< internal::async_storage<T> >( std::move(t) )) {}
virtual ~async_msg() {}
void set(const T& t) {
my_storage->set(t);
}
void set(T&& t) {
my_storage->set( std::move(t) );
}
protected:
// Can be overridden in derived class to inform that
// async calculation chain is over
virtual void finalize() const {}
private:
typedef std::shared_ptr< internal::async_storage<T> > async_storage_ptr;
async_storage_ptr my_storage;
};
#endif // __TBB__flow_graph_async_msg_impl_H
async_msg::async_msg
async_msg(const T &t)
Definition: _flow_graph_async_msg_impl.h:129
tbb::acquire
Acquire.
Definition: atomic.h:57
tbb::spin_mutex
A lock that occupies a single byte.
Definition: spin_mutex.h:39
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::relaxed
No ordering.
Definition: atomic.h:61
async_msg::async_msg_data_type
T async_msg_data_type
Definition: _flow_graph_async_msg_impl.h:125
async_msg::async_storage_ptr
std::shared_ptr< internal::async_storage< T > > async_storage_ptr
Definition: _flow_graph_async_msg_impl.h:149
async_msg::~async_msg
virtual ~async_msg()
Definition: _flow_graph_async_msg_impl.h:133
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
internal::async_storage::subscribe
task * subscribe(async_storage_client &client, graph &g)
Definition: _flow_graph_async_msg_impl.h:83
internal::async_storage::my_graph
graph * my_graph
Definition: _flow_graph_async_msg_impl.h:110
internal::async_storage::my_data
T my_data
Definition: _flow_graph_async_msg_impl.h:113
internal::async_storage
Definition: _flow_graph_async_msg_impl.h:27
tbb::flow::internal::SUCCESSFULLY_ENQUEUED
static tbb::task *const SUCCESSFULLY_ENQUEUED
Definition: _flow_graph_impl.h:61
internal::async_storage::async_storage
async_storage()
Definition: _flow_graph_async_msg_impl.h:31
internal::async_storage::my_mutex
tbb::spin_mutex my_mutex
Definition: _flow_graph_async_msg_impl.h:111
__TBB_STATIC_ASSERT
#define __TBB_STATIC_ASSERT(condition, msg)
Definition: tbb_stddef.h:553
async_msg::async_msg
async_msg(T &&t)
Definition: _flow_graph_async_msg_impl.h:131
tbb::move
void move(tbb_thread &t1, tbb_thread &t2)
Definition: tbb_thread.h:319
internal::async_storage::subscriber_list_type
std::vector< async_storage_client * > subscriber_list_type
Definition: _flow_graph_async_msg_impl.h:114
internal::async_storage::~async_storage
~async_storage()
Definition: _flow_graph_async_msg_impl.h:35
tbb::internal::strip::type
T type
Definition: _template_helpers.h:38
async_msg::set
void set(T &&t)
Definition: _flow_graph_async_msg_impl.h:139
async_msg::my_storage
async_storage_ptr my_storage
Definition: _flow_graph_async_msg_impl.h:150
internal::async_storage::set
bool set(C &&data)
Definition: _flow_graph_async_msg_impl.h:53
task
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task * task
Definition: ittnotify_static.h:119
async_msg
Definition: _flow_graph_async_msg_impl.h:121
tbb::internal::is_same_type
Detects whether two given types are the same.
Definition: _template_helpers.h:61
async_msg::async_msg
async_msg()
Definition: _flow_graph_async_msg_impl.h:127
value
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task __itt_model_task_instance ITT_FORMAT p void ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long value
Definition: ittnotify_static.h:192
internal::async_storage::my_data_ready
tbb::atomic< bool > my_data_ready
Definition: _flow_graph_async_msg_impl.h:112
async_msg::finalize
virtual void finalize() const
Definition: _flow_graph_async_msg_impl.h:146
tbb::internal
Identifiers declared inside namespace internal should never be used directly by client code.
Definition: atomic.h:65
internal::async_storage::async_storage
async_storage(C &&data)
Definition: _flow_graph_async_msg_impl.h:45
data
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task __itt_model_task_instance ITT_FORMAT p void ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type size_t void * data
Definition: ittnotify_static.h:198
internal::async_storage::async_storage_client
receiver< T > async_storage_client
Definition: _flow_graph_async_msg_impl.h:29
tbb::release
Release.
Definition: atomic.h:59
tbb::spin_mutex::scoped_lock
Represents acquisition of a mutex.
Definition: spin_mutex.h:53
internal::async_storage::my_clients
subscriber_list_type my_clients
Definition: _flow_graph_async_msg_impl.h:115
async_msg::set
void set(const T &t)
Definition: _flow_graph_async_msg_impl.h:135

Copyright © 2005-2020 Intel Corporation. All Rights Reserved.

Intel, Pentium, Intel Xeon, Itanium, Intel XScale and VTune are registered trademarks or trademarks of Intel Corporation or its subsidiaries in the United States and other countries.

* Other names and brands may be claimed as the property of others.