pipeline.h File Reference

#include <limits>
#include <dispenso/detail/pipeline_impl.h>

Go to the source code of this file.

Functions
template<typename F >
auto	dispenso::stage (F &&f, ssize_t limit)
template<typename... Stages>
void	dispenso::pipeline (ThreadPool &pool, Stages &&... sIn)
template<typename... Stages>
void	dispenso::pipeline (Stages &&... sIn)

Variables
constexpr ssize_t	dispenso::kStageNoLimit = std::numeric_limits<ssize_t>::max()

Detailed Description

A file providing utilities for parallel pipelining of work.

Definition in file pipeline.h.

Function Documentation

pipeline() [1/2]

template<typename... Stages>

void dispenso::pipeline

(

Stages &&...

sIn

)

Pipeline work in stages. Pipelines allow stages to specify parallelism limits by using the stage function, or a function-like object can simply be passed directly, indicating a serial stage. Even if stages are serial, there can be parallelism between stages, so in a 3 stage serial pipeline, the expected runtime is the max of the 3 stages runtimes (note that this is in the absence of pipeline overheads and with an infinitely long workstream. In practice speedup is somewhat less). Work will be run on dispenso's global thread pool. This function will block until the entire pipeline has completed.

Parameters

sIn

The stages to run. The first stage must be a Generator stage, the last must be a Sink stage, and intermediate stages are Transform stages. If there is only one stage, it takes no arguments, but returns a bool indicating completion (false means the pipeline is complete). Otherwise, the Generator stage takes no arguments and must return an OpResult or std::optional value, and an invalid/nullopt result indicates that the Generator is done (no more values forthcoming). Transform stages should accept the output of the prior stage (or output.value() in the case of OpResult or std::optional), and should return either a value or an OpResult or std::optional value if the Transform is capable of filtering results. Invalid/nullopt OpResult or std::optional values indicate that the value should be filtered, and not passed on to the next stage. The Sink stage should accept the output of the prior stage, just as a Transform stage does, but does not return any value (or at least the pipeline will ignore it).

Note

See the ThreadPool overload of pipeline() for exception behavior details.

Definition at line 122 of file pipeline.h.

pipeline() [2/2]

template<typename... Stages>

void dispenso::pipeline	(	ThreadPool &	pool,
		Stages &&...	sIn )

Pipeline work in stages. Pipelines allow stages to specify parallelism limits by using the stage function, or a function-like object can simply be passed directly, indicating a serial stage. Even if stages are serial, there can be parallelism between stages, so in a 3 stage serial pipeline, the expected runtime is the max of the 3 stages runtimes (note that this is in the absence of pipeline overheads and with an infinitely long workstream. In practice speedup is somewhat less). This function will block until the entire pipeline has completed.

Parameters

pool

The ThreadPool to run the work in. This inherently determines the upper bound for parallelism of the pipeline.

sIn

The stages to run. The first stage must be a Generator stage, the last must be a Sink stage, and intermediate stages are Transform stages. If there is only one stage, it takes no arguments, but returns a bool indicating completion (false means the pipeline is complete). Otherwise, the Generator stage takes no arguments and must return an OpResult or std::optional value, and an invalid/nullopt result indicates that the Generator is done (no more values forthcoming). Transform stages should accept the output of the prior stage (or output.value() in the case of OpResult or std::optional), and should return either a value or an OpResult or std::optional value if the Transform is capable of filtering results. Invalid/nullopt OpResult or std::optional values indicate that the value should be filtered, and not passed on to the next stage. The Sink stage should accept the output of the prior stage, just as a Transform stage does, but does not return any value (or at least the pipeline will ignore it).

Note

Exception behavior: If a stage function throws an exception, the pipeline will stop producing new work (the generator checks for exceptions between iterations) and allow any already-in-flight work to complete. Queued work that has not yet started will be discarded without execution. The first captured exception is rethrown from pipeline() when the pipeline winds down. Subsequent exceptions from concurrently in-flight stages are silently discarded. All internal state (concurrency counters, resource slots, completion events) is cleaned up via RAII, so the thread pool remains in a valid state after an exception and may be reused for subsequent pipelines. When exceptions are disabled at compile time (__cpp_exceptions not defined), all exception-related checks are eliminated with zero overhead.

Definition at line 89 of file pipeline.h.

stage()

template<typename F >

auto dispenso::stage	(	F &&	f,
		ssize_t	limit )

Create a stage for use in the pipeline function.

Parameters

f	A function-like object that can accept the result of the previous stage (if any), and which produces the output for the next stage (if any).
limit	How many threads may concurrently run work for this stage. Values larger than the number of threads in the associated thread pool of the used ConcurrentTaskSet will be capped to the size of the pool.

Returns

A stage object suitable for pipelining.

Definition at line 49 of file pipeline.h.

Variable Documentation

kStageNoLimit

ssize_t dispenso::kStageNoLimit = std::numeric_limits<ssize_t>::max()

constexpr

A simple constant representing maximum parallelism for a stage. This number has no particular significance, and is simply here for convenience.

Definition at line 36 of file pipeline.h.