Rust's 'Fan-Out Collector' Pattern for Concurrency
Simulating an App Update System using the Actor / Handle pattern with Channels
Introduction
Rust's Actor / Handle Pattern comes up frequently-ish in the wild, and for a good reason. It makes Rust concurrency operations a lot more intuitive. In this post I will go over my favorite variant of the Actor / Handle Pattern, the 'Fan-Out Collector' pattern.
I like the 'Fan-Out Collector' pattern for many reasons but the most relevant ones (for the type of work I usually do) are that the Handle only spawns a new thread when necessary, while the Actor (ie, "Manager") waits until it needs to collect something. It is not sitting in a continous loop on its own dedicated thread. This is a very "directional process" which appeals to me.
So, in this post I will be applying the Fan-Out Collector pattern for a simulated "App Update System", whereby three "Apps" will be able to request and download an update, and install it concurrently. Each "App" will be downloading / updating on its own thread.
This will be fun? 🤷♀️
Preliminary Setup
First let's bring in the necessary components. All std lib here, nothing fancy:
use std::{
sync::mpsc::{
self,
Sender,
Receiver,
},
thread::{
spawn as std_spawn,
sleep
},
time::Duration
};
Next, lets define our three "Apps":
struct BrowserApp {
name: String,
version: u32,
}
impl BrowserApp {
fn new(name: &str, version: u32) -> Self {
Self { name: name.into(), version }
}
}
struct GameApp {
name: String,
version: u32,
}
impl GameApp {
fn new(name: &str, version: u32) -> Self {
Self { name: name.into(), version }
}
}
struct SocialApp {
name: String,
version: u32,
}
impl SocialApp {
fn new(name: &str, version: u32) -> Self {
Self { name: name.into(), version }
}
}
Let's also define an Updatable trait. This will allow us to implement the (simulated) "downloading & install" logic to the Apps, while also allowing us to pass any App (that implements Updatable) to the Handle.
In each implementation we simulate the "download & install" logic by adding a sleep call. Note that the "SocialApp" is being simulated to fail the update process with a "Network Timeout".
trait Updatable {
fn name(&self) -> &str;
// Simulates "downloading & install" logic:
fn download_and_install(&self) -> Result<u32, String>;
}
// - impl for `BrowserApp`:
impl Updatable for BrowserApp {
fn name(&self) -> &str {
&self.name
}
fn download_and_install(&self) -> Result<u32, String> {
// download sim:
sleep(Duration::from_millis(500));
// Succeeds:
Ok( self.version )
}
}
// - impl for `GameApp`:
impl Updatable for GameApp {
fn name(&self) -> &str {
&self.name
}
fn download_and_install(&self) -> Result<u32, String> {
// download sim:
sleep(Duration::from_secs(1));
// Succeeds:
Ok( self.version )
}
}
// - impl for `SocialApp`:
impl Updatable for SocialApp {
fn name(&self) -> &str {
&self.name
}
fn download_and_install(&self) -> Result<u32, String> {
// download sim:
sleep(Duration::from_secs(1));
// FAILS:
Err( "Network timeout".into() )
}
}
Next, let's define the UpdateStatus enum. This is the message that the Actor / Handle will pass through their channel:
enum UpdateStatus {
Success { name: String, old_version: u32, new_version: u32 },
Failed { name: String, reason: String }
}
The Actor / Handle Pattern
Oh boy, the exciting part! Let's set up the Actor and Handler objects. Don't worry, its a piece of cake.
First, we set up the UpdateHandle, the "Handle" part of the "Actor / Handle Pattern" (this is also sometimes called the Private API). It contains the Sender of the UpdateStatus.
struct UpdateHandle {
sender: Sender<UpdateStatus>
}
The main implementation on UpdateHandle will be the .submit() method which does what it sounds like. It takes ownership of an "App", spawns a thread, and sends the results back through the channel.
Once again, an "App" is anything that implements the Updatable trait (above), as well as the Send trait (required by channels) and a 'static lifetime (so that the "App" lives long enough for the thread to finish).
The only other implementation on UpdateHandle will be the .close() method which is an empty method that only serves to consume the sender therby closing (ie: dropping) the original sender.
impl UpdateHandle {
// // `.submit()`:
fn submit<A: Updatable + Send + 'static>(&self, app: A) {
// - Clone `Sender`:
let tx: Sender<UpdateStatus> = self.sender.clone();
// - Spawn thread:
std_spawn(move || {
// - Get the app's name before consuming it:
let app_name: &str = app.name();
// - Call `download_and_install()`:
let res: Result<u32, String> = app.download_and_install();
// - Match on the result:
match res {
// - Send `UpdateStatus::Success` or `UpdateStatus::Failed` via tx
Ok(old_version) => {
let new_version: u32 = old_version + 1;
let mssg: UpdateStatus = UpdateStatus::Success {
name: app_name.into(), old_version, new_version
};
let _ = tx.send(mssg); // `let _ =` to shut up the warnings
},
Err(e) => {
let mssg: UpdateStatus = UpdateStatus::Failed {
name: app_name.into(), reason: e
};
let _ = tx.send(mssg); // `let _ =` to shut up the warnings
}
}
});
}
// // `.close()`:
// - Consumes the handle, dropping the sender:
fn close(self) {
/* no code needed, just consumes `self` */
}
}
Next, we set up the UpdateManager, the "Actor" part of the "Actor / Handle Pattern" (this is also sometimes called the Public API). It contains the Receiver of the UpdateStatus.
struct UpdateManager {
receiver: Receiver<UpdateStatus>
}
The two methods for UpdateManager will be .new(), which creates a channels and returns the Actor and Hanle themselves, and .collect_results(), which collects the UpdateStatus results as a vector by looping on .recv() until all senders are dropped (ie, the channel is closed).
impl UpdateManager {
fn new() -> (Self, UpdateHandle) {
let (tx, rx) = mpsc::channel::<UpdateStatus>();
(
Self { receiver: rx },
UpdateHandle { sender: tx }
)
}
fn collect_results(self) -> Vec<UpdateStatus> {
let mut ret: Vec<UpdateStatus> = vec![];
while let Ok(update_status) = self.receiver.recv() {
ret.push(update_status);
}
ret
}
}
Running in 'main()'
Alright, with all that setup out of the way let's bring everything together. All of the code below will be run in main().
Let's start with initializing the channel and returning the Actor and Handle objects:
let (actor, handle) = UpdateManager::new();
Next, lets "update" our three Apps. We do this by passing each app to the Handle's .submit() method. Recall that .submit() spawns a new thread for each App to work on independently. The process is completed by calling .close() afterwards to close the channel (by dropping the original sender):
handle.submit(BrowserApp::new("Firefox", 120));
handle.submit(GameApp::new("CyberQuest", 2));
handle.submit(SocialApp::new("Chirp", 9));
handle.close();
We can now collect the results using the Actor's aptly-named .collect_results() method:
let results: Vec<UpdateStatus> = actor.collect_results();
Printing our results shows the expected outcome, two of the Apps updated successfully, while the SocialApp ("Chirp") failed with a timeout error:
for r in &results {
match r {
UpdateStatus::Success { name, old_version, new_version } => {
println!("OK {} updated: v{} -> v{}", name, old_version, new_version);
}
UpdateStatus::Failed { name, reason } => {
println!("ERR {} failed: {}", name, reason);
}
}
}
// OK Firefox updated: v120 -> v121
// OK CyberQuest updated: v2 -> v3
// ERR Chirp failed: Network timeout
This entire process takes ~1 second to execute (the longest sleep of one App), even though the total sleep time for all three apps is 2.5 seconds. This is obviously the concurrency gain of running each App in its own thread. Neat!
Denouement
And thats it. A very useful channel pattern that is very easy to implement. That's all I have to say about it. Try it out!
As always, thanks for reading!