StreamBroker

StreamBroker

#include <sbio/core/broker.hh>

template<IOTraits IO, class EPolicy, FormatTraits< IO, EPolicy > FTraits, class Derived = void>
class StreamBroker

Defined in src/lib/sbio/core/broker.hh:142

A StreamBroker organizes Streams and mediates data requests.

The StreamBroker has a state machine and exposes data-format dependent metadata and data buffers.

Template Parameters

• IO The type of the IO strategy being used.

• EPolicy The Execution policy to use for reading data.

• FTraits The data format to read.

• Derived The sub-class type if defined. All functionality can be exposed purely from the FTraits template parameter, so sub-classing is not strictly necessary. I.e., this class can be instantiated directly.

Public Methods

Return	Name	Description
	StreamBroker inline	A default constructor is provided for simplicity.
	StreamBroker inline	Construct the Broker with a specific, data-format-dependent, set of parameters.
SBIO_HD void	configure_broker inline	If a Broker has been default-constructed, it can be moved through the INIT state via this function.
SBIO_HD void	allocate_storage inline	Perform data format specific allocations in preparation for streaming data.
SBIO_HDIOStatus	open_data_stream inline	Connect and open brokered Streams.
SBIO_HDIOStatus	discover_metadata inline	After Stream connection, perform metadata discovery.
SBIO_HDIOStatus	prepare inline	Run through the initial allocate, connect, discover steps in one.
SBIO_HDIOStatus	index_stream inline	Index the data Stream if appropriate and implemented.
SBIO_HDIOStatus	fetch_step inline	Retrieve data for the specified index using the provided lookup pattern.
SBIO_HDIOStatus	fetch_steps inline
SBIO_HDIOStatus	process inline
SBIO_HDIOStatus	run inline
SBIO_HDIOStatus	step inline
SBIO_HDBrokerState	state const inline	Return the current broker state along the state machine.
SBIO_HD std::size_t	capacity const inline	Return the current capacity for data formats that support indexing.
SBIO_HD FTraits::DataUnit *	current_buffer inline	Access the current data after a fetch.
SBIO_HDDataResult	get_data_in_buffer inline	Parse specific data from the buffer after a fetch.
SBIO_HDStreamMetadata &	metadata inline	The compiled metadata from the brokered Stream(s).
SBIO_HD const StreamMetadata &	metadata const inline	The compiled metadata from the brokered Stream(s).
SBIO_HDGenericStreamConfig< DataFormat > &	config inline	The set of StreamParameters configuration used to instantiate the broker.
SBIO_HD const GenericStreamConfig< DataFormat > &	config const inline	The set of StreamParameters configuration used to instantiate the broker.
SBIO_HD void	set_metadata inline	Configure the Broker with a set of metadata.
SBIO_HDStreamType &	stream inline
SBIO_HD const StreamType &	stream const inline
SBIO_HD auto	sync_vars inline
SBIO_HD	~StreamBroker	Defaulted destructor.
	StreamBroker noexcept	Defaulted constructor.
StreamBroker &	operator= noexcept	Defaulted assignment operator.
SBIO_HD std::uint32_t	stream_idx const inline
SBIO_HDStreamCatalog< FTraits > &	catalog inline noexcept
SBIO_HD const StreamCatalog< FTraits > &	catalog const inline noexcept
SBIO_HDIOStatus	preceding inline requires UsesGenericIndex

---

StreamBroker

inline

inline StreamBroker()

Defined in src/lib/sbio/core/broker.hh:256

A default constructor is provided for simplicity.

The Broker will remain in the INIT state until explicitly configured, if using this constructor.

---

StreamBroker

inline

inline StreamBroker(const GenericStreamConfig< DataFormat > & cfg)

Defined in src/lib/sbio/core/broker.hh:265

Construct the Broker with a specific, data-format-dependent, set of parameters.

Parameters

Parameter	Type	Description
cfg	const GenericStreamConfig< DataFormat > &	The data-format-dependent configuration parameters.

---

configure_broker

inline

inline SBIO_HD void configure_broker(const GenericStreamConfig< DataFormat > & cfg)

Defined in src/lib/sbio/core/broker.hh:278

If a Broker has been default-constructed, it can be moved through the INIT state via this function.

This corresponds to the INIT stage of the state machine.

Parameters

Parameter	Type	Description
cfg	const GenericStreamConfig< DataFormat > &	The data format specif configuration parameters.

