{"id":10481,"date":"2024-12-23T10:40:08","date_gmt":"2024-12-23T10:40:08","guid":{"rendered":"https:\/\/www.modernescpp.com\/?p=10481"},"modified":"2025-07-03T14:18:14","modified_gmt":"2025-07-03T14:18:14","slug":"stdexecution-more-senders","status":"publish","type":"post","link":"https:\/\/www.modernescpp.com\/index.php\/stdexecution-more-senders\/","title":{"rendered":"std::execution: More Senders"},"content":{"rendered":"\n
\"\"<\/figure>\n\n\n\n

std::execution <\/code>offers three types of senders: factories, adapters, and consumers. I’ll take a closer look at these today. <\/p>\n\n\n\n

Most of the following content is from proposal P2300R10<\/a>. I will try to represent it more concisely.<\/p>\n\n\n\n

<\/span>Sender Factory<\/span><\/h2>\n\n\n\n

A sender factory is an algorithm <\/em>that takes no senders as parameters and returns a sender. I already presented them in my last post: std::execution: Sender<\/a><\/code>.<\/p>\n\n\n\n

<\/span>Sender Adaptor <\/span><\/h2>\n\n\n\n

A sender adaptor is an algorithm that takes one or more senders as parameters and returns a sender.<\/p>\n\n\n\n

Sender adaptors are lazy. Sender consumers such as this_thread::sync_wait <\/code>start senders.<\/p>\n\n\n\n

<\/p>\n\n\n\n

<\/span>execution::let_*<\/code><\/span><\/h3>\n\n\n\n
execution::<\/span>sender auto<\/span> let_value(\n    execution::<\/span>sender auto<\/span> input,\n    std::<\/span>invocable<<\/span>values-<\/span>sent-<\/span>by(input)...><\/span> function\n);\n\nexecution::<\/span>sender auto<\/span> let_error(\n    execution::<\/span>sender auto<\/span> input,\n    std::<\/span>invocable<<\/span>errors-<\/span>sent-<\/span>by(input)...><\/span> function\n);\n\nexecution::<\/span>sender auto<\/span> let_stopped(\n    execution::<\/span>sender auto<\/span> input,\n    std::<\/span>invocable auto<\/span> function\n);\n<\/pre><\/div>\n\n\n\n
let_value<\/code> is very similar to then<\/code>: when it is started, it invokes the provided function with the values sent by the input sender as arguments. However, where the sender returned from then<\/code> sends exactly what that function ends up returning – let_value<\/code> requires that the function return a sender, and the sender returned by let_value<\/code> sends the values sent by the sender returned from the callback. <\/p>\n\n\n\n
A nice example about let_value, let_error<\/code>, and let_stopped<\/code> has the prototype library stdexec<\/a>.<\/p>\n\n\n\n
The following example shows the main program of an HTTP server that handles multiple requests concurrently. (https:\/\/github.com\/NVIDIA\/stdexec\/blob\/main\/examples\/server_theme\/let_value.cpp<\/a>)<\/a><\/p>\n\n\n\n
\/*<\/span>\n * Copyright (c) 2022 Lucian Radu Teodorescu<\/span>\n *<\/span>\n * Licensed under the Apache License Version 2.0 with LLVM Exceptions<\/span>\n * (the "License"); you may not use this file except in compliance with<\/span>\n * the License. You may obtain a copy of the License at<\/span>\n *<\/span>\n *   https:\/\/llvm.org\/LICENSE.txt<\/span>\n *<\/span>\n * Unless required by applicable law or agreed to in writing, software<\/span>\n * distributed under the License is distributed on an "AS IS" BASIS,<\/span>\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.<\/span>\n * See the License for the specific language governing permissions and<\/span>\n * limitations under the License.<\/span>\n *\/<\/span>\n int<\/span> main<\/span>() {\n  \/\/ Create a thread pool and get a scheduler from it<\/span>\n  exec::<\/span>static_thread_pool pool{8<\/span>};\n  ex::<\/span>scheduler auto<\/span> sched =<\/span> pool.get_scheduler();\n\n  \/\/ Fake a couple of requests<\/span>\n  for<\/span> (int<\/span> i =<\/span> 0<\/span>; i <<\/span> 10<\/span>; i++<\/span>) {\n    \/\/ The whole flow for transforming incoming requests into responses<\/span>\n    ex::<\/span>sender auto<\/span> snd =<\/span>\n      \/\/ get a sender when a new request comes<\/span>\n      schedule_request_start(sched, i)\n      \/\/ make sure the request is valid; throw if not<\/span>\n      |<\/span> ex::<\/span>let_value(validate_request)\n      \/\/ process the request in a function that may be using a different execution context<\/span>\n      |<\/span> ex::<\/span>let_value(handle_request)\n      \/\/ If there are errors transform them into proper responses<\/span>\n      |<\/span> ex::<\/span>let_error(error_to_response)\n      \/\/ If the flow is cancelled, send back a proper response<\/span>\n      |<\/span> ex::<\/span>let_stopped(stopped_to_response)\n      \/\/ write the result back to the client<\/span>\n      |<\/span> ex::<\/span>let_value(send_response)\n      \/\/ done<\/span>\n      ;\n\n    \/\/ execute the whole flow asynchronously<\/span>\n    ex::<\/span>start_detached(std::<\/span>move(snd));\n  }\n\n  pool.request_stop();\n\n  return<\/span> 0<\/span>;\n}\n<\/pre><\/div>\n\n\n\n
An exec::static_thread_pool pool <\/code>is created with 8 threads, and the get_scheduler<\/code> method is called to obtain a scheduler object named sched<\/code>.<\/p>\n\n\n\n
The program then simulates handling multiple requests by iterating over a loop 10 times. It constructs a pipeline of operations for each iteration to transform incoming requests into responses. An ex::sender<\/code> snd <\/code>represents this pipeline.<\/p>\n\n\n\n
The pipeline starts with the schedule_request_start<\/code> function, which creates a sender that represents the start of handling a new request. The request is then validated using the ex::let_value<\/code> function, which applies the validate_request<\/code>. If the request is not valid, an exception is thrown.<\/p>\n\n\n\n
Next, the request is processed using the ex::let_value<\/code> function that applies the handle_request function. If any errors occur during processing, they are transformed into proper responses using the ex::let_error<\/code> function applying the error_to_response <\/code>function. If the flow is cancelled, a proper response is sent back using the ex::let_stopped <\/code>function applying the stopped_to_response <\/code>function. Finally, the result is written back to the client using the ex::let_value<\/code> function, which applies the send_response<\/code> function.<\/p>\n\n\n\n
The ex::start_detached<\/code> call detaches the execution from the current thread.<\/p>\n\n\n\n
<\/span>execution::into_variant<\/code><\/span><\/h3>\n\n\n\n
execution::<\/span>sender auto<\/span> into_variant(\n    execution::<\/span>sender auto<\/span> snd\n);\n<\/pre><\/div>\n\n\n\n
Returns a sender which sends a variant of tuples of all the possible sets of types sent by the input sender. <\/p>\n\n\n\n
<\/span>execution::stopped_as_optional<\/code><\/span><\/h3>\n\n\n\n
execution::<\/span>sender auto<\/span> stopped_as_optional(\n    single-<\/span>sender auto<\/span> snd\n);\n<\/pre><\/div>\n\n\n\n
Returns a sender that maps the value channel from a T<\/code> to an optional<decay_t<T>><\/code>, and maps the stopped channel to a value of an empty optional<decay_t<T>><\/code>.<\/p>\n\n\n\n
<\/span>execution::stopped_as_error<\/code><\/span><\/h3>\n\n\n\n
template<\/span><<\/span>move_constructible Error><\/span>\nexecution::<\/span>sender auto<\/span> stopped_as_error(\n    execution::<\/span>sender auto<\/span> snd,\n    Error err\n);\n<\/pre><\/div>\n\n\n\n
Returns a sender that maps the stopped channel to an error of err<\/code>.<\/p>\n\n\n\n
<\/span>execution::bulk<\/code><\/span><\/h3>\n\n\n\n
execution::<\/span>sender auto<\/span> bulk(\n    execution::<\/span>sender auto<\/span> input,\n    std::<\/span>integral auto<\/span> shape,\n    invocable<<\/span>decltype(size), values-<\/span>sent-<\/span>by(input)...><\/span> function\n);\n<\/pre><\/div>\n\n\n\n
Returns a sender that describes the callable call<\/code> invoked on input <\/code>according to shape.<\/code><\/p>\n\n\n\n
<\/span>execution::split<\/code><\/span><\/h3>\n\n\n\n
execution::<\/span>sender auto<\/span> split(execution::<\/span>sender auto<\/span> sender);\n<\/pre><\/div>\n\n\n\n
 If the provided sender is a multi-shot sender, return that sender. Otherwise, return a multi-shot sender that sends values equivalent to those sent by the provided sender. <\/p>\n\n\n\n
Some senders may only support launching their operation once, while others may be repeatable.<\/p>\n\n\n\n
<\/span>execution::when_all<\/code>*<\/span><\/h3>\n\n\n\n
execution::<\/span>sender auto<\/span> when_all(\n    execution::<\/span>sender auto<\/span> ...inputs\n);\n\nexecution::<\/span>sender auto<\/span> when_all_with_variant(\n    execution::<\/span>sender auto<\/span> ...inputs\n);\n<\/pre><\/div>\n\n\n\n
when_all<\/code> returns a sender that completes once all of the input senders have completed. when_all_with_variant<\/code> does the same, but it adapts all the input senders using into_variant<\/code>, and so it does not constrain the input arguments as when_all<\/code> does.<\/p>\n\n\n\n
execution::<\/span>scheduler auto<\/span> sched =<\/span> thread_pool.scheduler();\n\nexecution::<\/span>sender auto<\/span> sends_1 =<\/span> ...;\nexecution::<\/span>sender auto<\/span> sends_abc =<\/span> ...;\n\nexecution::<\/span>sender auto<\/span> both =<\/span> execution::<\/span>when_all(\n    sends_1,\n    sends_abc\n);\n\nexecution::<\/span>sender auto<\/span> final =<\/span> execution::<\/span>then(both, [](auto<\/span>... args){\n    std::<\/span>cout <<<\/span> std::<\/span>format("the two args: {}, {}"<\/span>, args...);\n});\n\/\/ when final executes, it will print "the two args: 1, abc"<\/span>\n<\/pre><\/div>\n\n\n\n
<\/span>Sender Consumer <\/span><\/h2>\n\n\n\n
A sender consumer is an algorithm that takes one or more senders as parameters and does not return a sender.<\/em><\/p>\n\n\n\n
<\/span>this_thread::sync_wait<\/code><\/span><\/h3>\n\n\n\n
auto<\/span> sync_wait(\n    execution::<\/span>sender auto<\/span> sender\n) requires (always-<\/span>sends-<\/span>same-<\/span>values(sender))\n    -><\/span> std::<\/span>optional<<\/span>std::<\/span>tuple<<\/span>values-<\/span>sent-<\/span>by(sender)>><\/span>;\n<\/pre><\/div>\n\n\n\n
this_thread::sync_wait<\/code> is a sender consumer that submits the work described by the provided sender for execution, blocking the current std::thread<\/code> or thread of main<\/code><\/strong> until the work is completed, and returns an optional tuple of values sent by the provided sender on its completion of work. sync_wait<\/code> is one way to exit<\/em> the senders domain, retrieving the task graph result.<\/p>\n\n\n\n
If the provided sender sends an error instead of values, sync_wait<\/code> throws that error as an exception, or rethrows the original exception if the error is of type std::exception_ptr<\/code>.<\/p>\n\n\n\n
If the provided sender sends the “stopped” signal instead of values, sync_wait<\/code> returns an empty optional.<\/p>\n\n\n\n
<\/span>A Short Christmas Break<\/span><\/h2>\n\n\n\n
I will take a two-week Christmas break. My next post will be published on January 13th.<\/p>\n","protected":false},"excerpt":{"rendered":"
std::execution offers three types of senders: factories, adapters, and consumers. I’ll take a closer look at these today. Most of the following content is from proposal P2300R10. I will try to represent it more concisely. Sender Factory A sender factory is an algorithm that takes no senders as parameters and returns a sender. I already […]<\/p>\n","protected":false},"author":21,"featured_media":10445,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[559],"tags":[561],"class_list":["post-10481","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-c26-blog","tag-execution"],"_links":{"self":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/10481","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/users\/21"}],"replies":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/comments?post=10481"}],"version-history":[{"count":16,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/10481\/revisions"}],"predecessor-version":[{"id":10522,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/posts\/10481\/revisions\/10522"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/media\/10445"}],"wp:attachment":[{"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/media?parent=10481"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/categories?post=10481"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.modernescpp.com\/index.php\/wp-json\/wp\/v2\/tags?post=10481"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}