MPIThreadedExecution

MPIThreadedExecution

#include <sbio/execution/mpi_threaded.hh>

class MPIThreadedExecution

Defined in src/lib/sbio/execution/mpi_threaded.hh:64

Inherits: Execution< MPIThreadedExecution >

The MPIThreadedExecution policy is designed for a hybrid MPI/threaded approach.

The main features of the policy are:

• Allocations for types of the [IndexRole](api-sbio-IndexRole.md#indexrole) are made over shared MPI windows.

• Shared values are accordingly synchronized via the sync group mechanism on pre/post update hooks.

• As [IndexRole](api-sbio-IndexRole.md#indexrole) buffers are shared, redundant INDEXING is avoided by only allowing that Broker state for rank 0.

• Steps are processed in a simple round-robin fashion with a modulo world-size increment for each rank.

• Exhaustion of available indexed capacity automatically retriggers INDEXING if applicable for the Brokers.

• Allocations for the types of [DataRole](api-sbio-DataRole.md#datarole) and [TableRole](api-sbio-TableRole.md#tablerole) use ThreadLocalBuffer to allow a single rank to parallelize data across multiple threads.

List of all members

Name	Kind	Owner
MaxInactiveRanks	variable	Declared here
ParallelSupport	variable	Declared here
configure_impl	function	Declared here
allocate_storage_impl	function	Declared here
allocate_impl_helper	function	Declared here
pre_update_impl	function	Declared here
should_index_impl	function	Declared here
post_update_impl	function	Declared here
get_data_impl	function	Declared here
get_data_steps_impl	function	Declared here
is_current_rank_inactive	function	Declared here
next_impl	function	Declared here
BufferTypeFor	typedef	Declared here
m_world_comm	variable	Declared here
m_active_comm	variable	Declared here
m_active_rank	variable	Declared here
m_active_size	variable	Declared here
m_inactive_ranks	variable	Declared here
m_num_inactive_ranks	variable	Declared here
m_main_rank	variable	Declared here
m_main_rank_loops	variable	Declared here
m_num_threads	variable	Declared here
m_shmem_comm	variable	Declared here
m_rank	variable	Declared here
m_size	variable	Declared here
m_broker_mutexes	variable	Declared here
m_trigger_mutex	variable	Declared here
m_local_idx	variable	Declared here
m_shared_capacity	variable	Declared here
m_exhausted	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::size_t	MaxInactiveRanks static constexpr
constexpr std::bitset< static_cast< std::size_t >(ParallelizationMethods::NUM_METHODS) >	ParallelSupport static constexpr

---

MaxInactiveRanks

static constexpr

constexpr std::size_t MaxInactiveRanks { 1024 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:66

---

ParallelSupport

static constexpr

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

Defined in src/lib/sbio/execution/mpi_threaded.hh:94

Public Static Methods

Return	Name	Description
void	configure_impl static inline
auto	allocate_storage_impl static inline requires FormatTraits<FTraits, IO, MPIThreadedExecution>
auto	allocate_impl_helper static inline
void	pre_update_impl static inline	Before updates, fence shared memory.
bool	should_index_impl static inline	Check on if indexing should be done by the Broker.
void	post_update_impl static inline	After updates, synchronize shared resources.
IOStatus	get_data_impl static inline	The MPIThreadedExecution policy splits BrokerGroup data fetch and resolution.
IOStatus	get_data_steps_impl static inline	The MPIThreadedExecution policy multi-contiguous step fetch and get.
bool	is_current_rank_inactive static inline
FTraits::StepIdxType	next_impl static inline	The MPIThreadedExecution policy generates step indices modulo MPI world size.

---

configure_impl

static inline

static inline void configure_impl(constConfig & config)

Defined in src/lib/sbio/execution/mpi_threaded.hh:96

---

allocate_storage_impl

static inline requires FormatTraits<FTraits, IO, MPIThreadedExecution>

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

Defined in src/lib/sbio/execution/mpi_threaded.hh:194

---

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/mpi_threaded.hh:207

---

pre_update_impl

static inline

template<classRole, classStorageT> static inline void pre_update_impl(StorageT & storage)

Defined in src/lib/sbio/execution/mpi_threaded.hh:262

Before updates, fence shared memory.

This policy establishes a shared memory window over index data and additional roles which have been marked as [Shareable](api-sbio-Shareable.md#shareable). The pre_update hook includes a fence on that window so we can synchronize updates to the memory across ranks.

Parameters

Parameter	Type	Description
storage	[StorageT](api-sbio-Execution.md#should_process) &	The storage to synchronize.

---

should_index_impl

static inline

static inline bool should_index_impl()

Defined in src/lib/sbio/execution/mpi_threaded.hh:292

Check on if indexing should be done by the Broker.

File indexing is performed only by rank 0. This execution policy provides a shared memory buffer for the index storage - synchronization of that buffer ensures that other ranks will see any data rank 0 puts into it.

Returns

true for rank 0, else false.

---

post_update_impl

static inline

template<classRole, classStorageT, classSyncT> static inline void post_update_impl(StorageT & storage, SyncT && sync_vars, IOStatus status)

Defined in src/lib/sbio/execution/mpi_threaded.hh:310

After updates, synchronize shared resources.

Paired with the pre_update hook, the post_update ensures the synchronization of [IndexRole](api-sbio-IndexRole.md#indexrole) storage is completed.

Any remaining sync_vars are synchronized via a simple MPI_Bcast.

Parameters

Parameter	Type	Description
storage	[StorageT](api-sbio-Execution.md#should_process) &	The storage to synchronize.
sync_vars	[SyncT](api-sbio-Execution.md#should_process) &&	The attributes that require synchronization
status	[IOStatus](api-sbio.md#iostatus)	The IOStatus from the INDEXING state.

---

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/mpi_threaded.hh:426

The MPIThreadedExecution 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/mpi_threaded.hh:471

The MPIThreadedExecution 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)

---

is_current_rank_inactive

static inline

static inline bool is_current_rank_inactive()

Defined in src/lib/sbio/execution/mpi_threaded.hh:502

---

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/mpi_threaded.hh:540

The MPIThreadedExecution policy generates step indices modulo MPI world size.

This policy generates steps with consideration of both the MPI world, as well as the threads in a single MPI rank. 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. Furthermore, the increment on the counter is done modulo the size of the MPI world. This gives each rank a fixed offset for received indices:

• Step 1: Rank 0 processes 0, Rank 1 processes 1

• Step 2: Rank 0 processes 2, Rank 1 processes 3 … and so on.

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, IndexRole >||std::is_same_v< typename Descriptor::role, GroupRole >||std::is_same_v< typename Descriptor::hint, Shareable >, MPISharedBuffer, 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/mpi_threaded.hh:69

Private Static Attributes

Return	Name	Description
MPI_Comm	m_world_comm static	Communicator for synchronizing across the whole MPI world.
MPI_Comm	m_active_comm static
int	m_active_rank static
int	m_active_size static
int	m_inactive_ranks static
std::size_t	m_num_inactive_ranks static
int	m_main_rank static
bool	m_main_rank_loops static
std::size_t	m_num_threads static
MPI_Comm	m_shmem_comm static	Communicator used when generating shareable buffers.
int	m_rank static	This processes rank in the MPI world.
int	m_size static	The size of the MPI world.
std::mutex	m_broker_mutexes static	Set of mutexes to allow different brokers of a group from different threads to fetch in parallel.
std::mutex	m_trigger_mutex static	Mutex for thread synchronization on reindexing
std::atomic< std::size_t >	m_local_idx static	Thread-local index within the rank’s set of indices to distribute.
std::atomic< std::size_t >	m_shared_capacity static	Intra-rank capacity store for inter-thread synchronization of indices.
std::atomic< bool >	m_exhausted static	Latch for if the rank has exhausted all indices.
std::shared_ptr< spdlog::logger >	m_logger static	Execution policy logger.

---

m_world_comm

static

MPI_Comm m_world_comm {  }

Defined in src/lib/sbio/execution/mpi_threaded.hh:656

Communicator for synchronizing across the whole MPI world.

---

m_active_comm

static

MPI_Comm m_active_comm {  }

Defined in src/lib/sbio/execution/mpi_threaded.hh:657

---

m_active_rank

static

int m_active_rank { -1 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:658

---

m_active_size

static

int m_active_size { -1 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:659

---

m_inactive_ranks

static

int m_inactive_ranks {}

Defined in src/lib/sbio/execution/mpi_threaded.hh:660

---

m_num_inactive_ranks

static

std::size_t m_num_inactive_ranks { 0 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:661

---

m_main_rank

static

int m_main_rank { 0 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:662

---

m_main_rank_loops

static

bool m_main_rank_loops {  }

Defined in src/lib/sbio/execution/mpi_threaded.hh:663

---

m_num_threads

static

std::size_t m_num_threads { 0 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:665

---

m_shmem_comm

static

MPI_Comm m_shmem_comm {  }

Defined in src/lib/sbio/execution/mpi_threaded.hh:670

Communicator used when generating shareable buffers.

---

m_rank

static

int m_rank { -1 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:671

This processes rank in the MPI world.

---

m_size

static

int m_size { -1 }

Defined in src/lib/sbio/execution/mpi_threaded.hh:672

The size of the MPI world.

---

m_broker_mutexes

static

std::mutex m_broker_mutexes

Defined in src/lib/sbio/execution/mpi_threaded.hh:677

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

---

m_trigger_mutex

static

std::mutex m_trigger_mutex

Defined in src/lib/sbio/execution/mpi_threaded.hh:681

Mutex for thread synchronization on reindexing

---

m_local_idx

static

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

Defined in src/lib/sbio/execution/mpi_threaded.hh:685

Thread-local index within the rank’s set of indices to distribute.

---

m_shared_capacity

static

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

Defined in src/lib/sbio/execution/mpi_threaded.hh:689

Intra-rank capacity store for inter-thread synchronization of indices.

---

m_exhausted

static

std::atomic< bool > m_exhausted {  }

Defined in src/lib/sbio/execution/mpi_threaded.hh:693

Latch for if the rank has exhausted all indices.

---

m_logger

static

std::shared_ptr< spdlog::logger > m_logger

Defined in src/lib/sbio/execution/mpi_threaded.hh:694

Execution policy logger.