---

allocate_storage

inline

inline SBIO_HD void allocate_storage()

Defined in src/lib/sbio/core/broker.hh:288

Perform data format specific allocations in preparation for streaming data.

This corresponds to the ALLOCATE stage of the state machine.

---

open_data_stream

inline

inline SBIO_HDIOStatus open_data_stream()

Defined in src/lib/sbio/core/broker.hh:315

Connect and open brokered Streams.

This corresponds to the CONNECT stage of the state machine.

Returns

The IOStatus result from connecting.

---

discover_metadata

inline

inline SBIO_HDIOStatus discover_metadata()

Defined in src/lib/sbio/core/broker.hh:369

After Stream connection, perform metadata discovery.

This function determines what contents are available from the brokered Streams. This information is then available to the caller (or higher-level abstractions).

This corresponds to the DISCOVERY stage of the state machine.

Returns

The IOStatus result for whether metadata reads were successful.

---

prepare

inline

inline SBIO_HDIOStatus prepare()

Defined in src/lib/sbio/core/broker.hh:399

Run through the initial allocate, connect, discover steps in one.

This is simply a convenience wrapper instead of calling each of the early state transitions independently. The down-side is the status will be cummulative so a failure may possibly not be immediately clear as originiating from one step or another.

Returns

The IOStatus result from all 3 steps.

---

index_stream

inline

inline SBIO_HDIOStatus index_stream()

Defined in src/lib/sbio/core/broker.hh:436

Index the data Stream if appropriate and implemented.

Indexing affords the possibility of ordering the data from (a) Stream(s) for indexable lookup. Not all data formats support a notion of indexing.

This corresponds to the INDEXING stage of the state machine.

Returns

An IOStatus for whether indexing was successful.

---

fetch_step

inline

inline SBIO_HDIOStatus fetch_step(StepIdxType step_idx, const DataAccessPtn ptn, SegmentCursor< FTraits > & cursor)

Defined in src/lib/sbio/core/broker.hh:481

Retrieve data for the specified index using the provided lookup pattern.

The index used for lookup is data format dependent. If the format supports indexing, then it may correspond to a chunk of data other than the most recent. If not, then this will always return the most recent data from the Stream.

Returns

An IOStatus for whether the lookup was succesful.

Parameters

Parameter	Type	Description
step_idx	StepIdxType	The data format-dependent data chunk index.
ptn	const DataAccessPtn	A lookup pattern. Some data formats support access in various ways.

---

fetch_steps

inline

inline SBIO_HDIOStatus fetch_steps(std::initializer_list< StepIdxType > steps, const DataAccessPtn ptn, SegmentCursor< FTraits > & cursor)

Defined in src/lib/sbio/core/broker.hh:512

---

process

inline

inline SBIO_HDIOStatus process()

Defined in src/lib/sbio/core/broker.hh:568

---

run

inline

inline SBIO_HDIOStatus run()

Defined in src/lib/sbio/core/broker.hh:576

---

step

inline

template<class CBType> inline SBIO_HDIOStatus step(CBType && callback)

Defined in src/lib/sbio/core/broker.hh:586

---

state

const inline

inline SBIO_HDBrokerState state() const

Defined in src/lib/sbio/core/broker.hh:603

Return the current broker state along the state machine.

Returns

The current broker state along the state machine.

---

capacity

const inline

inline SBIO_HD std::size_t capacity() const

Defined in src/lib/sbio/core/broker.hh:615

Return the current capacity for data formats that support indexing.

When indexing is supported, the capacity indicates how many indices are immediately available to jump to in the stream. Reading beyond the capacity requires another INDEXING transition, or a move to reading in true streaming manner.

Returns

The current capacity: the number of immediately available indices.

---

current_buffer

inline

inline SBIO_HD FTraits::DataUnit * current_buffer()

Defined in src/lib/sbio/core/broker.hh:641

Access the current data after a fetch.

The action of performing a fetch pulls the data from the stream into available memory managed by the Broker (in conjunction with the Execution policy). The fetch APIs, however, do not return access to this memory directly. Instead, this function should be used to access data after a fetch if the raw buffer is required. Otherwise, more specific APIs can be used to parse specific components from the buffer.

Returns

A pointer to the buffer filled after a fetch from (a) Stream(s).

---

get_data_in_buffer

inline

