Synchronous concurrency: Difference between revisions

Content added Content deleted

Inline

@@ Line 2,438: / Line 2,438: @@
 =={{header|Rust}}==
-{{works with|rustc 1.4.0-nightly|f84d53ca0 2015-09-06}}
+{{works with|rustc 1.46.0 (04488afe3 2020-08-24)|}}
 <lang rust>use std::fs::File;
-use std::io::BufReader;
 use std::io::BufRead;
+use std::io::BufReader;
-use std::thread::spawn;
+use std::sync::mpsc::{channel, sync_channel};
+use std::thread;
-use std::sync::mpsc::{SyncSender, Receiver, sync_channel};
 fn main() {
+    // The reader sends lines to the writer via an async channel, so the reader is never blocked.
-    let (tx, rx): (SyncSender<String>, Receiver<String>) = sync_channel::<String>(0);
+    let (reader_send, writer_recv) = channel();
+    // The writer sends the final count via a blocking channel with bound 0,
-    // Reader thread.
+    // meaning the buffer is exactly the size of the result.
-    spawn(move || {
+    let (writer_send, reader_recv) = sync_channel(0);
-        let file = File::open("input.txt").unwrap();
+    // Define the work the reader will do.
+    let reader_work = move || {
+        let file = File::open("input.txt").expect("Failed to open input.txt");
         let reader = BufReader::new(file);
         for line in reader.lines() {
             match line {
-                Ok(msg) => tx.send(msg).unwrap(),
+                Ok(msg) => reader_send
-                Err(e) => println!("{}", e)
+                    .send(msg)
+                    .expect("Failed to send via the channel"),
+                Err(e) => println!("{}", e),
             }
         }
+        // Dropping the sender disconnects it and tells the receiver the connection is closed.
-        drop(tx);
-    });
+        drop(reader_send);
+        // Now that we've sent all the lines,
-    // Writer thread.
+        // block until the writer gives us the final count.
-    spawn(move || {
-        let mut loop_count: u16 = 0;
+        let count = reader_recv
+            .recv()
+            .expect("Failed to receive count from printer.");
-        loop {
+        println!("{}", count);
+    };
-            let recvd = rx.recv();
-            match recvd {
+    // Define the work the writer will do.
+    let writer_work = move || {
+        let mut line_count = 0;
+        loop {
+            match writer_recv.recv() {
                 Ok(msg) => {
                     println!("{}", msg);
-                    loop_count += 1;
+                    line_count += 1;
-                },
+                }
-                Err(_) => break // rx.recv() will only err when tx is closed.
+                Err(_) => break, // indicates the connection has been closed by the sender.
             }
         }
-        println!("Line count: {}", loop_count);
+        // Send the final count back to the reader.
+        writer_send
-    }).join().unwrap();
+            .send(line_count)
+            .expect("Failed to send line count from writer.");
+        drop(writer_send);
+    };
+    // Spawn each as a thread.
+    let reader_handle = thread::spawn(reader_work);
+    thread::spawn(writer_work);
+    reader_handle
+        .join()
+        .expect("Failed to join the reader thread.");
 }</lang>