Rust
Summary - Channels
Combine mpsc and thread to build a threaded work queue.
Channels summary
mpsc::channel opens an unbounded channel, sync_channel(n) a bounded one. send transfers ownership; Err if there's no receiver. recv waits, try_recv doesn't, recv_timeout waits for a duration; once all senders close, the receiver ends. tx.clone() gives many producers. Async code uses tokio::sync::mpsc. Channels are the clean way to build data flow between stages instead of shared state.
Pipeline
Each stage is a thread; stages are connected with channels. When a stage finishes, its sender is dropped and the closure propagates down the chain to the following stages:
let (tx1, rx1) = mpsc::channel();
let (tx2, rx2) = mpsc::channel();
thread::spawn(move || for i in 1..=5 { tx1.send(i).unwrap(); });
thread::spawn(move || for n in rx1 { tx2.send(n * n).unwrap(); });
let sonuc: Vec<i32> = rx2.iter().collect();
Exercise Two-stage pipeline
Build the pipeline above: let the first thread send the numbers 1 to 5, and the second thread forward each number's square to the next channel. Let the main thread collect the results and print them with {:?}.
Show example solution
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx1, rx1) = mpsc::channel();
let (tx2, rx2) = mpsc::channel();
thread::spawn(move || {
for i in 1..=5 {
tx1.send(i).unwrap();
}
});
thread::spawn(move || {
for n in rx1 {
tx2.send(n * n).unwrap();
}
});
let sonuc: Vec<i32> = rx2.iter().collect();
println!("{:?}", sonuc);
}
Test yourself
-
In a two-stage pipeline (produce → square → collect), why is the order preserved?
- It isn't preserved
- Thanks to a Mutex
- Each stage has a single thread and channels are FIFO
- Rust sorts automatically
Show answer
Each stage has a single thread and channels are FIFO — If several workers are added to a stage, the order guarantee is lost.
-
Connect the stage
thread::spawn(move || for n in ____ {
tx2.send(n * n).unwrap();
});
Show answer
thread::spawn(move || for n in rx1 {
tx2.send(n * n).unwrap();
});
-
Group summary
- channel()
- Unbounded channel
- sync_channel(n)
- Bounded channel
- tx.clone()
- Many producers
- try_recv
- Without waiting
Next lesson: TCP listener
