C++ programming language: more channels for std::execution in C++26

Sender Factory is an algorithm, Which does not take senders as parameters and returns the sender. I already introduced them in my previous article: C++26: std::execution’s dispatcher factories, adapters, and consumers.

Advertisement

Most of the following material comes from the proposal P2300R10I will try to present it more concisely.

make the difference

Let’s accomplish some great work together: From December 1st to 24th, if you book one of my mentorship events, I’ll donate half the money to ALS research.

Here you can find more information about it MeThe structure My Consulting Programs and Personal Programs:

If you or your team would like a special selection from my consulting program, please contact me at rainer.grimm@ModernesCpp.de. We will find a solution.

transmitter adapter

A sender adapter is an algorithm that takes one or more senders as parameters and returns a sender.

Product Employees: Agile is dead – what does this mean for Product Owners?

The transmitter adapters are “lazy”. Sender consumers like this_thread::sync_wait Start the transmitter.

execution::let_*

execution::sender auto let_value(
    execution::sender auto input,
    std::invocable function
);

execution::sender auto let_error(
    execution::sender auto input,
    std::invocable function
);

execution::sender auto let_stopped(
    execution::sender auto input,
    std::invocable auto function
);

let_value Is then Very similar: on startup it calls the given function with the values ​​sent by the input sender as arguments. If the sender then However, this returns exactly what the function returns at the end – let_value The function is required to return a sender, and the returned sender sends the value returned by the callback to the sender.

a good example of let_value, let_error And let_stopped Can be found in the Prototype Library stdexec,

following example From the Nvidia GitHub repository Shows the main program of an HTTP server that processes multiple requests at the same time.

/*
 * Copyright (c) 2022 Lucian Radu Teodorescu
 *
 * Licensed under the Apache License Version 2.0 with LLVM Exceptions
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *   https://llvm.org/LICENSE.txt
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 int main() {
  // Create a thread pool and get a scheduler from it
  exec::static_thread_pool pool{8};
  ex::scheduler auto sched = pool.get_scheduler();

  // Fake a couple of requests
  for (int i = 0; i < 10; i++) {
    // The whole flow for transforming incoming requests into responses
    ex::sender auto snd =
      // get a sender when a new request comes
      schedule_request_start(sched, i)
      // make sure the request is valid; throw if not
      | ex::let_value(validate_request)
      // process the request in a function that may be using a different execution context
      | ex::let_value(handle_request)
      // If there are errors transform them into proper responses
      | ex::let_error(error_to_response)
      // If the flow is cancelled, send back a proper response
      | ex::let_stopped(stopped_to_response)
      // write the result back to the client
      | ex::let_value(send_response)
      // done
      ;

    // execute the whole flow asynchronously
    ex::start_detached(std::move(snd));
  }

  pool.request_stop();

  return 0;
}

A exec::static_thread_pool pool Made from 8 threads and get_scheduler Member function called on scheduler object sched To get.

The program then simulates processing multiple requests by iterating over a loop ten times. It creates a pipeline of operations for each iteration to transform incoming requests into responses. A ex::sender snd This represents the pipeline.

Pipeline starts from function schedule_request_startWhich creates a dispatcher that represents the beginning of processing a new request. The request is then processed using the function ex::let_value certified that validate_request come into force. If the request is not valid, an exception is thrown.

Next is the request with function ex::let_value processed by function handle_request come into force. If errors occur during processing, they are corrected using the function ex::let_error Converted to correct answers using the function error_to_response is applied. If the flow is canceled, it will be processed using a function that does stopped_to_response uses, a correct answer is returned. Finally the result is obtained using the function ex::let_value that work send_response Implemented, written back to customer.

Call ex::start_detached Detaches execution from the current thread.

execution::into_variant

execution::sender auto into_variant(
    execution::sender auto snd
);

Returns a sender that sends an array of tuples of all possible sets of data types sent by the input sender.

execution::stopped_as_optional

execution::sender auto stopped_as_optional(
    single-sender auto snd
);

Returns a sender that is the value channel of a T on one optional> And the channel is closed with the value of empty. optional> Shows.

execution::stopped_as_error

template
execution::sender auto stopped_as_error(
    execution::sender auto snd,
    Error err
);

Returns a transmitter that has an error on a closed channel err Assigned.

execution::bulk

execution::sender auto bulk(
    execution::sender auto input,
    std::integral auto shape,
    invocable function
);

Returns a dispatcher that is callable call describes one for input according to shape It is said.

execution::split

execution::sender auto split(execution::sender auto sender);

If the transmitter provided is a multiple transmitter, return that transmitter. Otherwise, return a multiple sender that sends values ​​equal to the values ​​sent by the provided sender.

Some channels may support starting their process only once, while others may be repeatable.

execution::when_all*

execution::sender auto when_all(
    execution::sender auto ...inputs
);

execution::sender auto when_all_with_variant(
    execution::sender auto ...inputs
);

when_all Returns a sender that is complete after all input senders have been completed. when_all_with_variant Works the same but matches all input senders using into_variant and therefore does not restrict input arguments when_all One.

execution::scheduler auto sched = thread_pool.scheduler();

execution::sender auto sends_1 = ...;
execution::sender auto sends_abc = ...;

execution::sender auto both = execution::when_all(
    sends_1,
    sends_abc
);

execution::sender auto final = execution::then(both, ()(auto... args){
    std::cout << std::format("the two args: {}, {}", args...);
});
// when final executes, it will print "the two args: 1, abc"

sender consumer

A sender consumer is an algorithm that takes one or more senders as parameters and does not return the sender.

this_thread::sync_wait

auto sync_wait(
    execution::sender auto sender
) requires (always-sends-same-values(sender))
    -> std::optional<:tuple>>;

this_thread::sync_wait A sender is a consumer that submits for execution a task described by a given sender. current one std::thread thread from or main Blocks until the task completes, and when the task completes returns an optional tuple of the values ​​sent by the provided sender. sync_wait There is a way to set the sender’s domain Holiday And get the result of function graph.

Resolved if the specified sender sends an error instead of the values sync_wait If the error is of data type then throw this error as an exception or rethrow the original exception. std::exception_ptr Is.

Returns if the specified transmitter sends a “stopped” signal instead of the values. sync_wait Returns an empty optional element.

short christmas holiday

I will take two weeks off for Christmas. My next article will be published on January 13.

(rme)