18. Nov 2024
Lesedauer 16 Min.
Asynchron
Rust-Kurs, Teil 8
Cooperative Scheduling in Rust-Anwendungen realisieren.
Rust ist eine neuartige, moderne Programmiersprache, die sich auf die Sicherheit, der Geschwindigkeit und auf die effiziente, fehlerfreie parallele Programmierung konzentriert. In dieser Artikelserie geht es um den Einstieg in Rust und nebenbei gelegentlich um einen Vergleich mit C/C++ sowie C#. Die zurückliegende Folge [1] hat sich näher mit Multi-Threading, Channels und parallelen Iteratoren beschäftigt. Die heutige Folge erklärt, wie Sie die asynchrone Programmierung in Rust realisieren können, mit der Sie in einer anderen Art und Weise mehrere Code-Pfade parallel ausführen können. Zunächst werden ein paar wichtige Begriffe vorgestellt.
Parallelism, Concurrency, Preemptive- und Cooperative Scheduling
Die beiden Begriffe Parallelism und Concurrency werden sehr oft synonym verwendet, obwohl es sich um unterschiedliche Konzepte handelt. Parallelism bedeutet, dass mehrere Tasks ihre Arbeit parallel zueinander verrichten. Die vorangegangene Folge hat hierzu das Konzept des Multi-Threading vorgestellt, mit dem das Parallelisieren von Threads auf einem Multi-Core-System einfach realisiert werden kann.Concurrency bedeutet auch, dass mehrere Tasks ausgeführt werden, die aber zeitlich zueinander versetzt ablaufen können. Bild 1 veranschaulicht diesen wichtigen Unterschied.
Concurrency versus Paralellism (Bild 1)
Autor
Um eine Concurrency von Tasks erreichen zu können, müssen Sie in der Lage sein, Tasks zu stoppen und ihren aktuellen Status zu speichern, damit Sie den Task zu einem späteren Zeitpunkt fortführen können.Das Betriebssystem stellt spezielle Multitasking-Funktionalitäten zur Verfügung, mit deren Hilfe Parallelism und Concurrency von Code-Pfaden realisiert werden können. Programmiersprachen wie Rust setzen auf diese Funktionalitäten auf. Multitasking kann hierbei auf zwei unterschiedliche Arten durchgeführt werden: per Cooperative Scheduling oder Preemptive Scheduling.Beim Preemptive Scheduling führt das Betriebssystem in regelmäßigen Abständen (unter Windows etwa alle vier Millisekunden) sogenannte Kontext-Switches durch. Dabei wird der Status des aktuellen Threads – also alle aktuellen Registerinhalte – in einer Datenstruktur gesichert und anschließend der Status eines anderen Threads in die Register geladen, damit dieser weiter ausgeführt werden kann. Da hierfür sehr viele Assembly-Instruktionen benötigt werden, ist ein Kontext-Switch sehr zeitintensiv und kann unter Umständen zu Skalierungsproblemen führen.Stellen Sie sich beispielsweise bei einem Datenbank-Server vor, dass für jede eingehende Abfrage ein neuer Thread gestartet wird. Bei Tausenden von gleichzeitig ablaufenden Abfragen müsste das Betriebssystem ein Kontext-Switching für Tausende Threads durchführen. Es ist leicht ersichtlich, dass davon die Performance kaum profitieren kann.Jeder Thread besteht zusätzlich aus verschiedenen Stack-Frames, welche die verwendeten lokalen Variablen speichern und den aktuellen Callstack umfassen. Stack-Frames behält das Betriebssystems im Hauptspeicher und generiert somit zusätzlichen Overhead. Beispiel Windows: Hier hat ein Stack-Frame für einen Thread eine initiale Größe von einem Megabyte. Trotz dieser Nachteile ist das Preemptive Scheduling die Multitasking-Variante, die moderne Betriebssysteme aktuell einsetzen.Neben dem Preemptive Scheduling gibt es aber noch das Cooperative Scheduling, bei dem das Scheduling primär von den Anwendungen selbst realisiert wird. Das heißt, dass eine Anwendung einen Code-Pfad für eine bestimmte Zeitdauer ausführt und nach Ablauf der Zeitdauer die CPU von sich aus freigibt, damit ein anderer Code-Pfad ausgeführt werden kann. Das Freigeben der CPU wird als CPU-Yielding bezeichnet.Da in diesem Verfahren keine Kontext-Switches seitens des Betriebssystems durchgeführt werden müssen, ist das Cooperative Scheduling um einiges schneller als das Preemptive Scheduling. Das Problem ist jedoch, dass Anwendungen selbst für das Freigeben der CPU verantwortlich sind. Dadurch kann ein einfacher Programmierfehler (das fehlende regelmäßige CPU-Yielding) das komplette Betriebssystem zum Absturz bringen. Diese Variante des Schedulings hat Microsoft unter Windows 3.1 eingesetzt – also vor einer sehr langen Zeit.Nebenbei sei noch ein weiteres Beispiel erwähnt: Der SQL Server implementiert intern ebenfalls ein Cooperative Scheduling zum Ausführen von Datenbank-Abfragen. Der SQL Server umgeht dabei das Preemptive Scheduling des Betriebssystems und führt seine Worker-Threads selbstständig aus. Daraus folgt auch, dass die Code-Pfade innerhalb des SQL Servers eigenständig dafür Sorge tragen müssen, dass die CPU in regelmäßigen Intervallen freigegeben wird. Jedes Mal, wenn ein CPU-Yielding innerhalb der Implementierung des SQL Servers durchgeführt wird, wird für die betreffende Abfrage der Wartetyp SOS_SCHEDULER_YIELD zurückgeliefert, die Abfrage wird in den Status RUNNABLE verschoben und wartet dann auf freie CPU-Zeit. In solchen und ähnlichen Fällen ist das Cooperative Scheduling durchaus sinnvoll.Als Daumenregel kann gesagt werden, dass Preemptive Scheduling immer dann die richtige Wahl ist, wenn CPU-intensive Arbeiten durchzuführen sind. Dazu können Sie Threads nutzen, deren Verwendung Sie bereits in der vorangegangenen Folge dieser Serie kennengelernt haben.Besteht Ihr Code jedoch aus Funktionsaufrufen, die zu Blockaden führen können (weil die Code-Ausführung gestoppt werden muss, bis der Funktionsaufruf seitens des Betriebssystems vollständig durchgeführt wurde), ist der Einsatz von Cooperative Scheduling sinnvoller. Dazu zählen unter anderem:
- Netzwerkzugriffe,
- Datenbankzugriffe,
- Internetzugriffe,
- Interaktion mit dem Storage-Subsystem,
- Warten bis ein Timer abgelaufen ist.
Asynchrone Rust Runtimes
Das asynchrone Programmiermodell von Rust ist nichts Neues oder Außergewöhnliches. Vergleichbar mit den Funktionalitäten in Rust sind beispielsweise die von JavaScript angebotenen Promises [2] oder das Task Asynchronous Programming Model (TAP) von .NET [3].Eine Besonderheit der asynchronen Programmierung in Rust ist die Tatsache, dass Rust nur die notwendige Infrastruktur zur Verfügung stellt. Hierbei handelt es sich um die Trait Future [4] und die beiden Schlüsselwörter async und await, die Bestandteil des Rust-Compilers rustc sind.Eine asynchrone Runtime gehört jedoch nicht zu den Bestandteilen eines klassischen Rust-Programms. Sie müssen sie deshalb explizit über eine passende Crate in Ihrem Programm referenzieren (über die Datei cargo.toml). Durch diesen Ansatz wird sichergestellt, dass Rust-Binaries sehr klein bleiben, da eine asynchrone Runtime nur dann mitkompiliert wird, wenn die Anwendung sie auch benötigt.Tabelle 1 gibt einen Überblick über die bekanntesten asynchronen Runtimes für Rust, die immer wieder in unterschiedlichen Projekten eingesetzt werden.Tabelle 1: Asynchrone Runtimes für Rust
Alle Beispiele in diesem Artikel basieren auf der Runtime async-std, da diese für die ersten Schritte leicht verständlich und programmiertechnisch an die synchrone Standard-Library von Rust angelehnt ist. Für ernsthafte Rust-Projekte ist jedoch die Tokio-Runtime zu empfehlen, da diese aktiv von der Rust-Community weiterentwickelt wird und für sie laufend neue Releases bereitgestellt werden.Abhängig von der gewählten Runtime und deren Konfiguration kann diese asynchronen Code mit nur einem Worker-Thread (Single-Threaded) oder mehreren Worker-Threads (Multi-Threaded) ausführen. Im Hintergrund nutzt die Runtime dafür einen Thread-Pool, dessen Größe man ebenfalls anpassen kann.
Asynchrone Programmierung mit Rust
Als Beispiel für die asynchrone Programmierung in Rust dient hier die Ausführung von Abfragen auf einem SQL Server.Zum besseren Verständnis wird in Listing 1 zunächst exemplarischer Code gezeigt, der sich zwar nicht kompilieren lässt, aber demonstriert, wie der Zugriff auf den SQL Server mit normalen synchronen Funktionen funktionieren würde und wo die Performance-Probleme versteckt sind. Wie Sie im Listing erkennen können, interagiert dieser Code an drei Stellen mit dem SQL Server:Listing 1: Synchroner Zugriff auf den SQL Server
// Exemplarisch. Der Code lässt sich nicht <br/>// kompilieren!<br/>fn main()<br/>{<br/> // <span class="hljs-number">1</span>. Connect to SQL Server.<br/> <span class="hljs-built_in">let</span> sql_client = <br/> SQLClient::connect(connectionString);<br/> // <span class="hljs-number">2</span>. Execute a simple SELECT query.<br/> <span class="hljs-built_in">let</span> select_query = Query::<span class="hljs-built_in">new</span>(sql_statement);<br/> <span class="hljs-built_in">let</span> mut data_stream = <br/> select_query.query(&amp;client);<br/> // <span class="hljs-number">3</span>. Retrieve <span class="hljs-keyword">and</span> process each <span class="hljs-built_in">row</span>...<br/> <span class="hljs-keyword">while</span> (<span class="hljs-built_in">let</span> Some(<span class="hljs-built_in">row</span>) = data_stream.try_next()<br/> {<br/> println!(<span class="hljs-string">"{:?}"</span>, <span class="hljs-built_in">row</span>);<br/> }<br/>} - Im ersten Schritt wird eine neue Datenbankverbindung zum SQL Server aufgebaut.
- Im nächsten Schritt wird eine SELECT-Abfrage zum SQL Server gesendet.
- Danach werden die zurückgelieferten Datensätze in der Anwendung verarbeitet.
Listing 2: Die Datei cargo.toml
[<span class="hljs-name">package</span>]<br/>name = <span class="hljs-string">"sqlserver_async"</span><br/>version = <span class="hljs-string">"0.1.0"</span><br/>edition = <span class="hljs-string">"2021"</span><br/>[<span class="hljs-name">dependencies</span>]<br/>async-std =<br/>{<br/> version = <span class="hljs-string">"1.12.0"</span>, <br/> features = [<span class="hljs-string">"attributes"</span>]<br/>}<br/>anyhow = <span class="hljs-string">"1.0.0"</span><br/>futures-util = <span class="hljs-string">"0.3"</span><br/># Windows<br/>[<span class="hljs-name">target.</span><span class="hljs-symbol">'cfg</span>(<br/> target_os = <span class="hljs-string">"windows"</span>)<span class="hljs-symbol">'.dependencies</span>]<br/>tiberius =<br/>{<br/> version=<span class="hljs-string">"0.12.2"</span>, <br/> features = [<span class="hljs-string">"chrono"</span>, <span class="hljs-string">"time"</span>]<br/>}<br/># Mac OS<br/>[<span class="hljs-name">target.</span><span class="hljs-symbol">'cfg</span>(<br/> target_os = <span class="hljs-string">"macos"</span>)<span class="hljs-symbol">'.dependencies</span>]<br/>tiberius =<br/>{<br/> version=<span class="hljs-string">"0.12.2"</span>, <br/> default-features = false, <br/> features = [<span class="hljs-string">"chrono"</span>, <span class="hljs-string">"time"</span>, <br/> <span class="hljs-string">"vendored-openssl"</span>]<br/>} Listing 3: Asynchroner Zugriff
<span class="hljs-keyword">async</span> fn execute_query(sql_statement: <span class="hljs-built_in">String</span>, <br/> <span class="hljs-attr">duration</span>: u64) -&gt; anyhow::Result&lt;<span class="hljs-function"><span class="hljs-params">()</span>&gt;</span><br/><span class="hljs-function">{</span><br/><span class="hljs-function"> // <span class="hljs-params">Create</span> <span class="hljs-params">a</span> <span class="hljs-params">new</span> <span class="hljs-params">connection</span> <span class="hljs-params">string</span>.</span><br/><span class="hljs-function"> <span class="hljs-params">let</span> <span class="hljs-params">config</span> = <span class="hljs-params">Config</span>::<span class="hljs-params">from_ado_string</span>(</span><br/><span class="hljs-function"><span class="hljs-params"> CONNECTION_STRING</span>)?;</span><br/><span class="hljs-function"> // <span class="hljs-params">Create</span> <span class="hljs-params">a</span> <span class="hljs-params">new</span> <span class="hljs-params">TcpStream</span> <span class="hljs-params">to</span> <span class="hljs-params">the</span></span><br/><span class="hljs-function"> // <span class="hljs-params">SQL</span> <span class="hljs-params">Server</span> <span class="hljs-params">endpoint</span>.</span><br/><span class="hljs-function"> <span class="hljs-params">println</span>!(<span class="hljs-params"><span class="hljs-string">"Connecting to SQL Server..."</span></span>);</span><br/><span class="hljs-function"> <span class="hljs-params">let</span> <span class="hljs-params">socket</span> = <span class="hljs-params">TcpStream</span>::<span class="hljs-params">connect</span>(</span><br/><span class="hljs-function"><span class="hljs-params"> config.get_addr(</span>)).<span class="hljs-params">await</span>?;</span><br/><span class="hljs-function"> <span class="hljs-params">socket</span>.<span class="hljs-params">set_nodelay</span>(<span class="hljs-params"><span class="hljs-literal">true</span></span>)?;</span><br/><span class="hljs-function"> // <span class="hljs-params">Connect</span> <span class="hljs-params">to</span> <span class="hljs-params">SQL</span> <span class="hljs-params">Server</span>.</span><br/><span class="hljs-function"> <span class="hljs-params">let</span> <span class="hljs-params">mut</span> <span class="hljs-params">client</span> = <span class="hljs-params">Client</span>::<span class="hljs-params">connect</span>(</span><br/><span class="hljs-function"><span class="hljs-params"> config, socket</span>).<span class="hljs-params">await</span>?;</span><br/><span class="hljs-function"> <span class="hljs-params">println</span>!(<span class="hljs-params"><span class="hljs-string">"Successfully connected to</span></span></span><br/><span class="hljs-function"><span class="hljs-params"><span class="hljs-string"> SQL Server."</span></span>);</span><br/><span class="hljs-function"> // <span class="hljs-params">Execute</span> <span class="hljs-params">a</span> <span class="hljs-params">simple</span> <span class="hljs-params">query</span> <span class="hljs-params">against</span> <span class="hljs-params">SQL</span> <span class="hljs-params">Server</span>.</span><br/><span class="hljs-function"> <span class="hljs-params">let</span> <span class="hljs-params">select_query</span> = <span class="hljs-params">Query</span>::<span class="hljs-params">new</span>(</span><br/><span class="hljs-function"><span class="hljs-params"> &amp;sql_statement</span>);</span><br/><span class="hljs-function"> <span class="hljs-params">let</span> <span class="hljs-params">mut</span> <span class="hljs-params">data_stream</span> = <span class="hljs-params">select_query</span>.<span class="hljs-params">query</span>(</span><br/><span class="hljs-function"><span class="hljs-params"> &amp;mut client</span>).<span class="hljs-params">await</span>?;</span><br/><span class="hljs-function"> // <span class="hljs-params">Loop</span> <span class="hljs-params">over</span> <span class="hljs-params">the</span> <span class="hljs-params">received</span> <span class="hljs-params">data</span>.</span><br/><span class="hljs-function"> <span class="hljs-params">while</span> <span class="hljs-params">let</span> <span class="hljs-params">Some</span>(<span class="hljs-params">row</span>) = </span><br/><span class="hljs-function"> <span class="hljs-params">data_stream</span>.<span class="hljs-params">try_next</span><span class="hljs-params">()</span>.<span class="hljs-params">await</span>?</span><br/><span class="hljs-function"> {</span><br/><span class="hljs-function"> <span class="hljs-params">match</span> <span class="hljs-params">row</span></span><br/><span class="hljs-function"> {</span><br/><span class="hljs-function"> <span class="hljs-params">QueryItem</span>::<span class="hljs-params">Metadata</span>(<span class="hljs-params">meta</span>) =&gt;</span><br/> {<br/> println!<span class="hljs-function">(<span class="hljs-params"><span class="hljs-string">"{:?}"</span>, meta</span>);</span><br/><span class="hljs-function"> <span class="hljs-params">println</span>!(<span class="hljs-params"><span class="hljs-string">""</span></span>);</span><br/><span class="hljs-function"> },</span><br/><span class="hljs-function"> <span class="hljs-params">QueryItem</span>::<span class="hljs-params">Row</span>(<span class="hljs-params">row</span>) =&gt;</span><br/> {<br/> println!(<span class="hljs-string">"{:?}"</span>, row);<br/> println!(<span class="hljs-string">""</span>);<br/> }<br/> }<br/> }<br/> Ok(())<br/>}
Die Funktion execute_query() wurde mit dem Schlüsselwort async definiert, wodurch Sie dem Rust-Compiler mitteilen, dass diese Funktion asynchron über Cooperative Scheduling ausgeführt werden soll. Jedes Mal, wenn Sie eine asynchrone Funktion aufrufen, die zu einer Blockade im Betriebssystem führt, wechselt die asynchrone Runtime über Cooperative Scheduling zu einem anderen Code-Pfad (der zuvor durch einen Funktionsaufruf blockiert war) und führt diesen weiter aus. Dabei handelt es sich um die zuvor angesprochene Technik des CPU-Yieldings. Wichtig zu wissen ist, dass hier das Betriebssystem nicht involviert ist, da das Cooperative Scheduling vollständig im User-Mode erfolgt. Die Frage, die sich nun stellt, ist, an welchen Punkten das CPU-Yielding durchgeführt werden soll. Dazu definiert die Programmiersprache Rust das Schlüsselwort await.Dieses Schlüsselwort können Sie beim Aufruf einer asynchronen Funktion angeben, wodurch seitens der verwendeten asynchronen Runtime das CPU-Yielding durchgeführt wird, wenn der Funktionsaufruf blockiert.In Listing 3 werden die folgenden CPU-Yielding-Points über das Schlüsselwort await definiert:
- TcpStream::connect(…).await?
- Client::connect(…).await?
- select_query.query(…).await?
- data_stream.try_next().await?
Futures
Jetzt geht es darum, wie man eine asynchrone Funktion in Rust aufrufen und verwenden kann. Der hier zuerst abgedruckte, logischste Ansatz – nämlich der direkte Aufruf der asynchronen Funktion – führt leider noch nicht zum gewünschten Ergebnis:
<span class="hljs-selector-tag">fn</span> <span class="hljs-selector-tag">main</span>()
{
<span class="hljs-selector-tag">execute_query</span>(
<span class="hljs-string">"SELECT @@VERSION"</span>.to_string(), <span class="hljs-number">0</span>);
<span class="hljs-selector-tag">println</span>!(<span class="hljs-string">"Done!"</span>);
}
Rufen Sie die asynchrone Funktion über einen normalen Funktionsaufruf, passiert einfach gar nichts, wie Sie in Bild 2 sehen können. Lediglich das Makro println! der Funktion main() wird ausgeführt, der Code innerhalb der Funktion execute_query() allerdings nicht. Im Bild hervorgehoben ist, dass der Rust-Compiler eine Warnung generiert hat, die besagt, dass eine sogenannte Future nichts durchführt, solange das Schlüsselwort await fehlt oder die Funktion poll() ausgeführt wird. Was genau ist nun aber eine Future, und was bedeutet diese Warnung?
Futures tun zunächst einmal nichts (Bild 2)
Autor
Eine Future in Rust ist nichts anderes als eine Datenstruktur, welche die Trait Future [3] implementiert, die zur Standard Library von Rust gehört. Die Definition dieser Trait ist einfach gehalten, da nur ein Output-Typ und die Funktion poll() definiert wird. Hier die Definition dieser Trait:
<span class="hljs-title">pub</span> trait <span class="hljs-type">Future</span>
{
<span class="hljs-keyword">type</span> <span class="hljs-type">Output</span>;
// <span class="hljs-type">Required</span> method
fn poll(
self: <span class="hljs-type">Pin</span><&mut <span class="hljs-type">Self</span>>,
cx: &mut <span class="hljs-type">Context</span><'_>) ->
<span class="hljs-type">Poll</span><<span class="hljs-type">Self</span>::<span class="hljs-type">Output</span>>;
}
Wie Sie erkennen können, hat die Funktion poll() der Trait Future eine veränderliche Referenz auf sich selbst (Parameter self) und eine veränderliche Referenz auf einen sogenannten Context (Parameter cx).Interessant ist hierbei der Parameter self, da dieser in eine Datenstruktur vom Typ Pin<T> eingebettet ist. In Rust müssen Futures immer gepinnt werden, das heißt, dass diese im Hauptspeicher nicht verschoben werden dürfen. Diese Vorgehensweise ist notwendig, da Futures sogenannte Self-Referential Structures [6] sind, also Verweise auf sich selbst beinhalten.Würde der Speicherort einer Future im Hauptspeicher verändert, würden diese Verweise ungültig. Durch Verwenden des Datentyps Pin<T> ist sichergestellt, dass der Speicherort einer Future nicht geändert werden kann, und somit bleiben auch die Verweise innerhalb der Future intakt.Rufen Sie nun eine asynchrone Funktion auf, generiert der Rust-Compiler eine Datenstruktur, die diese Trait implementiert, und liefert sie als Funktionsergebnis zurück. Innerhalb der Funktion poll() ist die Funktion enthalten, die Sie zuvor in der asynchronen Funktion implementiert haben.Der Rust-Compiler führt eine Code-Transformation der asynchronen Funktion in eine Future durch. Darum wird beim Funktionsaufruf auch die eigentliche Funktionsimplementierung nicht ausgeführt, da nur die Future-Datenstruktur generiert wird.Damit nun die asynchrone Funktion ausgeführt werden kann, müssen Sie die zurückgelieferte Future-Datenstruktur an einen sogenannten Executor übergeben, der in regelmäßigen Intervallen die Funktion poll() aufruft und so die Future Schritt für Schritt abarbeitet.Die Executor-Komponente ist Bestandteil der asynchronen Runtime, die in der Rust-Anwendung referenziert wurde – im Beispiel ist sie also Bestandteil der Runtime async-std.Listing 4 zeigt nun, wie Sie die Future über die Funktion task::block_on() der asynchronen Runtime ausführen. Das Ergebnis dieser Ausführung zeigt Bild 3.
Listing 4: Eine Future ausführen
fn execute_single_query()<br/>{<br/> <span class="hljs-regexp">// The current thread will be blocked until</span><br/><span class="hljs-regexp"> //</span> the future <span class="hljs-keyword">is</span> resolved.<br/> <span class="hljs-keyword">let</span> response1 = task::block_on(<br/> execute_query(<br/> <span class="hljs-string">"SELECT TOP 10 </span><br/><span class="hljs-string"> BusinessEntityID, FirstName, LastName, </span><br/><span class="hljs-string"> ModifiedDate FROM </span><br/><span class="hljs-string"> Person.Person"</span>.to_string(), <span class="hljs-number">0</span>));<br/> println!(<span class="hljs-string">"{:?}"</span>, response1);<br/> <span class="hljs-regexp">// The current thread will be blocked until</span><br/><span class="hljs-regexp"> //</span> the future <span class="hljs-keyword">is</span> resolved<br/> <span class="hljs-keyword">let</span> response2 = task::block_on(<br/> execute_query(<br/> <span class="hljs-string">"SELECT @@VERSION"</span>.to_string(), <span class="hljs-number">0</span>));<br/> println!(<span class="hljs-string">"{:?}"</span>, response2);<br/>} 
Die asynchrone Funktion wurde erfolgreich ausgeführt (Bild 3)
Autor
Wie Sie sehen, wird nun das Ergebnis der SQL-Abfrage
auf der Konsole ausgegeben – die asynchrone Funktion, beziehungsweise die Future, in die die asynchrone Funktion transformiert wurde, ist korrekt ausgeführt worden.Ein Problem, das sich in Listing 4 versteckt hat, ist die Tatsache, dass die beiden SQL-Statements seriell, also der Reihe nach ausgeführt werden. Die Funktion task::block_on() blockiert nämlich den aktuellen Thread so lange, bis die übergebene Future vollständig abgearbeitet wurde.Daraus folgt, dass das zweite SQL-Statement erst dann ausgeführt werden kann, wenn das erste Statement beendet wurde. Daher wird später noch besprochen, wie Sie mehrere Futures parallel abarbeiten können.
auf der Konsole ausgegeben – die asynchrone Funktion, beziehungsweise die Future, in die die asynchrone Funktion transformiert wurde, ist korrekt ausgeführt worden.Ein Problem, das sich in Listing 4 versteckt hat, ist die Tatsache, dass die beiden SQL-Statements seriell, also der Reihe nach ausgeführt werden. Die Funktion task::block_on() blockiert nämlich den aktuellen Thread so lange, bis die übergebene Future vollständig abgearbeitet wurde.Daraus folgt, dass das zweite SQL-Statement erst dann ausgeführt werden kann, wenn das erste Statement beendet wurde. Daher wird später noch besprochen, wie Sie mehrere Futures parallel abarbeiten können.
Futures als State Machines
Im vorangegangenen Abschnitt haben Sie gelernt, dass der Rust-Compiler aus einer asynchronen Funktion eine Future erzeugt, welche die Trait Future aus der Standard-Library implementiert. Und zum Ausführen einer Future benötigen Sie einen Executor, der Bestandteil der gewählten asynchronen Runtime ist.Was genau ist nun aber eine Future, und wie kann ein Executor eine solche Future ausführen? Die zurückgelieferte Datenstruktur, welche die Trait Future implementiert, ist nichts anderes als eine State Machine, die aus mehreren unterschiedlichen Status besteht.Wie Sie zuvor in Listing 3 gesehen haben, wurde beim Aufruf einer asynchronen Funktion das Schlüsselwort await verwendet. Bei jeder Verwendung dieses Schlüsselworts handelt es sich einerseits um einen CPU-Yielding-Point und anderseits um einen Status innerhalb der State Machine, der in der zurückgelieferten Future durch den Rust-Compiler generiert wird. Die folgenden Code-Zeilen zeigen exemplarisch, welche Stati für die asynchrone Funktion aus Listing 3 durch den Rust Compiler generiert werden.
<span class="hljs-class"><span class="hljs-keyword">enum</span> <span class="hljs-title">State</span></span>
<span class="hljs-class">{</span>
<span class="hljs-class"> <span class="hljs-title">Unpolled</span>,</span>
<span class="hljs-class"> <span class="hljs-title">State0</span>, // <span class="hljs-title">TcpStream::connect</span>()</span>
<span class="hljs-class"> <span class="hljs-title">State1</span>, // <span class="hljs-title">Client::connect</span>()</span>
<span class="hljs-class"> <span class="hljs-title">State2</span>, // <span class="hljs-title">select_query</span>.<span class="hljs-title">query</span>()</span>
<span class="hljs-class"> <span class="hljs-title">State3</span> // <span class="hljs-title">data_stream</span>.<span class="hljs-title">try_next</span>()</span>
<span class="hljs-class"> <span class="hljs-title">Resolved</span></span>
<span class="hljs-class">}</span>
Wie Sie erkennen können, startet eine Future immer im Status Unpolled. Bei jeder Verwendung des Schlüsselworts await wird ein weiterer Status zur State Machine hinzugefügt, und wenn die asynchrone Funktion vollständig abgearbeitet wurde endet jede State Machine im Status Resolved.Um nun von einem Status in den nächsten Status wechseln zu können, muss die Executor-Komponente auf der Future-Datenstruktur die Funktion poll() aufrufen (die Bestandteil der Trait Future ist). Diese führt dann eine entsprechende State Transition durch und wechselt vom aktuellen in den nächsten Status. Dies geschieht immer dann, wenn in der Code-Ausführung ein Punkt erreicht wird, an dem das Schlüsselwort await verwendet wurde.Da diese Code-Punkte jedoch einen asynchronen Funktionsaufruf beinhalten, wird mit höchster Wahrscheinlichkeit der Funktionsaufruf noch nicht vollständig abgearbeitet sein, wodurch der generierte Code ein CPU-Yielding durchführt, wodurch auf dem gleichen Thread ein anderer Code-Pfad zur Ausführung kommt.Sobald der ursprüngliche asynchrone Funktionsaufruf seine Arbeit erledigt hat, wird dies der asynchronen Runtime über eine sogenannte Waker-Komponente mitgeteilt, wodurch bei einer der nächsten CPU-Yielding-Operationen der ursprüngliche Task weiter ausgeführt wird – und zwar genau an der Stelle, wo er zuvor angehalten wurde.Das ist die Grundidee des Cooperative Scheduling, da hier seitens des Betriebssystems kein Kontext-Switching erforderlich ist.Sämtliche Scheduling-Operationen finden direkt im User-Mode statt, ohne das hier Funktionalitäten aufgerufen werden müssen, die im Kernel-Mode des Betriebssystems ablaufen. Dadurch ist das Konzept des Cooperative Scheduling für blockierende Funktionsaufrufe um einiges besser geeignet als das Konzept des Preemptive Scheduling. Bild 4 zeigt einen Ausschnitt aus dem Code, den der Rust-Compiler für asynchrone Funktionen generiert. Diesen sogenannten HIR Code (High Level Intermediate Representation Code) können Sie sich über folgenden Kommandozeilenbefehl ausgeben lassen:
Ausschnitt des Codes, den der Rust-Compiler für asynchrone Funktionen generiert (Bild 4)
Autor
cargo rustc <span class="hljs-comment">-- -Z unpretty=hir</span>
Wichtig ist dabei, dass Sie eine Nightly-Build-Version von Rust einsetzen, da ein normaler Release-Build diesen Befehl nicht unterstützt. Installieren lässt sich eine solche Version mithilfe des folgenden Kommandozeilenbefehls:
<span class="hljs-title">rustup</span> <span class="hljs-keyword">default</span> nightly
Möchten Sie zur normalen Release-Build zurückwechseln, verwenden Sie den folgenden Kommandozeilenbefehl:
<span class="hljs-title">rustup</span> <span class="hljs-keyword">default</span> stable Parallel ausgeführte Futures
In Listing 4 haben Sie gesehen, wie Sie asynchrone Funktionen über die Funktion task::block_on() aufrufen. Übergeben Sie dieser Funktion jedoch nur eine Future zur Ausführung, wird der aktuelle Thread so lange blockiert, bis diese Future ihre Code-Ausführung beendet hat. Damit erreichen Sie aber keine richtige Parallelisierung, da bei einem CPU-Yielding-Point keine anderen Code-Pfade zur Ausführung zur Verfügung stehen.Daher bietet Ihnen die asynchrone Runtime async-std die Funktion task::spawn() an, mit der Sie eine asynchrone Funktion asynchron im Hintergrund ausführen können, ohne dass der aktuelle Thread blockiert wird. Durch diese Vorgehensweise erreichen Sie, dass Futures parallel ausgeführt werden. Listing 5 zeigt ein einfaches Beispiel.Listing 5: Mehrere Futures ausführen
fn execute_multiple_queries_concurrently()<br/>{<br/> <span class="hljs-built_in">let</span> mut queries = vec![];<br/> queries.<span class="hljs-built_in">push</span>(<br/> <span class="hljs-string">"SELECT TOP 10 BusinessEntityID, </span><br/><span class="hljs-string"> FirstName, LastName, ModifiedDate FROM </span><br/><span class="hljs-string"> Person.Person"</span>.to_string());<br/> queries.<span class="hljs-built_in">push</span>(<span class="hljs-string">"SELECT @@VERSION"</span>.to_string());<br/> queries.<span class="hljs-built_in">push</span>(<span class="hljs-string">"SELECT COUNT(*) FROM </span><br/><span class="hljs-string"> Person.Person"</span>.to_string());<br/> queries.<span class="hljs-built_in">push</span>(<span class="hljs-string">"SELECT TOP 10 AddressID, </span><br/><span class="hljs-string"> AddressLine1, PostalCode FROM </span><br/><span class="hljs-string"> Person.Address"</span>.to_string());<br/> // The current thread will be blocked until<br/> // the top future <span class="hljs-built_in">is</span> resolved.<br/> <span class="hljs-built_in">let</span> response = task::block_on(<br/> many_select_queries(queries));<br/> println!(<span class="hljs-string">"{:?}"</span>, response);<br/>}<br/>async fn many_select_queries(queries: Vec&lt;String&gt;)<br/>{<br/> <span class="hljs-built_in">let</span> mut duration: u64 = queries.len() as u64 + <span class="hljs-number">1</span>;<br/> <span class="hljs-built_in">let</span> mut handles = vec![];<br/> <span class="hljs-keyword">for</span> query <span class="hljs-keyword">in</span> queries<br/> {<br/> // Runs the <span class="hljs-built_in">new</span> task asynchronously<br/> // <span class="hljs-keyword">in</span> the <span class="hljs-built_in">background</span>.<br/> // Other code can run concurrently...<br/> handles.<span class="hljs-built_in">push</span>(task::spawn(<br/> execute_query(query.to_string(), duration)));<br/> };<br/> // Wait until all futures are resolved.<br/> <span class="hljs-keyword">for</span> handle <span class="hljs-keyword">in</span> handles<br/> {<br/> <span class="hljs-built_in">let</span> <span class="hljs-symbol">_</span> = handle.await;<br/> }<br/>}
Im Code werden innerhalb der Funktion execute_multiple_queries_concurrently() die verschiedenen Abfragen zusammengefasst, die anschließend über die Funktion many_select_queries() asynchron aufgerufen werden. Hierbei wird der Thread über die Funktion task::block_on() so lange blockiert, bis alle Abfragen beendet wurden. Daraus folgt, dass es sich bei der Funktion many_select_queries() wiederum um eine asynchrone Funktion handelt, die als Ergebnis eine Future zurückliefert.Innerhalb der Implementierung dieser Funktion wird nun für jede übergebene Abfrage die Funktion task::spawn() aufgerufen, die als Ergebnis eine Variable vom Typ JoinHandle zurückliefert. Auf dieser Variablen wird anschließend in einer Schleife wiederum die Funktion await aufgerufen, welche die internen State Machines der Futures von einem Status in den nächsten Status bewegt und somit die asynchronen Funktionen Schritt für Schritt abarbeitet. Bild 5 zeigt die Ergebnisse der verschiedenen Abfragen. Wenn Sie genau hinsehen, erkennen Sie, dass die Datensätze der einzelnen Abfragen auf der Konsole in einer anderen Reihenfolge ausgegeben werden, da sie nun parallelisiert über das Cooperative Scheduling ausgeführt wurden.
Ergebnisse der parallelen SQL-Abfragen (Bild 5)
Autor
Synchroner Code in Futures
Eine weitere Besonderheit von asynchronen Funktionen in Rust ist, wie Sie in diesen Funktionen synchrone Funktionsaufrufe durchführen sollten. Wenn Sie eine synchrone Funktion ganz normal aufrufen, blockiert der aktuelle Thread so lange, bis der synchrone Funktionsaufruf vollständig durchgeführt wurde.Daraus folgt aber auch, dass kein Cooperative Scheduling seitens des Executors durchgeführt werden kann, da bis zur Fertigstellung des synchronen Funktionsaufrufs kein CPU-Yielding-Point innerhalb der asynchronen Funktion erreicht werden kann. Dadurch wird auch die Executor-Komponente der asynchronen Runtime blockiert. Daraus ergeben sich dann entsprechende Performanceprobleme.Listing 6 illustriert dieses Problem innerhalb der Funktion execute_query() durch den expliziten Aufruf der synchronen Funktion std::thread::sleep(), die den aktuellen Thread für die angegebene Zeit pausiert.Listing 6: Synchrone Funktionsaufrufe 1/2
// Loop over the received data.<br/><span class="hljs-keyword">while</span> <span class="hljs-keyword">let</span> Some(row) = <br/> data_stream.try_next().await?<br/>{<br/> match row<br/> {<br/> QueryItem<span class="hljs-type">::</span>Metadata(meta) =&gt;<br/> {<br/> println!(<span class="hljs-string">"{:?}"</span>, meta);<br/> println!(<span class="hljs-string">""</span>);<br/> },<br/> QueryItem<span class="hljs-type">::</span>Row(row) =&gt;<br/> {<br/> println!(<span class="hljs-string">"{:?}"</span>, row);<br/> println!(<span class="hljs-string">""</span>);<br/> // This call blocks the executor of the <br/> // runtime, because there is no .await!<br/> println!(<span class="hljs-string">"Long running CPU intensive task..."</span>);<br/> std<span class="hljs-type">::</span>thread<span class="hljs-type">::</span>sleep(<br/> core<span class="hljs-type">::</span>time<span class="hljs-type">::</span>Duration<span class="hljs-type">::</span>from_secs(duration));<br/> }<br/> }<br/>}
Rufen Sie diesen Code mit std::thread:sleep() auf, werden Sie feststellen, dass die Ausführung um einiges länger dauert, da nach jedem zurückgelieferten Datensatz der aktuelle Thread für einige Sekunden angehalten wird – wie lange, ist abhängig vom übergebenen Parameter duration. Während dieser Zwangspause können jedoch auch kein Cooperative Scheduling durchgeführt und keine anderen Codepfade ausgeführt werden.Um dieses Problem zu vermeiden, können Sie in der Runtime async-std die Funktion task::spawn_blocking() verwenden. Diese Funktion erwartet eine Closure, die anschließend in einem eigenen Thread ausgeführt wird. Dadurch wird der Haupt-Thread nicht blockiert, und seitens des Executors kann das Cooperative Scheduling durchgeführt werden. In Listing 7 sehen Sie die dafür erforderlichen Änderungen.
Listing 7: Synchrone Funktionsaufrufe 2/2
// Loop over the received data.<br/><span class="hljs-keyword">while</span> <span class="hljs-built_in">let</span> Some(<span class="hljs-built_in">row</span>) = data_stream.try_next().await?<br/>{<br/> match <span class="hljs-built_in">row</span><br/> {<br/> QueryItem::Metadata(meta) =&gt;<br/> {<br/> println!(<span class="hljs-string">"{:?}"</span>, meta);<br/> println!(<span class="hljs-string">""</span>);<br/> },<br/> QueryItem::Row(<span class="hljs-built_in">row</span>) =&gt;<br/> {<br/> println!(<span class="hljs-string">"{:?}"</span>, <span class="hljs-built_in">row</span>);<br/> println!(<span class="hljs-string">""</span>);<br/> // The CPU intensive task <span class="hljs-built_in">is</span> running on a<br/> // separate thread, <span class="hljs-keyword">and</span> will <span class="hljs-keyword">not</span> <span class="hljs-built_in">block</span> the<br/> // executor.<br/> task::spawn_blocking(move ||<br/> {<br/> println!(<span class="hljs-string">"Long running CPU</span><br/><span class="hljs-string"> intensive task..."</span>);<br/> <span class="hljs-built_in">std</span>::thread::sleep(<br/> core::<span class="hljs-built_in">time</span>::Duration::from_secs(duration));<br/> });<br/> }<br/> }<br/>}
Der Thread, der über die Funktion task::spawn_blocking() gestartet wird, wird aus einem Thread-Pool genommen, der für blockierende Funktionsaufrufe reserviert ist. Wenn Sie – wie in diesem Beispiel – beim Aufruf der Funktion task::spawn_blocking() das Schlüsselwort await weglassen, führt der Haupt-Thread einfach seine Arbeit fort, ohne auf die neu gestarteten Threads zu warten.Daraus resultiert, dass der Haupt-Thread seine Arbeit um einiges schneller beendet hat als die nebenläufig gestarteten Threads, wodurch Ihr Programm beendet wird und die nebenläufigen Threads abgebrochen werden. Dieses Problem können Sie vermeiden, indem die nebenläufigen Threads mithilfe eines Channels mit dem Haupt-Thread kommunizieren und sich untereinander koordinieren.Zusätzlich bietet die gewählte asynchrone Runtime die Möglichkeit an, freiwillig an beliebig gewählten Code-Punkten ein CPU-Yielding durchzuführen. Dies geschieht durch den Aufruf der Funktion task::yield_now(). Dieser Ansatz ist zum Beispiel dann sehr von Vorteil, wenn Sie in einer asynchronen Funktion einen längeren CPU-intensiven Algorithmus implementieren möchten.Generell sollten Sie aufgrund der größer werdenden Komplexität vermeiden, in einer asynchronen Funktion CPU-intensive Operationen durchzuführen beziehungsweise synchrone Funktionen auszuführen. Je mehr asynchrone Funktionen mit entsprechenden CPU-Yielding-Points zur Verfügung stehen, desto besser kann der Executor im Hintergrund das Cooperative Scheduling durchführen.Eine weitere Funktionalität, die Rust in Kombination mit asynchronen Funktionen anbietet, sind sogenannte Async Blocks. Ein Async Block ist nichts anderes als ein Code-Abschnitt, der als Ergebnis eine Future zurückliefert, die wiederum über eine asynchrone Runtime ausgeführt werden kann. Hier ein einfaches Beispiel:
<span class="hljs-function"><span class="hljs-keyword">fn</span> <span class="hljs-title">async_blocks</span></span>()
{
<span class="hljs-keyword">let</span> query1 = async
{
<span class="hljs-keyword">let</span> _ = execute_query(
<span class="hljs-string">"SELECT TOP 10 BusinessEntityID, </span>
<span class="hljs-string"> FirstName, LastName, ModifiedDate FROM </span>
<span class="hljs-string"> Person.Person"</span>.to_string(), <span class="hljs-number">0</span>).await;
};
task::block_on(query1);
}
Ein Async Block bietet ebenfalls die Möglichkeit, CPU-Yielding-Points über das Schlüsselwort await zu definieren, an denen ein Cooperative Scheduling durch die asynchrone Runtime durchgeführt wird.
Fazit
In diesem Artikel haben Sie das asynchrone Programmiermodell von Rust kennengelernt, mit dem Sie in einer einfachen Art und Weise ein Cooperative Scheduling in Ihren Anwendungen realisieren können. Die Standard-Library von Rust definiert dafür die Trait Future, die über eine asynchrone Runtime Schritt für Schritt abgearbeitet werden kann.Fussnoten
- [1] Klaus Aschenbrenner, Multithreading, dotnetpro 10/2024, Seite 131 ff., http://www.dotnetpro.de/A2410Rust
- [2] Promises in JavaScript, http://www.dotnetpro.de/SL2411Rust1
- [3] TAP im .NET Framework, http://www.dotnetpro.de/SL2412Rust2
- [4] Trait Future, http://www.dotnetpro.de/SL2412Rust3
- [5] async-std, https://docs.rs/async-std
- [6] Self Referential, http://www.dotnetpro.de/SL2412Rust4
- [7] Tokio, https://docs.rs/tokio
- [8] smol, https://docs.rs/smol
- [9] fuchsia-async, http://www.dotnetpro.de/SL2412Rust5
- Tiberius, http://www.dotnetpro.de/SL2412Rust6