inline SBIO_HDDataResult get_data_in_buffer(const DataRequest & req, const DataAccessPtn ptn, SegmentCursor< FTraits > & cursor, std::size_t batch_idx = 0)

Defined in src/lib/sbio/core/broker.hh:666

Parse specific data from the buffer after a fetch.

This function parses, in a data format-dependent manner, a portion of data from a buffer fetched from (a) Stream(s).

Returns

A data-format-dependent result object with the requested data.

Parameters

Parameter	Type	Description
req	const DataRequest &	A data-format-dependent struct with a parse request.
ptn	const DataAccessPtn	A data-format-dependent access pattern for formats that can be read in different ways.
batch_idx	std::size_t	If reading by batches, the index for which of the steps in the batch to be resolved.

---

metadata

inline

inline SBIO_HDStreamMetadata & metadata()

Defined in src/lib/sbio/core/broker.hh:690

The compiled metadata from the brokered Stream(s).

This function should only be used after having passed the DISCOVERY state, or having set the metadata explicitly.

Returns

The compiled metadata from the brokered Stream(s).

---

metadata

const inline

inline SBIO_HD const StreamMetadata & metadata() const

Defined in src/lib/sbio/core/broker.hh:699

The compiled metadata from the brokered Stream(s).

This function should only be used after having passed the DISCOVERY state, or having set the metadata explicitly.

Returns

The compiled metadata from the brokered Stream(s).

---

config

inline

inline SBIO_HDGenericStreamConfig< DataFormat > & config()

Defined in src/lib/sbio/core/broker.hh:706

The set of StreamParameters configuration used to instantiate the broker.

Returns

The StreamParameters configuration.

---

config

const inline

inline SBIO_HD const GenericStreamConfig< DataFormat > & config() const

Defined in src/lib/sbio/core/broker.hh:712

The set of StreamParameters configuration used to instantiate the broker.

Returns

The StreamParameters configuration.

---

set_metadata

inline

inline SBIO_HD void set_metadata(StreamMetadata & metadata)

Defined in src/lib/sbio/core/broker.hh:728

Configure the Broker with a set of metadata.

Under some Execution policies, not all Brokers will necessarily move through all states of the state machine. In case the DISCOVERY state is not explicitly moved to, then the metadata for the Broker can be set using this function. For example, multiple Brokers may be reading different parts of the same Stream(s) in parallel.

The metadata is not only used by the caller, but also internally by the Broker, or at least, it may be used. For that reason, this function is made available.

---

stream

inline

inline SBIO_HDStreamType & stream(std::size_t role_idx)

Defined in src/lib/sbio/core/broker.hh:732

---

stream

const inline

inline SBIO_HD const StreamType & stream(std::size_t role_idx) const

Defined in src/lib/sbio/core/broker.hh:736

---

sync_vars

inline

inline SBIO_HD auto sync_vars()

Defined in src/lib/sbio/core/broker.hh:740

---

~StreamBroker

SBIO_HD ~StreamBroker() = default

Defined in src/lib/sbio/core/broker.hh:753

Defaulted destructor.

---

StreamBroker

noexcept

StreamBroker(StreamBroker && other) noexcept = default

Defined in src/lib/sbio/core/broker.hh:755

Defaulted constructor.

---

operator=

noexcept

StreamBroker & operator=(StreamBroker && other) noexcept = default

Defined in src/lib/sbio/core/broker.hh:756

Defaulted assignment operator.

---

stream_idx

const inline

inline SBIO_HD std::uint32_t stream_idx() const

Defined in src/lib/sbio/core/broker.hh:759

---

catalog

inline noexcept

inline SBIO_HDStreamCatalog< FTraits > & catalog() noexcept

Defined in src/lib/sbio/core/broker.hh:761

---

catalog

const inline noexcept

inline SBIO_HD const StreamCatalog< FTraits > & catalog() const noexcept

Defined in src/lib/sbio/core/broker.hh:764

---

preceding

inline requires UsesGenericIndex

template<typename KindT> inline SBIO_HDIOStatus preceding(KindT kind, KindT ref_kind, hd_std::uint64_t ref_ordinal, hd_std::uint64_t & ordinal, SegmentCursor< FTraits > & cursor) requires UsesGenericIndex

Defined in src/lib/sbio/core/broker.hh:770

Public Static Attributes

Return	Name	Description
constexpr bool	UsesGenericIndex static constexpr	Whether the broker builds and queries the index itself (generic path).
constexpr std::size_t	StreamCount static constexpr
constexpr std::size_t	NumStepKinds static constexpr

