MPIThreadedDistribution

#include <sbio/execution/mpi_threaded.hh>

class MPIThreadedDistribution

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

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

The main features of the policy are:

Allocations for types of the IndexRole are made over shared MPI windows.
Shared values are accordingly synchronized via the sync group mechanism on pre/post update hooks.
As 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 and TableRole use ThreadLocalBuffer to allow a single rank to parallelize data across multiple threads.

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:70

ParallelSupport

static constexpr

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

Type: constexpr std::bitset< static_cast< std::size_t >(ParallelizationMethods::NUM_METHODS) >

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

Public Static Methods

Return	Name	Description
constexpr `BufferScope`	`scope_for` `static` `inline` `constexpr`	Index and Shareable buffers are shared by all ranks on a node with MPI Windows. Data and Table buffers are per-thread, and any other buffers are per-process.
`bool`	`validate_environment_impl` `static` `inline`	The requirements are that MPI was not finalized, and MPI_THREAD_MULTIPLE support.
`void`	`configure_impl` `static` `inline`
`std::size_t`	`unit_ordinal` `static` `inline`	The ordinal (associated number) for the calling parallel unit on calling node.
`void`	`init_logger` `static` `inline`	Setup a spdlog logger.
`void`	`allocate_node_shared` `static` `inline`	Setup a node-wide shared buffer.
`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.
`StepBatch`< typename FTraits::StepIdxType >	`next_batch_impl` `static` `inline`	The MPIThreadedExecution policy generates contiguous batches of step indices.
`void`	`end_iteration_impl` `static` `inline`	Exit the iteration with `state`, releasing any held steps.

scope_for

static inline constexpr

template<typename Descriptor> constexpr static inline constexpr BufferScope scope_for()

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

Index and Shareable buffers are shared by all ranks on a node with MPI Windows. Data and Table buffers are per-thread, and any other buffers are per-process.

validate_environment_impl

static inline

static inline bool validate_environment_impl(const Config & config, const char *& error)

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

The requirements are that MPI was not finalized, and MPI_THREAD_MULTIPLE support.

configure_impl

static inline

static inline void configure_impl(const Config & config)

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

unit_ordinal

static inline

static inline std::size_t unit_ordinal()

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

The ordinal (associated number) for the calling parallel unit on calling node.

The units are numbered rank-first, and then by the rank’s threads in the order that the threads ask for a number. I.e., the first thread of every rank on a node comes before the second thread.

init_logger

static inline

static inline void init_logger()

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

Setup a spdlog logger.

allocate_node_shared

static inline

static inline void allocate_node_shared(MPISharedBuffer & buf, std::size_t sz)

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

Setup a node-wide shared buffer.

For the MPIThreaded distribution strategy, an MPI Window is used for sharing. Within the window an area is reserved by the distribution manager (a “catalog”) for distributing state among the participant ranks. If a rank is inactive, it instead will have a private buffer in lieu of the window.

pre_update_impl

static inline

template<class Role, class StorageT> static inline void pre_update_impl(StorageT & storage)

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

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 &`	The storage to synchronize.

should_index_impl

static inline

static inline bool should_index_impl()

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

Check on if indexing should be done by the Broker.

File indexing is performed by one rank on each individual node (machine/server). The first rank of the node (rank 0, or lowest, of a node-wide communicator) fills the indexing role.

The EPolicy provides a shared-memory buffer for the index storage, so the other ranks can read what the indexer puts into it.

Returns

true for the node’s indexing rank (or an inactive rank), else false.

post_update_impl

static inline

template<class Role, class StorageT, class SyncT> static inline void post_update_impl(StorageT & storage, SyncT && sync_vars, IOStatus status)

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

After updates, synchronize shared resources.

Paired with the pre_update hook, the post_update ensures the synchronization of IndexRole storage is completed.

Remaining sync_vars (including index catalog) get shared through the window as well, via a special catalog area. The catalog area is setup by the indexer for the node, and then it publishes a new sequence number. The other ranks can read back after this. (No additional messages are passed, so communication cannot end up causing a stall if a rank leaves early.)

Parameters

Parameter	Type	Description
`storage`	`StorageT &`	The storage to synchronize.
`sync_vars`	`SyncT &&`	The attributes that require synchronization
`status`	`IOStatus`	The IOStatus from the INDEXING state.

