---
<a id="en"></a>
# PickFast : Lock-Free Weighted Load Balancer for Low-Latency Selection
High-performance weighted random selection library with atomic EMA weight updates. Designed for load balancing, A/B testing, resource scheduling, and scenarios requiring probability-based selection with dynamic weight adjustment.
## Navigation
- [Features](#features)
- [Usage Demonstration](#usage-demonstration)
- [Integration with Race Crate](#integration-with-race-crate)
- [Custom Rank Strategy](#custom-rank-strategy)
- [Design Rationale](#design-rationale)
- [API Reference](#api-reference)
- [Tech Stack](#tech-stack)
- [Directory Structure](#directory-structure)
- [Historical Anecdote](#historical-anecdote)
## Features
- **Lock-Free Updates**: Uses `AtomicU32` and Compare-And-Swap (CAS) for thread-safe weight updates without locks
- **Adaptive Weighting**: Implements Exponential Moving Average (EMA) to smooth latency fluctuations
- **Failure Handling**: Provides `failed()` method to quickly penalize underperforming nodes
- **Weight Floor Protection**: Ensures minimum weight of 1 to prevent nodes from being completely excluded
- **Cache Friendly**: Struct alignment prevents false sharing in multi-core environments
- **Flexible Strategies**: Supports custom ranking strategies via the `Rank` trait
- **Zero Allocation**: All operations are allocation-free after initialization
## Usage Demonstration
DNS server load balancing scenario demonstrating automatic preference for low-latency nodes:
```rust
use std::{net::IpAddr, sync::Arc, thread};
use pick_fast::PickFast;
#[derive(Debug, Clone, Copy)]
struct DnsServer {
ip: IpAddr,
}
// Initialize DNS servers with different latency characteristics
let servers = [
DnsServer { ip: "8.8.8.8".parse().unwrap() }, // Google DNS - 100ms
DnsServer { ip: "1.1.1.1".parse().unwrap() }, // Cloudflare - 80ms
DnsServer { ip: "223.5.5.5".parse().unwrap() }, // AliDNS - 5ms (fastest)
DnsServer { ip: "208.67.222.222".parse().unwrap() }, // OpenDNS - 60ms
];
// Create load balancer with default Inverse strategy
let lb = Arc::new(PickFast::new(servers));
// Simulate concurrent usage across multiple threads
let handles: Vec<_> = (0..4)
let latency_us = match node.ip.to_string().as_str() {
"8.8.8.8" => 100_000, // 100ms
"1.1.1.1" => 80_000, // 80ms
"223.5.5.5" => 5_000, // 5ms (fastest)
"208.67.222.222" => 60_000, // 60ms
_ => 100_000,
};
// Update node performance with observed latency
c_lb.set(node.index, latency_us);
// Optionally mark node as failed if latency is too high
if latency_us > 200_000 { // > 200ms
c_lb.failed(node.index);
}
}
})
})
.collect();
// Wait for all threads to complete
for handle in handles {
handle.join().unwrap();
}
// Check final weights - faster nodes should have higher weights
let fast_weight = lb.li[2].weight.load(std::sync::atomic::Ordering::Relaxed);
let slow_weight = lb.li[0].weight.load(std::sync::atomic::Ordering::Relaxed);
println!("Fast node weight: {fast_weight}, Slow node weight: {slow_weight}");
println!("Ratio: {:.2}", fast_weight as f64 / slow_weight as f64);
```
## DNS Load Balancing with race Integration
```rust
use pick_fast::PickFast;
use race::race;
use std::{net::IpAddr, sync::Arc, time::Duration};
use tokio::net::lookup_host;
#[derive(Debug, Clone, Copy)]
struct DnsServer {
ip: IpAddr,
}
// Real DNS server IP addresses
const DNS_SERVERS: [&str; 4] = [
"8.8.8.8:53", // Google DNS
"1.1.1.1:53", // Cloudflare DNS
"9.9.9.9:53", // Quad9 DNS
"208.67.222.222:53", // OpenDNS
];
// Create server nodes
let servers: Vec<DnsServer> = DNS_SERVERS
.iter()
})
.collect();
let lb = Arc::new(PickFast::<DnsServer, pick_fast::Inverse>::new(servers));
// DNS task with performance tracking
#[derive(Debug, Clone)]
struct DnsTask {
host: &'static str,
index: usize,
start: std::time::Instant,
}
// Use iter() to get weighted random start sequence
let iter = lb.iter();
let selected_servers: Vec<_> = iter.take(4).collect(); // Try top 4 servers
// DNS resolution with automatic weight updates
let lb_clone = lb.clone();
async move {
// Simulate DNS query latency
let start = std::time::Instant::now();
match lookup_host("example.com:80").await {
Ok(mut addrs) => {
if let Some(addr) = addrs.next() {
let duration = start.elapsed();
let latency_us = duration.as_micros() as u32;
// Success: update with actual latency
lb.set(task.index, latency_us);
println!(" ✅ {} resolved in {duration:?} (latency: {latency_us}μs)", task.host);
Ok(addr.ip())
} else {
// No address found: penalize server
lb.failed(task.index);
println!(" ❌ {} failed", task.host);
Err(std::io::Error::new(std::io::ErrorKind::NotFound, "No address found"))
}
}
Err(e) => {
// Network error: penalize server
lb.failed(task.index);
println!(" ❌ {} failed: {e}", task.host);
Err(e)
}
}
}
};
// Create DNS task iterator for race
DnsTask {
host: DNS_SERVERS[index],
index,
start: std::time::Instant::now(),
}
});
// Execute staggered DNS resolution
let rx = race(resolve_with_feedback, dns_tasks_iter, Duration::from_millis(500));
// Wait for first successful resolution
while let Ok(result) = rx.recv().await {
match result {
Ok(ip) => {
println!("🎯 First successful resolution: {ip}");
drop(rx); // Stop other tasks
break;
}
Err(_) => continue, // Try next server
}
}
```
## Custom Rank Strategy
The `Rank` trait defines how observed values (e.g., latency) convert to selection weights.
```rust
use pick_fast::Rank;
/// Priority-based ranking: higher priority = higher weight
pub struct Priority;
impl Rank for Priority {
fn calc(priority: u32) -> u32 {
priority // Direct mapping: priority value becomes weight
}
}
// Usage
let lb = PickFast::<Task, Priority>::new(tasks);
lb.set(index, 100); // Set priority to 100
```
Built-in `Inverse` strategy suits latency-based selection:
```text
Weight = (1 << 22) / max(Latency, 1)
Initial Weight = (1 << 22) / max(Node Count, 1)
```
Design choice: 2²² base ensures 256 nodes at 1μs latency won't overflow `u32`. Initial weight is inversely proportional to node count for reasonable initial distribution.
## Design Rationale
Architecture focuses on minimizing synchronization overhead and maximizing throughput.
### Call Flow
```mermaid
graph TD
A[Client Request] --> B["pick()"]
B --> C[Atomic Load Total Weight]
C --> D[Random Target Selection]
D --> E["Linear Scan Nodes O(N)"]
E --> F[Return Handle]
G[Performance Feedback] --> H["set()"]
H --> I[Calculate Target Weight via Rank]
I --> J[CAS Update Node Weight with EMA]
J --> K[Atomic Update Total Weight]
```
### EMA Smoothing
```text
New Weight = (Old Weight + Target Weight) / 2
```
Smoothing prevents drastic weight changes from transient spikes.
## API Reference
### Core Types
| `PickFast<T, M>` | Main load balancer struct. `T`: node data, `M`: rank model |
| `PickFast.li` | `Vec<Node<T>>` - Node list |
| `PickFast.total` | `AtomicU32` - Cached total weight |
| `Node<T>` | Node struct containing data and weight |
| `Node.data` | `T` - Node data |
| `Node.weight` | `AtomicU32` - Node weight |
### Key Methods
| `new(data: impl IntoIterator<Item = T>) -> Self` | Create instance from iterator |
| `len(&self) -> usize` | Get node count |
| `is_empty(&self) -> bool` | Check if empty |
| `pick(&self) -> Handle<'_, T>` | Select node based on current weights. O(1) weight load + O(N) scan |
| `set(&self, index: usize, val: u32)` | Update node observation with EMA smoothing (minimum weight: 1) |
| `failed(&self, index: usize)` | Mark node as failed, halving its weight (minimum weight: 1) |
| `iter(&self) -> CIter<'_, Node<T>>` | Create circular iterator with weighted random start position (requires `iter` feature) |
### Other Types
| `Handle<'a, T>` | Smart pointer to selected node, contains `index` and `node` reference |
| `Rank` | Trait for custom weight calculation logic |
| `Inverse` | Default strategy: weight inversely proportional to latency |
## Tech Stack
| Language | Rust (Edition 2024) |
| Randomness | `fastrand` |
| Concurrency | `std::sync::atomic` |
| Testing/Visualization | `plotters`, `svg` |
## Directory Structure
```text
.
├── Cargo.toml # Project configuration
├── src/
│ └── lib.rs # Core implementation
├── tests/
│ └── main.rs # Integration tests and chart generation
└── readme/
├── en.md # English documentation
├── zh.md # Chinese documentation
├── rank-en.svg # English performance chart
└── rank-zh.svg # Chinese performance chart
```
## Historical Anecdote
Weighted load balancing has roots in network packet scheduling. The "Weighted Round Robin" (WRR) concept was formalized in 1991 for ATM (Asynchronous Transfer Mode) networks, where heterogeneous link speeds required differential treatment.
The evolution from WRR to modern weighted random selection represents a paradigm shift: instead of deterministic slot allocation, probabilistic approaches like `pick_fast` offer natural load distribution. Combined with EMA smoothing—a technique borrowed from stock market technical analysis dating back to the 1960s—the algorithm adapts gracefully to varying network conditions.
Interestingly, the Compare-And-Swap primitive used here traces back to IBM System/370 in 1970, making lock-free programming concepts over 50 years old—yet they remain the cornerstone of modern high-performance concurrent systems.
---
## About
This project is an open-source component of [js0.site ⋅ Refactoring the Internet Plan](https://js0.site).
We are redefining the development paradigm of the Internet in a componentized way. Welcome to follow us:
* [Google Group](https://groups.google.com/g/js0-site)
* [js0site.bsky.social](https://bsky.app/profile/js0site.bsky.social)
---
<a id="zh"></a>
# PickFast : 无锁加权负载均衡，优选低延迟节点
高性能加权随机选择库，支持原子 EMA 权重更新。专为负载均衡、A/B 测试、资源调度，以及需要基于概率选择与动态权重调整的场景设计。
## 目录
- [项目特性](#项目特性)
- [使用演示](#使用演示)
- [与 Race Crate 集成](#与-race-crate-集成)
- [自定义权重计算](#自定义权重计算)
- [设计思路](#设计思路)
- [API 介绍](#api-介绍)
- [技术堆栈](#技术堆栈)
- [目录结构](#目录结构)
- [历史小故事](#历史小故事)
## 项目特性
- **无锁更新**：基于 `AtomicU32` 和 CAS 实现线程安全的权重更新，无需互斥锁
- **自适应权重**：集成指数移动平均 (EMA) 算法，平滑延时波动
- **故障处理**：提供 `failed()` 方法快速惩罚性能不佳的节点
- **权重下限保护**：确保最小权重为 1，防止节点被完全排除
- **缓存友好**：结构对齐避免多核环境下的伪共享
- **灵活策略**：通过 `Rank` 特性支持自定义权重计算逻辑
- **零分配**：初始化后所有操作无内存分配
## 使用演示
DNS 服务器负载均衡场景，展示自动优选低延迟节点的效果：
```rust
use std::{net::IpAddr, sync::Arc, thread};
use pick_fast::PickFast;
#[derive(Debug, Clone, Copy)]
struct DnsServer {
ip: IpAddr,
}
// 初始化不同延迟特性的 DNS 服务器
let servers = [
DnsServer { ip: "8.8.8.8".parse().unwrap() }, // Google DNS - 100ms
DnsServer { ip: "1.1.1.1".parse().unwrap() }, // Cloudflare - 80ms
DnsServer { ip: "223.5.5.5".parse().unwrap() }, // 阿里 DNS - 5ms (最快)
DnsServer { ip: "208.67.222.222".parse().unwrap() }, // OpenDNS - 60ms
];
// 创建负载均衡器，使用默认 Inverse 策略
let lb = Arc::new(PickFast::new(servers));
// 模拟多线程并发使用
let handles: Vec<_> = (0..4)
let latency_us = match node.ip.to_string().as_str() {
"8.8.8.8" => 100_000, // 100ms
"1.1.1.1" => 80_000, // 80ms
"223.5.5.5" => 5_000, // 5ms (最快)
"208.67.222.222" => 60_000, // 60ms
_ => 100_000,
};
// 用观测到的延迟更新节点性能
c_lb.set(node.index, latency_us);
// 如果延迟过高，可选择标记节点失败
if latency_us > 200_000 { // > 200ms
c_lb.failed(node.index);
}
}
})
})
.collect();
// 等待所有线程完成
for handle in handles {
handle.join().unwrap();
}
// 检查最终权重 - 快节点应该有更高权重
let fast_weight = lb.li[2].weight.load(std::sync::atomic::Ordering::Relaxed);
let slow_weight = lb.li[0].weight.load(std::sync::atomic::Ordering::Relaxed);
println!("快节点权重: {fast_weight}, 慢节点权重: {slow_weight}");
println!("比例: {:.2}", fast_weight as f64 / slow_weight as f64);
```
## DNS 负载均衡与 race 集成
```rust
use pick_fast::PickFast;
use race::race;
use std::{net::IpAddr, sync::Arc, time::Duration};
use tokio::net::lookup_host;
#[derive(Debug, Clone, Copy)]
struct DnsServer {
ip: IpAddr,
}
// 真实 DNS 服务器 IP 地址
const DNS_SERVERS: [&str; 4] = [
"8.8.8.8:53", // Google DNS
"1.1.1.1:53", // Cloudflare DNS
"9.9.9.9:53", // Quad9 DNS
"208.67.222.222:53", // OpenDNS
];
// 创建服务器节点
let servers: Vec<DnsServer> = DNS_SERVERS
.iter()
})
.collect();
let lb = Arc::new(PickFast::<DnsServer, pick_fast::Inverse>::new(servers));
// 带性能跟踪的 DNS 任务
#[derive(Debug, Clone)]
struct DnsTask {
host: &'static str,
index: usize,
start: std::time::Instant,
}
// 使用 iter() 获取加权随机起始序列
let iter = lb.iter();
let selected_servers: Vec<_> = iter.take(4).collect(); // 尝试前 4 个服务器
// 带自动权重更新的 DNS 解析
let lb_clone = lb.clone();
async move {
// 模拟DNS查询延时
let start = std::time::Instant::now();
match lookup_host("example.com:80").await {
Ok(mut addrs) => {
if let Some(addr) = addrs.next() {
let duration = start.elapsed();
let latency_us = duration.as_micros() as u32;
// 成功：用实际延时更新
lb.set(task.index, latency_us);
println!(" ✅ {} 查询成功，耗时 {duration:?} (延时: {latency_us}μs)", task.host);
Ok(addr.ip())
} else {
// 未找到地址：惩罚服务器
lb.failed(task.index);
println!(" ❌ {} 查询失败", task.host);
Err(std::io::Error::new(std::io::ErrorKind::NotFound, "未找到地址"))
}
}
Err(e) => {
// 网络错误：惩罚服务器
lb.failed(task.index);
println!(" ❌ {} 查询失败: {e}", task.host);
Err(e)
}
}
}
};
// 创建 DNS 任务迭代器用于 race
DnsTask {
host: DNS_SERVERS[index],
index,
start: std::time::Instant::now(),
}
});
// 执行阶梯式 DNS 解析
let rx = race(resolve_with_feedback, dns_tasks_iter, Duration::from_millis(500));
// 等待第一个成功的解析结果
while let Ok(result) = rx.recv().await {
match result {
Ok(ip) => {
println!("🎯 首个成功解析: {ip}");
drop(rx); // 停止其他任务
break;
}
Err(_) => continue, // 尝试下一个服务器
}
}
```
## 自定义权重计算
`Rank` 特性定义观测值（如延时）如何转换为选择权重。
```rust
use pick_fast::Rank;
/// 优先级排序：优先级越高，权重越大
pub struct Priority;
impl Rank for Priority {
fn calc(priority: u32) -> u32 {
priority // 直接映射：优先级值即权重
}
}
// 使用示例
let lb = PickFast::<Task, Priority>::new(tasks);
lb.set(index, 100); // 设置优先级为 100
```
内置 `Inverse` 策略适用于基于延时的选择：
```text
权重 = (1 << 22) / max(延时, 1)
初始权重 = (1 << 22) / max(节点数, 1)
```
设计考量：2²² 基数确保 256 节点在 1μs 延时下不会溢出 `u32`。初始权重与节点数成反比，确保合理的初始分布。
## 设计思路
架构核心：最小化同步开销，最大化吞吐量。
### 调用流程
```mermaid
graph TD
A[客户端请求] --> B["pick()"]
B --> C[原子加载总权重]
C --> D[随机目标选择]
D --> E["线性扫描节点 O(N)"]
E --> F[返回 Handle]
G[性能反馈] --> H["set()"]
H --> I[通过 Rank 计算目标权重]
I --> J[CAS 更新节点权重 含EMA]
J --> K[原子更新总权重]
```
### EMA 平滑
```text
新权重 = (旧权重 + 目标权重) / 2
```
平滑机制防止瞬时尖峰导致权重剧烈变化。
## API 介绍
### 核心类型
| `PickFast<T, M>` | 负载均衡器主体。`T`: 节点数据，`M`: 权重模型 |
| `PickFast.li` | `Vec<Node<T>>` - 节点列表 |
| `PickFast.total` | `AtomicU32` - 缓存的总权重 |
| `Node<T>` | 节点结构，包含数据和权重 |
| `Node.data` | `T` - 节点数据 |
| `Node.weight` | `AtomicU32` - 节点权重 |
### 关键方法
| `new(data: impl IntoIterator<Item = T>) -> Self` | 从迭代器创建实例 |
| `len(&self) -> usize` | 获取节点数量 |
| `is_empty(&self) -> bool` | 检查是否为空 |
| `pick(&self) -> Handle<'_, T>` | 基于当前权重选择节点。O(1) 权重加载 + O(N) 扫描 |
| `set(&self, index: usize, val: u32)` | 更新节点观测值，使用 EMA 平滑（最小权重：1） |
| `failed(&self, index: usize)` | 标记节点失败，权重减半（最小权重：1） |
| `iter(&self) -> CIter<'_, Node<T>>` | 创建循环迭代器，加权随机起始位置（需要 `iter` 特性） |
### 其他类型
| `Handle<'a, T>` | 选中节点的智能指针，包含 `index` 和 `node` 引用 |
| `Rank` | 自定义权重计算逻辑的特性 |
| `Inverse` | 默认策略：权重与延时成反比 |
## 技术堆栈
| 编程语言 | Rust (Edition 2024) |
| 随机算法 | `fastrand` |
| 并发控制 | `std::sync::atomic` |
| 测试与可视化 | `plotters`, `svg` |
## 目录结构
```text
.
├── Cargo.toml # 项目配置
├── src/
│ └── lib.rs # 核心实现
├── tests/
│ └── main.rs # 集成测试与图表生成
└── readme/
├── en.md # 英文文档
├── zh.md # 中文文档
├── rank-en.svg # 英文性能图表
└── rank-zh.svg # 中文性能图表
```
## 历史小故事
加权负载均衡起源于网络数据包调度。1991 年，"加权轮询" (Weighted Round Robin, WRR) 概念在 ATM (异步传输模式) 网络中被正式提出，用于处理异构链路速度的差异化调度需求。
从 WRR 到现代加权随机选择，代表着范式转变：从确定性槽位分配，转向概率方法实现自然的负载分布。结合指数移动平均 (EMA) 平滑——该技术源自 1960 年代的股票市场技术分析——算法能优雅地适应网络条件变化。
有趣的是，这里使用的 CAS (Compare-And-Swap) 原语可追溯至 1970 年的 IBM System/370，使得无锁编程概念已有 50 余年历史——但它仍是现代高性能并发系统的基石。
---
## 关于
本项目为 [js0.site ⋅ 重构互联网计划](https://js0.site) 的开源组件。
我们正在以组件化的方式重新定义互联网的开发范式，欢迎关注：
* [谷歌邮件列表](https://groups.google.com/g/js0-site)
* [js0site.bsky.social](https://bsky.app/profile/js0site.bsky.social)