rust 53 lines · 7 steps

Deduplicating concurrent work in Rust

A single-flight primitive that collapses concurrent calls for the same key into one computation and shares the result.

Explained by highlit
Advanced concurrency deduplication async broadcast-channel shared-state
1use std::collections::HashMap;
2use std::future::Future;
3use std::hash::Hash;
4use std::sync::{Arc, Mutex};
5 
6use tokio::sync::broadcast;
7 
8pub struct SingleFlight<K, V> {
9 inflight: Mutex<HashMap<K, broadcast::Sender<V>>>,
10}
11 
12impl<K, V> SingleFlight<K, V>
13where
14 K: Eq + Hash + Clone,
15 V: Clone,
16{
17 pub fn new() -> Self {
18 Self { inflight: Mutex::new(HashMap::new()) }
19 }
20 
21 pub async fn run<F, Fut>(self: &Arc<Self>, key: K, compute: F) -> V
22 where
23 F: FnOnce() -> Fut,
24 Fut: Future<Output = V>,
25 {
26 let mut receiver = {
27 let mut inflight = self.inflight.lock().unwrap();
28 match inflight.get(&key) {
29 Some(tx) => Some(tx.subscribe()),
30 None => {
31 let (tx, _) = broadcast::channel(1);
32 inflight.insert(key.clone(), tx);
33 None
34 }
35 }
36 };
37 
38 if let Some(rx) = receiver.as_mut() {
39 if let Ok(value) = rx.recv().await {
40 return value;
41 }
42 return Box::pin(self.run(key, compute)).await;
43 }
44 
45 let value = compute().await;
46 
47 let tx = self.inflight.lock().unwrap().remove(&key);
48 if let Some(tx) = tx {
49 let _ = tx.send(value.clone());
50 }
51 value
52 }
53}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Sharing one broadcast sender per key lets many waiters receive a single computed value.
  2. 2Holding a mutex only around the map lookup keeps the lock off the await points.
  3. 3Recovering from a dropped sender by retrying prevents callers from hanging when a leader fails.

Related explainers

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
rust 
use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{self, BufRead, BufReader, BufWriter, Lines, Write};

K-way merge of sorted logs in Rust

binary-heap k-way-merge streaming-io
Intermediate 8 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
javascript 
import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks
Advanced 10 steps

Share this explainer

Here's the card — post it anywhere.

Deduplicating concurrent work in Rust — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code