Batching and unbatching processors

Description

Processors that aggregate events into batches or extract individual events from batches.

Functions
template<typename Event, typename Downstream>
auto	tcspc::batch (std::shared_ptr< bucket_source< Event > > buffer_provider, arg::batch_size< std::size_t > batch_size, Downstream downstream)
	Create a processor that batches events into buckets for buffering.
template<typename DataTypes = default_data_types, typename Downstream>
auto	tcspc::batch_bin_increment_clusters (std::shared_ptr< bucket_source< typename DataTypes::bin_index_type > > buffer_provider, arg::bucket_size< std::size_t > bucket_size, arg::batch_size< std::size_t > batch_size, Downstream downstream)
	Create a processor that collects bin increment clusters into encoded batches.
template<typename ContainerEvent, typename Downstream>
auto	tcspc::unbatch (Downstream downstream)
	Create a processor transforming batches of events to individual events.
template<typename DataTypes = default_data_types, typename Downstream>
auto	tcspc::unbatch_bin_increment_clusters (Downstream downstream)
	Create a processor that splits encoded batches of bin increment clusters into individual clusters.

Function Documentation

batch()

template<typename Event, typename Downstream>

auto tcspc::batch	(	std::shared_ptr< bucket_source< Event > >	buffer_provider,
		arg::batch_size< std::size_t >	batch_size,
		Downstream	downstream )

Create a processor that batches events into buckets for buffering.

Collects every batch_size events into a bucket. The buckets are obtained from the given buffer_provider.

The buckets are emitted as rvalue reference.

This processor does not perform time-based batching, so may introduce arbitrary delays to real-time event streams. For this reason, batching should not be performed (and is not necessary) for intermediate buffering of real-time streams in most cases.

See also

tcspc::unbatch()

Template Parameters

Event	the event type (must be default-constructible; trivial type recommended)
Downstream	downstream processor type

Parameters

buffer_provider	bucket source providing event buffers
batch_size	number of events to collect in each bucket
downstream	downstream processor

Returns

processor

Events handled

• Event: collected into tcspc::bucket<Event> and emitted as batch
• Flush: emit any buffered events as tcspc::bucket<Event>; pass through

batch_bin_increment_clusters()

template<typename DataTypes = default_data_types, typename Downstream>

auto tcspc::batch_bin_increment_clusters	(	std::shared_ptr< bucket_source< typename DataTypes::bin_index_type > >	buffer_provider,
		arg::bucket_size< std::size_t >	bucket_size,
		arg::batch_size< std::size_t >	batch_size,
		Downstream	downstream )

Create a processor that collects bin increment clusters into encoded batches.

This is an optimized analogue of tcspc::batch() for the specific case of tcspc::bin_increment_cluster_event; it avoids allocating memory for each cluster individually. It must be paired with tcspc::unbatch_bin_increment_clusters().

The bucket_size should be large enough that all clusters (easily) fit in a single bucket (including the encoded cluster size). If a cluster takes up more than bucket_size when encoded, it will be emitted as a batch containing just that cluster. The buffer_provider needs to be prepared to handle this case, if it is expected.

Template Parameters

DataTypes	data type set specifying bin_index_type
Downstream	downstream processor type (usually deduced)

Parameters

buffer_provider	bucket source providing buffers for the batches
bucket_size	size of buckets to use; batches are emitted when the next cluster will not fit in the bucket
batch_size	maximum number of clusters to include in a batch; if 0, limit only by bucket size
downstream	downstream processor

Returns

processor

Events handled

• bin_increment_cluster_event<DataTypes>: collect and encode into tcspc::bucket<typename DataTypes::bin_index_type> and emit as batch
• All other types: pass through with no action
• Flush: emit any buffered clusters as tcspc::bucket<typename DataTypes::bin_index_type>; pass through

unbatch()

template<typename ContainerEvent, typename Downstream>

auto tcspc::unbatch

(

Downstream

downstream

)

Create a processor transforming batches of events to individual events.

Events in (ordered) containers or buckets are emitted one by one.

See also

tcspc::batch()

Template Parameters

ContainerEvent	the type of the batch event to unbatch
Downstream	downstream processor type

Parameters

downstream

downstream processor

Returns

processor

Events handled

• ContainerEvent: each element event emitted in order
• All other types: pass through with no action
• Flush: pass through with no action

unbatch_bin_increment_clusters()

template<typename DataTypes = default_data_types, typename Downstream>

auto tcspc::unbatch_bin_increment_clusters

(

Downstream

downstream

)

Create a processor that splits encoded batches of bin increment clusters into individual clusters.

This is an optimized analogue of tcspc::unbatch() for the specific case of tcspc::bin_increment_cluster_event. It must be paired with tcspc::batch_bin_increment_clusters().

Template Parameters

DataTypes	data type set specifying bin_index_type and the emitted events
Downstream	downstream processor type (usually deduced)

Parameters

downstream

downstream processor

Returns

processor

Events handled

• tcspc::bucket<typename DataTypes::bin_index_type> or tcspc::bucket<typename DataTypes::bin_index_type const>: decode and emit each cluster as bin_increment_cluster_event<DataTypes>
• All other types: pass through with no action
• Flush: pass through with no action