---

UsesGenericIndex

static constexpr

constexpr bool UsesGenericIndex { HasGenericIndex<FTraits> }

Defined in src/lib/sbio/core/broker.hh:170

Whether the broker builds and queries the index itself (generic path).

---

StreamCount

static constexpr

constexpr std::size_t StreamCount { FTraits::StreamTypes::size() }

Defined in src/lib/sbio/core/broker.hh:240

---

NumStepKinds

static constexpr

constexpr std::size_t NumStepKinds = []() {
      if constexpr (requires { FTraits::NumStepKinds; }) {
        return FTraits::NumStepKinds;
      } else {
        return 1;
      }
    }()

Defined in src/lib/sbio/core/broker.hh:242

Protected Attributes

Return	Name	Description
StreamType	m_streams
GenericStreamConfig< DataFormat >	m_config
BrokerState	m_broker_state
StreamMetadata	m_metadata_inv
SBStorageType	m_storage
StreamCatalog< FTraits >	m_catalog
IndexingCursor< FTraits >	m_indexing_cursor

---

m_streams

StreamType m_streams[StreamCount]

Defined in src/lib/sbio/core/broker.hh:815

---

m_config

GenericStreamConfig< DataFormat > m_config

Type: GenericStreamConfig< DataFormat >

Defined in src/lib/sbio/core/broker.hh:816

---

m_broker_state

BrokerState m_broker_state

Type: BrokerState

Defined in src/lib/sbio/core/broker.hh:817

---

m_metadata_inv

StreamMetadata m_metadata_inv

Type: StreamMetadata

Defined in src/lib/sbio/core/broker.hh:819

---

m_storage

SBStorageType m_storage

Type: SBStorageType

Defined in src/lib/sbio/core/broker.hh:820

---

m_catalog

StreamCatalog< FTraits > m_catalog {}

Type: StreamCatalog< FTraits >

Defined in src/lib/sbio/core/broker.hh:822

---

m_indexing_cursor

IndexingCursor< FTraits > m_indexing_cursor {}

Type: IndexingCursor< FTraits >

Defined in src/lib/sbio/core/broker.hh:823

Public Types

Name	Description
IOPolicy	The type of the IO strategy being used.
ExecutionPolicy	The Execution policy type.
DataFormat	The type of data being read.
StreamType	The type of Stream: I.e., the IO strategy and data format being read.
SBStorageType	The type of the StreamBroker’s Storage.
IOMemTag	The memory space that the IOPolicy reads data into.
EPolicyConfig	The Execution policy configuration object type.
StreamConfig	The individual Stream configuration object type.
StreamMetadata	The type of the general metadata object for the data format’s Stream.
DataAccessPtn	The type of the enumerator used to specify access patterns used for the format.
DataRequest	The type of a request object used to query for data.
StepIdxType	The type used to request a specific step from the Stream.
MemTag	Based on choice of ExecutionPolicy, data will return in host or device buffers.

---

IOPolicy

using IOPolicy = IO

Defined in src/lib/sbio/core/broker.hh:147

The type of the IO strategy being used.

---

ExecutionPolicy

using ExecutionPolicy = EPolicy

Defined in src/lib/sbio/core/broker.hh:151

The Execution policy type.

---

DataFormat

using DataFormat = FTraits

Defined in src/lib/sbio/core/broker.hh:155

The type of data being read.

---

StreamType

using StreamType = Stream< IO, FTraits >

Type: Stream< IO, FTraits >

Defined in src/lib/sbio/core/broker.hh:160

The type of Stream: I.e., the IO strategy and data format being read.

---

SBStorageType

using SBStorageType = Storage< typename FTraits::BrokerBufferRequirements, EPolicy >

Type: Storage< typename FTraits::BrokerBufferRequirements, EPolicy >

Defined in src/lib/sbio/core/broker.hh:165

The type of the StreamBroker’s Storage.

---

IOMemTag

using IOMemTag = hd_std::conditional_t< has_capabilities(EPolicy::template required_io_capabilities< IO >(), IOCapability::ReadToDevice), ncarray::DevTag, ncarray::HostTag >

Type: hd_std::conditional_t< has_capabilities(EPolicy::template required_io_capabilities< IO >(), IOCapability::ReadToDevice), ncarray::DevTag, ncarray::HostTag >

