std::execution::bulk, std::execution::bulk_chunked, std::execution::bulk_unchunked

 

C++

Compiler support

Freestanding and hosted

Language

Standard library

Standard library headers

Named requirements

Feature test macros (C++20)

Language support library

Concepts library (C++20)

Diagnostics library

Memory management library

Metaprogramming library (C++11)

General utilities library

Containers library

Iterators library

Ranges library (C++20)

Algorithms library

Strings library

Text processing library

Numerics library

Date and time library

Input/output library

Filesystem library (C++17)

Concurrency support library (C++11)

Execution control library (C++26)

Technical specifications

Symbols index

External libraries

 

Execution control library

Execution concepts scheduler receiver receiver_of operation_state    sender sender_in sender_to Execution components Execution contexts run_loop Execution domains default_domain Forward progress guarantee forward_progress_guarantee Environments empty_env    get_env    env_of_t Queries std::stop_token_of_t std::get_stop_token std::get_allocator std::forwarding_query get_scheduler get_domain get_forward_progress_guarantee get_delegation_scheduler get_completion_scheduler Completion signatures completion_signatures completion_signatures_of_t transform_completion_signatures transform_completion_signatures_of    value_types_of_t error_types_of_t    tag_of_t sends_stopped    Sender adaptor closures sender_adaptor_closure Coroutine utility as_awaitable with_awaitable_senders

Core operations Operation state connect connect_result_t start Completion set_value set_error set_stopped Transformation transform_env transform_sender apply_sender Sender algorithms Sender factories just just_error just_stopped read_env schedule schedule_result_t Sender adaptors starts_on continues_on    on then upon_error upon_stopped let_value let_error let_stopped schedule_from when_all when_all_with_variant into_variant stopped_as_optional stopped_as_error bulk split Sender consumers this_thread::sync_wait this_thread::sync_wait_with_variant

 

Defined in header <execution>
std::execution::sender auto bulk( std::execution::sender auto input,            std::integral auto size,            auto&& policy,            std::invocable<decltype(size),                           /*values-sent-by*/(input)...> function );	(1)	(since C++26)
std::execution::sender auto bulk_chunked( std::execution::sender auto input,                    std::integral auto size,                    auto&& policy,                    std::invocable<decltype(size), decltype(size),                                   /*values-sent-by*/(input)...> function2 );	(2)	(since C++26)
std::execution::sender auto bulk_unchunked( std::execution::sender auto input,                      std::integral auto size,                      std::invocable<decltype(size), decltype(size),                                     /*values-sent-by*/(input)...> function );	(3)	(since C++26)

Parameters

input	-	sender which once executed sends the values upon which the function executes
policy	-	the execution policy attached to function/function2
function	-	invocable to be called for every index in range [​0​, size), passing also the values produced by the input sender
function2	-	same as function but called with a pair of indices (b, e), with b < e, so that, for each index i in range [[0, size) there is exactly one call to function2 so that b <= i < e.

Return value

Returns a sender describing the task graph described by the input sender, with an added node of invoking the provided function with indices in range [​0​, size), passing also the values sent by the input sender as arguments.

function/function2 is guaranteed to not begin executing until the returned sender is started.

Error completions

All errors passed in by input are forwarded.

In addition, the sender is allowed to complete with an std::exception_ptr error that contains:

• any exception thrown by function
• std::bad_alloc if the implementation fails to allocate required resources
• an exception derived from std::runtime_error for other internal errors (e.g., cannot propagate the exception from the execution context to the caller).

Cancellation

The uncustomized std::execution::bulk, std::execution::bulk_chunk and std::execution::bulk_unchunked forward the stopped completion signal from input. They do not provide additional mechanism to produce stopped completion signal.

Notes

When calling std::execution::bulk and std::execution::bulk_chunked, different invocations of function/function2 may happen on the same execution agent.

When calling std::execution::bulk_unchunked, different invocations of function must happen on different execution agents.

The default implementation of std::execution::bulk is based on std::execution::bulk_chunked. While customizing std::execution::bulk is possible, it is expected that most of the time only std::execution::bulk_chunked is customized.

Without a customization of std::execution::bulk and std::execution::bulk_chunked, the behavior of std::execution::bulk and std::execution::bulk_chunk is to execute function serially, which is not particularly useful. Implementations are expected to have customizations that would make running std::execution::bulk and std::execution::bulk_chunked on different schedulers more useful.

std::execution::bulk_unchunked is meant to be used whenever function may have dependencies between different invocations, and it requires concurrent forward progress guarantees (parallel forward progress is not enough). Running std::execution::bulk_unchunked with a size of 1000 will require 1000 execution agents (e.g., threads) to run concurrently.

std::execution::bulk_unchunked does not require an execution policy, as is already expected for function to be able to run concurrently.

Examples

Possible usage of execution::bulk.

std::vector<double> x;
std::vector<double> y;
//...
sender auto process_elements
    = just(get_coefficient())
    | bulk(x.size(), [&](size_t i, double a)
    {
        y[i] = a * x[i] + y[i];
    });
// process_elements describes the work described by calling a function to
// get a coefficient `a`, and using it to execute
//   y[i] = a * x[i] + y[i]
// for each `i` in range [0, x.size())

Possible usage of execution::bulk_chunked.

std::vector<std::uint32_t> data = ...;
std::atomic<std::uint32_t> sum{0};
sender auto s = bulk_chunked(just(), par, 100000,
    [&sum, &data](int begin, int end)
    {
        auto partial_sum = std::accumulate(data.begin() + begin, data.begin() + end, 0U);
        sum.fetch_add(partial_sum);
    });
// the atomic object will not be touched 100000 times; will execute faster than bulk()