ThreadedExecution

ThreadedExecution

#include <sbio/execution/threaded.hh>

class ThreadedExecution

Defined in src/lib/sbio/execution/threaded.hh:55

Inherits: Execution< ThreadedExecution >

ThreadedExecution manages a single-process/multi-thread IO strategy.

This execution policy the necessary index distribution, and data synchronization when operating brokers in a single process/multi-threaded environment. It is designed to allow data-buffers ([DataRole](api-sbio-DataRole.md#datarole)), and broker group buffers ([TableRole](api-sbio-TableRole.md#tablerole)) to use a ThreadLocalBuffer allowing for true thread-level parallelism. Other buffer roles are intentionally shared using a simpler HostBuffer which makes them available process wide.

List of all members

Name	Kind	Owner
ParallelSupport	variable	Declared here
configure_impl	function	Declared here
allocate_storage_impl	function	Declared here
allocate_impl_helper	function	Declared here
get_data_impl	function	Declared here
get_data_steps_impl	function	Declared here
next_impl	function	Declared here
BufferTypeFor	typedef	Declared here
m_num_threads	variable	Declared here
m_event_idx	variable	Declared here
m_trigger_mutex	variable	Declared here
m_shared_capacity	variable	Declared here
m_exhausted	variable	Declared here
m_broker_mutexes	variable	Declared here
m_logger	variable	Declared here
ParallelSupport	variable	Inherited from Execution
result_memory_space	function	Inherited from Execution
configure	function	Inherited from Execution
allocate_storage	function	Inherited from Execution
allocate_storage	function	Inherited from Execution
get_block_from_pool	function	Inherited from Execution
should_index	function	Inherited from Execution
should_process	function	Inherited from Execution
on_step	function	Inherited from Execution
pre_update	function	Inherited from Execution
post_update	function	Inherited from Execution
execute_read	function	Inherited from Execution
acquire_broker_view	function	Inherited from Execution
allocate_group_storage	function	Inherited from Execution
get_data	function	Inherited from Execution
get_data_steps	function	Inherited from Execution
next	function	Inherited from Execution
BufferTypeFor	typedef	Inherited from Execution

Inherited from Execution

Kind	Name	Description
variable	ParallelSupport static constexpr	The set of parallelization methods supported by the Execution policy.
function	result_memory_space static inline constexpr	The memory space that results (via get_data) are returned in.
function	configure static inline
function	allocate_storage static inline requires FormatTraits<FTraits, IO, Derived>	Controls the allocation of storage.
function	allocate_storage static inline requires FormatTraits<FTraits, IO, Derived>
function	get_block_from_pool static inline
function	should_index static inline	Decide whether or not this StreamBroker should proceed to the indexing state.
function	should_process static inline	Decide whether or not this StreamBroker should process this particular step.
function	on_step static inline	An opportunity to provide explicit synchronization immediately after every step.
function	pre_update static inline	An opportunity to provide explicit synchronization before any Storage changes.
function	post_update static inline	An opportunity to provide explicit synchronization after any Storage changes.
function	execute_read static inline
function	acquire_broker_view static inline
function	allocate_group_storage static inline requires FormatTraits<FTraits, IO, Derived>
function	get_data static inline	Run the data fetching and then querying of the filled buffers.
function	get_data_steps static inline	Run the data fetching and then querying of the filled buffers, for a BATCh of steps.
function	next static inline	Request the next step index to read data for.
typedef	BufferTypeFor	Sub-classes must define the used buffer types.

Public Static Attributes

Return	Name	Description
constexpr std::bitset< static_cast< std::size_t >(ParallelizationMethods::NUM_METHODS) >	ParallelSupport static constexpr

---

ParallelSupport

static constexpr

constexpr std::bitset< static_cast< std::size_t >(ParallelizationMethods::NUM_METHODS) > ParallelSupport { 0x1 }

Defined in src/lib/sbio/execution/threaded.hh:72

Public Static Methods

Return	Name	Description
void	configure_impl static inline
auto	allocate_storage_impl static inline requires FormatTraits<FTraits, IO, ThreadedExecution>	Allocate ThreadLocalBuffer storage for Index/DataRole and HostBuffer otherwise.
auto	allocate_impl_helper static inline
IOStatus	get_data_impl static inline	The ThreadedExecution policy splits BrokerGroup data fetch and resolution.
IOStatus	get_data_steps_impl static inline	The ThreadedExecution policy multi-contiguous step fetch and get.
FTraits::StepIdxType	next_impl static inline	The ThreadedExecution policy generates step indices in monotonically.

---

configure_impl

static inline

static inline void configure_impl(constConfig & config)

Defined in src/lib/sbio/execution/threaded.hh:74

---

allocate_storage_impl

static inline requires FormatTraits<FTraits, IO, ThreadedExecution>

template<IsTypeList Requirements, class IO, class FTraits> static inline auto allocate_storage_impl(constAllocationRequest< FTraits > & request) requires FormatTraits<FTraits, IO, ThreadedExecution>

Defined in src/lib/sbio/execution/threaded.hh:97

Allocate ThreadLocalBuffer storage for Index/DataRole and HostBuffer otherwise.

Beyond the role-based specialization, this execution policy does not perform any complex calculation when serving allocation requests. If there is memory available to serve the request, it will be fulfilled withou modification.

Returns

Allocated storage per the request.

Parameters

Parameter	Type	Description
request	[const](api-sbio-Execution.md#should_process)[AllocationRequest](api-sbio-AllocationRequest.md#allocationrequest)< FTraits > &	The allocation request.

---

allocate_impl_helper

static inline

template<typename... Descriptors, class FTraits> static inline auto allocate_impl_helper(TypeList< Descriptors... >, constAllocationRequest< FTraits > & request)

Defined in src/lib/sbio/execution/threaded.hh:110

---

get_data_impl

static inline

template<class FTraits, classFetchCBType, classGetCBType> static inline IOStatus get_data_impl(typename FTraits::StepIdxType step_idx, FetchCBType && unit_fetcher, std::size_t num_fetches, GetCBType && unit_get_data, std::size_t num_accesses)

Defined in src/lib/sbio/execution/threaded.hh:163

The ThreadedExecution policy splits BrokerGroup data fetch and resolution.

When the BrokerGroup requests data of a specific kind for a specific index, the process will be split into two stages. First, the IO fetch portion, where the data is actually pulled from the stream into memory buffers, is done behind a lock. This lock simplifies life on the Broker-side, as any shared mutable state (counters, or internal tracking variables) does not need to be protected while data is being read. It, of course, has a performance penalty, however, as the fetch is serialized.

After the fetch portion concludes, the data resolution stage can be done in parallel.

Returns

The IOStatus from the fetch and get procedure.

Parameters

Parameter	Type	Description
step_idx	[typename](api-sbio-Execution.md#should_process) FTraits::StepIdxType	The step_idx for which data should be fetched and read. @oaram[in] unit_fetcher A per-broker callback from the BrokerGroup to fetch data.
num_fetches	std::size_t	The number of fetches to perform. (Generally equal to the number of brokers)
unit_get_data	[GetCBType](api-sbio-Execution.md#should_process) &&	A per-broker callback from the BrokerGroup to resolve a piece of requested data inside the bytes just fetched.
num_accesses	std::size_t	The number of gets to perform. (Generally equal to the number of brokers)

---

get_data_steps_impl

static inline

template<class FTraits, classFetchCBType, classGetCBType> static inline IOStatus get_data_steps_impl(const std::initializer_list< typename FTraits::StepIdxType > steps, FetchCBType && unit_fetcher, std::size_t num_fetches, GetCBType && unit_get_data, std::size_t num_accesses)

Defined in src/lib/sbio/execution/threaded.hh:208

The ThreadedExecution policy multi-contiguous step fetch and get.

Returns

The IOStatus from the fetch and get procedure.

Parameters

Parameter	Type	Description
steps	[const](api-sbio-Execution.md#should_process) std::initializer_list< [typename](api-sbio-Execution.md#should_process) FTraits::StepIdxType >	The steps to read { start, stop }. Currently adding a third step for strided access is not yet supported. @oaram[in] unit_fetcher A per-broker callback from the BrokerGroup to fetch data.
num_fetches	std::size_t	The number of fetches to perform. (Generally equal to the number of brokers)
unit_get_data	[GetCBType](api-sbio-Execution.md#should_process) &&	A per-broker callback from the BrokerGroup to resolve a piece of requested data inside the bytes just fetched.
num_accesses	std::size_t	The number of gets to perform. (Generally equal to the number of brokers)

---

next_impl

static inline

template<class FTraits, classIndexTrigger> static inline FTraits::StepIdxType next_impl(typename FTraits::StepIdxType & max_capacity, IndexTrigger && trigger)

Defined in src/lib/sbio/execution/threaded.hh:264

The ThreadedExecution policy generates step indices in monotonically.

As the index generator is shared by all threads in the process, it is generated on an atomic counter. This requires the use of acquire/release semantics, and thus has some ammount of overhead compared to a lock-free policy.

Likewise, a lock is explicitly acquired in the event that the index capacity has been reached, and thus a reindexing trigger must be called.

Note

This policy is intended to be used with indexable formats. This function will work if it is not indexable; however, the callback trigger must then be designed in a way to terminate the generation of indices in some fashion, or it will continue forever.

Returns

The next step_idx.

Parameters

Parameter	Type	Description
max_capacity	[typename](api-sbio-Execution.md#should_process) FTraits::StepIdxType &	The current max capacity (currently available indices).
trigger	[IndexTrigger](api-sbio-Execution.md#should_process) &&	The reindex callback routine.

Public Types

Name	Description
BufferTypeFor

---

BufferTypeFor

using BufferTypeFor = std::conditional_t< std::is_same_v< typename Descriptor::role, DataRole >||std::is_same_v< typename Descriptor::role, TableRole >, ThreadLocalBuffer, HostBuffer >

Defined in src/lib/sbio/execution/threaded.hh:58

Private Static Attributes

Return	Name	Description
std::size_t	m_num_threads static
std::atomic< std::size_t >	m_event_idx static
std::mutex	m_trigger_mutex static
std::atomic< std::size_t >	m_shared_capacity static
std::atomic< bool >	m_exhausted static
std::mutex	m_broker_mutexes static	Set of mutexes to allow different brokers of a group from different threads to fetch in parallel.
std::shared_ptr< spdlog::logger >	m_logger static

---

m_num_threads

static

std::size_t m_num_threads { 0 }

Defined in src/lib/sbio/execution/threaded.hh:332

---

m_event_idx

static

std::atomic< std::size_t > m_event_idx { 0 }

Defined in src/lib/sbio/execution/threaded.hh:334

---

m_trigger_mutex

static

std::mutex m_trigger_mutex

Defined in src/lib/sbio/execution/threaded.hh:336

---

m_shared_capacity

static

std::atomic< std::size_t > m_shared_capacity { 0 }

Defined in src/lib/sbio/execution/threaded.hh:338

---

m_exhausted

static

std::atomic< bool > m_exhausted {  }

Defined in src/lib/sbio/execution/threaded.hh:340

---

m_broker_mutexes

static

std::mutex m_broker_mutexes

Defined in src/lib/sbio/execution/threaded.hh:346

Set of mutexes to allow different brokers of a group from different threads to fetch in parallel.

---

m_logger

static

std::shared_ptr< spdlog::logger > m_logger

Defined in src/lib/sbio/execution/threaded.hh:348