Programmiersprache C++: Mehr Sender für std::execution in C++26

Eine Sender Factory ist ein Algorithmus, der keine Sender als Parameter verwendet und einen Sender zurückgibt. Ich habe sie bereits in meinem letzten Artikel vorgestellt: C++26: Die Sender-Fabriken, Adapter und Consumer von std::execution.

Modernes C++ – Rainer Grimm

Rainer Grimm ist seit vielen Jahren als Softwarearchitekt, Team- und Schulungsleiter tätig. Er schreibt gerne Artikel zu den Programmiersprachen C++, Python und Haskell, spricht aber auch gerne und häufig auf Fachkonferenzen. Auf seinem Blog Modernes C++ beschäftigt er sich intensiv mit seiner Leidenschaft C++.

Der Großteil des folgenden Inhalts stammt aus dem Proposal P2300R10. Ich werde versuchen, es prägnanter darzustellen.

Mache den Unterschied

Lasst uns gemeinsam etwas Großes vollbringen: Vom 1. bis 24. Dezember spende ich die Hälfte des Geldes für die ALS-Forschung, wenn Sie eines meiner Mentoring-Programme buchen.

Hier finden Sie weitere Informationen über mich, die Struktur meiner Mentoring-Programme und die einzelnen Programme:

• Fundamentals for C++ Professionals
• Design Patterns and Architectural Patterns with C++
• C++20: Get the Details
• Concurrency with Modern C++
• Generic Programming (Templates) with C++
• Embedded Programming with Modern C++

Wenn Sie oder Ihr Team eine spezielle Auswahl aus meinem Mentoring-Programm wünschen, kontaktieren Sie mich bitte unter rainer.grimm@ModernesCpp.de. Wir finden eine Lösung.

Sender-Adapter

Ein Sender-Adapter ist ein Algorithmus, der einen oder mehrere Sender als Parameter verwendet und einen Sender zurückgibt.

Sender-Adapter sind „lazy“. Sender-Consumer wie this_thread::sync_wait starten Sender.

execution::let_*

execution::sender auto let_value(
    execution::sender auto input,
    std::invocable<values-sent-by(input)...> function
);

execution::sender auto let_error(
    execution::sender auto input,
    std::invocable<errors-sent-by(input)...> function
);

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

let_value ist then sehr ähnlich: Beim Start ruft es die bereitgestellte Funktion mit den vom Eingabesender als Argumente gesendeten Werten auf. Wenn der Absender von then zurückkehrt, sendet er jedoch genau das, was die Funktion am Ende zurückgibt – let_value erfordert, dass die Funktion einen Absender zurückgibt, und der zurückgegebene Absender sendet die vom Absender gesendeten Werte, die vom Rückruf zurückgegeben wurden.

Ein schönes Beispiel für let_value, let_error und let_stopped findet sich in der Prototypbibliothek stdexec.

Das folgende Beispiel aus einem GitHub-Repository von Nvidia zeigt das Hauptprogramm eines HTTP-Servers, der mehrere Anfragen gleichzeitig verarbeitet.

/*
 * 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;
}

Ein exec::static_thread_pool pool wird mit 8 Threads erstellt und die get_scheduler Memberfunktion wird aufgerufen, um ein Scheduler-Objekt sched zu erhalten.

Das Programm simuliert dann die Bearbeitung mehrerer Anfragen, indem es zehnmal über eine Schleife iteriert. Es erstellt für jede Iteration eine Pipeline von Operationen, um eingehende Anfragen in Antworten umzuwandeln. Ein ex::sender snd repräsentiert diese Pipeline.

Die Pipeline beginnt mit der Funktion schedule_request_start, die einen Sender erstellt, der den Beginn der Bearbeitung einer neuen Anfrage darstellt. Die Anfrage wird dann mit der Funktion ex::let_value validiert, die die validate_request anwendet. Wenn die Anfrage nicht gültig ist, wird eine Ausnahme ausgelöst.

Als Nächstes wird die Anfrage mit der Funktion ex::let_value verarbeitet, die die Funktion handle_request anwendet. Wenn während der Verarbeitung Fehler auftreten, werden diese mithilfe der Funktion ex::let_error in korrekte Antworten umgewandelt, wobei die Funktion error_to_response angewendet wird. Wenn der Ablauf abgebrochen wird, wird mithilfe der Funktion, die die Funktion stopped_to_response nutzt, eine korrekte Antwort zurückgesendet. Schließlich wird das Ergebnis mithilfe der Funktion ex::let_value , die die Funktion send_response anwendet, an den Client zurückgeschrieben.

Der Aufruf ex::start_detached trennt die Ausführung vom aktuellen Thread.

execution::into_variant

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

gibt einen Sender zurück, der eine Variante von Tupeln aller möglichen Mengen von Datentypen sendet, die vom Eingabesender gesendet wurden.

execution::stopped_as_optional

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

gibt einen Sender zurück, der den Wertkanal von einem T auf ein optional<decay_t<T>> und den gestoppten Kanal auf einen Wert eines leeren optional<decay_t<T>> abbildet.

execution::stopped_as_error

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

gibt einen Sender zurück, der den gestoppten Kanal einem Fehler von err zuordnet.

execution::bulk

execution::sender auto bulk(
    execution::sender auto input,
    std::integral auto shape,
    invocable<decltype(size), values-sent-by(input)...> function
);

gibt einen Sender zurück, der den aufrufbaren call beschreibt, der für input gemäß shape aufgerufen wird.

execution::split

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

Wenn der bereitgestellte Sender ein Mehrfach-Sender ist, gib diesen Sender zurück. Andernfalls gib einen Mehrfach-Sender zurück, der Werte sendet, die denen entsprechen, die vom bereitgestellten Sender gesendet wurden.

Einige Sender unterstützen möglicherweise nur den einmaligen Start ihres Vorgangs, während andere wiederholbar sind.

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 gibt einen Sender zurück, der abgeschlossen ist, sobald alle Eingabesender abgeschlossen sind. when_all_with_variant macht dasselbe, passt aber alle Eingabesender mithilfe von into_variant an und schränkt die Eingabeargumente daher nicht wie when_all ein.

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

Ein Sender Consumer ist ein Algorithmus, der einen oder mehrere Sender als Parameter verwendet und keinen Sender zurückgibt.

this_thread::sync_wait

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

this_thread::sync_wait ist ein Sender Consumer, der die vom bereitgestellten Sender beschriebene Arbeit zur Ausführung einreicht, den aktuellen std::thread oder Thread von main blockiert, bis die Arbeit abgeschlossen ist, und nach Abschluss der Arbeit ein optionales Tupel von Werten zurückgibt, die vom bereitgestellten Sender gesendet wurden. sync_wait ist eine Möglichkeit, die Domäne des Senders zu verlassen und das Ergebnis des Task-Graphen abzurufen.

Wenn der angegebene Sender einen Fehler anstelle von Werten sendet, löst sync_wait diesen Fehler als Ausnahme aus oder löst die ursprüngliche Ausnahme erneut aus, wenn der Fehler vom Datentyp std::exception_ptr ist.

Wenn der angegebene Sender das Signal "stopped" anstelle von Werten sendet, gibt sync_wait ein leeres optionales Element zurück.

Kurze Weihnachtspause

Ich werde eine zweiwöchige Weihnachtspause einlegen. Mein nächster Artikel wird am 13. Januar veröffentlicht. (rme)