get_data_impl

static inline

template<class FTraits, class FetchCBType, class GetCBType> 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:639

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 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 &&`	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, class FetchCBType, class GetCBType> 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:683

The MPIThreadedExecution policy multi-contiguous step fetch and get.

Returns

The IOStatus from the fetch and get procedure.

Parameters

Parameter	Type	Description
`steps`	`const std::initializer_list< typename 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 &&`	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:713

next_impl

static inline

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

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

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.

Returns

The next step_idx.

Parameters

Parameter	Type	Description
`max_capacity`	`typename FTraits::StepIdxType &`	The current max capacity (currently available indices).
`trigger`	`IndexTrigger &&`	The reindex callback routine.

next_batch_impl

static inline

template<class FTraits, class IndexTrigger> static inline StepBatch< typename FTraits::StepIdxType > next_batch_impl(IterationState & state, std::size_t batch_size, typename FTraits::StepIdxType & max_capacity, IndexTrigger && trigger)

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

The MPIThreadedExecution policy generates contiguous batches of step indices.

Each capacity window [window start, capacity) is cut into blocks of batch_size steps, which are dealt round-robin over the iterating ranks. The dealing continues across windows: the g-th block overall goes to the worker g % worker_count. The threads of a rank share a round counter, and round j of the window gives the rank its j-th block of the window. The final block of a window may be shorter.

Reindexing follows next_impl, under the trigger mutex, after the rank’s handed out batches are returned and the node released the window.

Returns

The next batch [first, last), or the Exhausted batch.

Parameters

Parameter	Type	Description
`batch_size`	`std::size_t`	The maximum size of a batch.
`max_capacity`	`typename FTraits::StepIdxType &`	The current max capacity (currently available indices).
`trigger`	`IndexTrigger &&`	The reindex callback routine.

end_iteration_impl

static inline

template<class FTraits, class IndexTrigger> static inline void end_iteration_impl(IterationState & state, typename FTraits::StepIdxType & max_capacity, IndexTrigger && trigger)

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

Exit the iteration with state, releasing any held steps.

When the last iterating thread of the rank leaves, before data is exhuasted, the rank is marked as departed and no longer iterates. Any calls on next will just return exhausted - so the node doesn’t wait for it during reindexing. However, a node INDEXER will keep reindexing while any other rank on the node is around, or until the data is fully exhausted.

Parameters

Parameter	Type	Description
`max_capacity`	`typename FTraits::StepIdxType &`	The current max capacity (of step indices).
`trigger`	`IndexTrigger &&`	The reindex callback trigger.

Public Types

Name	Description
`NodeSharedBuffer`

NodeSharedBuffer

using NodeSharedBuffer = MPISharedBuffer

Type: MPISharedBuffer

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

Private Static Attributes

Return	Name	Description
`constexpr std::size_t`	`CatalogBytes` `static` `constexpr`	Window area for sync_vars.
`constexpr std::size_t`	`CatalogHeaderBytes` `static` `constexpr`	Sequence number slot.
`constexpr int`	`NodeRelease` `static` `constexpr`	Done with the window - may be rewritten.
`constexpr int`	`NodeDeparted` `static` `constexpr`	Left the iteration early.
`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`
`std::atomic< std::size_t >`	`m_next_thread_ordinal` `static`	For unit_ordinal routine.
`MPI_Comm`	`m_shmem_comm` `static`	Communicator used when generating shareable buffers.
`int`	`m_shmem_rank` `static`	This rank within its node (m_shmem_comm).
`int`	`m_shmem_size` `static`	The number of (active) ranks on this node.
`int`	`m_node_indexer` `static`	The node rank which indexes for the node.
`std::vector< int >`	`m_node_iterates` `static`	For each rank on the node, whether it iterates (calls `next`).
`int`	`m_rank` `static`	This processes rank in the MPI world.
`int`	`m_size` `static`	The size of the MPI world.
`std::shared_ptr< spdlog::logger >`	`m_logger` `static`	Execution policy logger.

CatalogBytes

static constexpr

constexpr std::size_t CatalogBytes { 4096 }

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

Window area for sync_vars.

CatalogHeaderBytes

static constexpr

constexpr std::size_t CatalogHeaderBytes { 64 }

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

Sequence number slot.

NodeRelease

static constexpr

constexpr int NodeRelease { 1 }

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

Done with the window - may be rewritten.

NodeDeparted

static constexpr

constexpr int NodeDeparted { 2 }

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

Left the iteration early.

m_world_comm

static

MPI_Comm m_world_comm { MPI_COMM_NULL }

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

Communicator for synchronizing across the whole MPI world.

m_active_comm

static

MPI_Comm m_active_comm { MPI_COMM_NULL }

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

m_active_rank

static

int m_active_rank { -1 }

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

m_active_size

static

int m_active_size { -1 }

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

m_inactive_ranks

static

int m_inactive_ranks[MaxInactiveRanks]

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

m_num_inactive_ranks

static

std::size_t m_num_inactive_ranks { 0 }

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

m_main_rank

static

int m_main_rank { 0 }

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

m_main_rank_loops

static

bool m_main_rank_loops { true }

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

m_num_threads

static

std::size_t m_num_threads { 0 }

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

m_next_thread_ordinal

static

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

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

For unit_ordinal routine.

m_shmem_comm

static

MPI_Comm m_shmem_comm { MPI_COMM_NULL }

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

Communicator used when generating shareable buffers.

m_shmem_rank

static

int m_shmem_rank { -1 }

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

This rank within its node (m_shmem_comm).

m_shmem_size

static

int m_shmem_size { 0 }

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

The number of (active) ranks on this node.

m_node_indexer

static

int m_node_indexer { -1 }

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

The node rank which indexes for the node.

m_node_iterates

static

std::vector< int > m_node_iterates {}

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

For each rank on the node, whether it iterates (calls next).

m_rank

static

int m_rank { -1 }

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

This processes rank in the MPI world.

m_size

static

int m_size { -1 }

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

The size of the MPI world.

m_logger

static

std::shared_ptr< spdlog::logger > m_logger

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

Execution policy logger.

Private Static Methods

Return	Name	Description
`constexpr std::size_t`	`catalog_offset` `static` `inline` `constexpr`	Offset of the catalog area after a shared buffer of `size` bytes.
`char *`	`catalog_area` `static` `inline`	The catalog used by the processing ranks (inset into the rest of the buffer).
`bool`	`rank_iterates` `static` `inline`	Whether this rank takes steps (calls `next`) at all.
`void`	`publish_window` `static` `inline`	Change the capacity window of `state` (under the trigger mutex).
`bool`	`is_node_indexer` `static` `inline`
`int`	`node_tag` `static` `inline`	Message tag of the node messages of an iteration.
`void`	`release_window` `static` `inline`	Release the current index window of the node before it is rewritten.
`bool`	`others_iterating` `static` `inline`	Whether other ranks of the node still iterate.
`bool`	`join_iteration` `static` `inline`	Count this thread as iterating with `state` (once per iteration).
`void`	`release_step` `static` `inline`

catalog_offset

static inline constexpr

constexpr static inline constexpr std::size_t catalog_offset(std::size_t size)

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

Offset of the catalog area after a shared buffer of size bytes.

catalog_area

static inline

template<class BufT> static inline char * catalog_area(BufT & buf)

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

The catalog used by the processing ranks (inset into the rest of the buffer).

rank_iterates

static inline

static inline bool rank_iterates()

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

Whether this rank takes steps (calls next) at all.

publish_window

static inline

template<class Update> static inline void publish_window(IterationState & state, Update && update)

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

Change the capacity window of state (under the trigger mutex).

Threads reading the window see either the old or the new one, never a mix.

is_node_indexer

static inline

static inline bool is_node_indexer()

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

node_tag

static inline

static inline int node_tag(const IterationState & state)

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

Message tag of the node messages of an iteration.

release_window

static inline

static inline void release_window(IterationState & state)

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

Release the current index window of the node before it is rewritten.

The other ranks of the node tell the node’s indexer that they are done with the window (or left the iteration entirely). The indexer waits for all of them. Called after the rank drained its handed out steps.

others_iterating

static inline

static inline bool others_iterating(const IterationState & state)

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

Whether other ranks of the node still iterate.

join_iteration

static inline

static inline bool join_iteration(IterationState & state)

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

Count this thread as iterating with state (once per iteration).

Returns

false if the rank departed the iteration.

release_step

static inline

static inline void release_step(IterationState & state)

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