#[derive(Debug, Clone)]
pub struct Order {
pub id: u64,
pub amount_cents: i64,
pub currency: [u8; 3],
}
pub fn shard_indices(len: usize, shards: usize) -> Vec<(usize, usize)> {
if len == 0 || shards == 0 {
return Vec::new();
}
let shards = shards.min(len);
let base = len / shards;
let rem = len % shards;
let mut ranges = Vec::with_capacity(shards);
let mut start = 0usize;
for i in 0..shards {
let extra = if i < rem { 1 } else { 0 };
let end = start + base + extra;
ranges.push((start, end));
start = end;
}
ranges
}
pub fn into_shards<'a>(orders: &'a [Order], shards: usize) -> Vec<&'a [Order]> {
shard_indices(orders.len(), shards)
.into_iter()
.map(|(start, end)| &orders[start..end])
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn empty_slice_has_no_shards() {
assert!(shard_indices(0, 4).is_empty());
}
#[test]
fn uneven_split_is_balanced() {
let ranges = shard_indices(10, 3);
let sizes: Vec<usize> = ranges.iter().map(|(s, e)| e - s).collect();
assert_eq!(sizes, vec![4, 3, 3]);
assert_eq!(ranges.last().unwrap().1, 10);
}
}
use rayon::prelude::*;
use crate::sharding::{into_shards, Order};
#[derive(Debug, Clone, Default)]
pub struct ShardSummary {
pub count: u64,
pub total_cents: i64,
pub max_cents: i64,
}
impl ShardSummary {
fn merge(mut self, other: ShardSummary) -> ShardSummary {
self.count += other.count;
self.total_cents += other.total_cents;
self.max_cents = self.max_cents.max(other.max_cents);
self
}
}
fn summarize_shard(shard: &[Order]) -> ShardSummary {
let mut summary = ShardSummary::default();
for order in shard {
summary.count += 1;
summary.total_cents += order.amount_cents;
summary.max_cents = summary.max_cents.max(order.amount_cents);
}
summary
}
pub fn process_orders(orders: &[Order], shards: usize) -> ShardSummary {
let partitions = into_shards(orders, shards);
partitions
.par_iter()
.map(|shard| summarize_shard(shard))
.reduce(ShardSummary::default, ShardSummary::merge)
}
pub fn recommended_shards() -> usize {
rayon::current_num_threads().max(1)
}
This snippet shows how a batch of orders is split into balanced shards and processed across worker threads, a common pattern when a single large collection must be crunched in parallel without oversaturating the thread pool. The sharding module in the first tab implements the core split logic, while the pipeline module in the second tab consumes those shards with Rayon and aggregates the per-shard results.
In sharding.rs, shard_indices computes contiguous slice boundaries for a given length and shard count using integer division and a remainder. The remainder is spread across the first rem shards so that shard sizes differ by at most one element; this beats naive len / n chunking, which leaves an awkward oversized or undersized final chunk and skews load. The function guards against shards == 0 and against shards exceeding len, returning at most one shard per element so no empty shards are produced. into_shards then borrows the input slice and yields &[Order] subslices rather than cloning, keeping the partition zero-copy.
Contiguous ranges are chosen deliberately: they preserve locality, work with any Copy-free payload, and let each shard be handed to a thread as a plain borrow. The trade-off is that if per-order cost varies wildly, equal-sized shards can still be unbalanced in wall-clock terms; for that case work-stealing or a smaller shard size is preferable, which is exactly what Rayon's dynamic scheduling helps with.
In pipeline.rs, process_orders calls into_shards and drives the parallel pass. Using par_iter over the shard vector, each shard is reduced independently by summarize_shard, and reduce folds the ShardSummary values into one total with merge. Because each shard is a disjoint borrow, there is no shared mutable state and therefore no locking on the hot path. ShardSummary::merge is associative, which is what makes the parallel reduce correct regardless of how Rayon splits the work.
The #[cfg(test)] block pins the boundary behavior: an empty slice yields nothing, and a length that does not divide evenly still produces near-equal shards. Choosing shard count near the CPU count, then letting Rayon steal, is the usual sweet spot for throughput.
Related snips
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
use clap::Parser;
#[derive(Parser, Debug)]
#[command(author, version, about)]
struct Args {
#[arg(short, long)]clap for CLI argument parsing with derive macros
Share this code
Here's the card — post it anywhere.