Defined in src/lib/sbio/core/broker.hh:183

The memory space that the IOPolicy reads data into.

Set to device memory if the IOPolicy declares ReadToDevice, else host.

---

EPolicyConfig

using EPolicyConfig = typename EPolicy::Config

Defined in src/lib/sbio/core/broker.hh:197

The Execution policy configuration object type.

epolicy_config objects configure the global behaviour of the Execution policy being used. The Execution policy must be configured before any Streams are opened as it controls all aspects of IO down to the allocation of Storage.

---

StreamConfig

using StreamConfig = typename FTraits::StreamParameters

Defined in src/lib/sbio/core/broker.hh:204

The individual Stream configuration object type.

stream_config objects are used to set up each individual Stream so that it can connect and read from its individual data.

---

StreamMetadata

using StreamMetadata = MetadataInventory< FTraits >

Type: MetadataInventory< FTraits >

Defined in src/lib/sbio/core/broker.hh:209

The type of the general metadata object for the data format’s Stream.

---

DataAccessPtn

using DataAccessPtn = typename FTraits::DataAccessPtn

Defined in src/lib/sbio/core/broker.hh:214

The type of the enumerator used to specify access patterns used for the format.

---

DataRequest

using DataRequest = typename FTraits::DataRequest

Defined in src/lib/sbio/core/broker.hh:218

The type of a request object used to query for data.

---

StepIdxType

using StepIdxType = typename FTraits::StepIdxType

Defined in src/lib/sbio/core/broker.hh:225

The type used to request a specific step from the Stream.

This type is required and guaranteed to be convertible std::size_t; however, different data format’s may use different underlying types.

---

MemTag

using MemTag = hd_std::conditional_t< ExecutionPolicy::result_memory_space()==MemorySpace::Host, ncarray::HostTag, ncarray::DevTag >

Type: hd_std::conditional_t< ExecutionPolicy::result_memory_space()==MemorySpace::Host, ncarray::HostTag, ncarray::DevTag >

Defined in src/lib/sbio/core/broker.hh:234

Based on choice of ExecutionPolicy, data will return in host or device buffers.

As data is returned via ncarray array objects, the MemTag converts a MemorySpace indicator in sbio to the tagging system used for host/device array constructs in ncarray.

Private Methods

Return	Name	Description
SBIO_HDIOStatus	find_row const inline
SBIO_HDIOStatus	generic_index inline
SBIO_HDIOStatus	generic_fetch inline
SBIO_HDIOStatus	generic_fetch_range inline

---

find_row

const inline

template<typename KindT> inline SBIO_HDIOStatus find_row(const StepOffset< FTraits > * rows, KindT kind, hd_std::uint64_t ordinal, hd_std::size_t hint, hd_std::uint32_t & row) const

Defined in src/lib/sbio/core/broker.hh:852

---

generic_index

inline

template<class SV> inline SBIO_HDIOStatus generic_index(SV & sv)

Defined in src/lib/sbio/core/broker.hh:885

---

generic_fetch

inline

template<class SV> inline SBIO_HDIOStatus generic_fetch(SV & sv, StepIdxType step_idx, DataAccessPtn ptn, SegmentCursor< FTraits > & cursor)

Defined in src/lib/sbio/core/broker.hh:1021

---

generic_fetch_range

inline

template<class SV> inline SBIO_HDIOStatus generic_fetch_range(SV & sv, StepIdxType first, StepIdxType count, DataAccessPtn ptn, SegmentCursor< FTraits > & cursor)

Defined in src/lib/sbio/core/broker.hh:1062

Private Static Methods

Return	Name	Description
SBIO_HD constexpr hd_std::size_t	kind_idx static inline constexpr
SBIO_HD hd_std::size_t	unit_size static inline
SBIO_HDStepOffset< FTraits > *	acquire_rows static inline

---

kind_idx

static inline constexpr

template<typename KindT> constexpr static inline SBIO_HD constexpr hd_std::size_t kind_idx(KindT kind)

Defined in src/lib/sbio/core/broker.hh:827

---

unit_size

static inline

template<typename UnitT> static inline SBIO_HD hd_std::size_t unit_size(const UnitT * unit)

Defined in src/lib/sbio/core/broker.hh:832

---

acquire_rows

static inline

template<class SV> static inline SBIO_HDStepOffset< FTraits > * acquire_rows(SV & sv)

Defined in src/lib/sbio/core/broker.hh:841