442 lines
12 KiB
Markdown
442 lines
12 KiB
Markdown
# announcement
|
|
|
|
[](https://crates.io/crates/announcement)
|
|
[](https://docs.rs/announcement)
|
|
[](https://opensource.org/licenses/MIT)
|
|
[](https://www.rust-lang.org)
|
|
|
|
A runtime-agnostic oneshot broadcast channel for Rust.
|
|
|
|
Broadcast a value once to multiple listeners with minimal overhead and zero data duplication when used with `Arc<T>`.
|
|
|
|
## Features
|
|
|
|
- **Simple API** - Create, announce, listen
|
|
- **Fast** - Built on `Arc<OnceLock>` and `event-listener`
|
|
- **Runtime-agnostic** - Works with tokio, async-std, smol, or any executor
|
|
- **Type-safe** - Can only announce once, listeners are cloneable
|
|
- **Lightweight** - Minimal dependencies
|
|
- **Cancel-safe** - All async operations are cancel-safe
|
|
|
|
## Performance
|
|
|
|
**Sub-microsecond operations with linear scaling**
|
|
|
|
- Channel creation: ~50ns
|
|
- Listener creation: ~10ns per listener
|
|
- Announcement: ~60ns + 11ns per listener
|
|
- Lock-free retrieval: ~0.6ns (sub-nanosecond)
|
|
|
|
**Critical recommendation**: Use `Arc<T>` for types >64 bytes or with 3+ listeners for dramatic performance gains (up to 600x faster for large types).
|
|
|
|
📊 **[Full Performance Analysis →](PERFORMANCE_ANALYSIS.md)** - Comprehensive benchmarks, scaling analysis, and optimization guide
|
|
|
|
## Installation
|
|
|
|
Add this to your `Cargo.toml`:
|
|
|
|
```toml
|
|
[dependencies]
|
|
announcement = "0.1"
|
|
```
|
|
|
|
For tracing support:
|
|
```toml
|
|
[dependencies]
|
|
announcement = { version = "0.1", features = ["tracing"] }
|
|
```
|
|
|
|
For blocking-only (no std):
|
|
```toml
|
|
[dependencies]
|
|
announcement = { version = "0.1", default-features = false }
|
|
```
|
|
|
|
## Quick Start
|
|
|
|
```rust
|
|
use announcement::Announcement;
|
|
|
|
#[tokio::main]
|
|
async fn main() {
|
|
// Create announcement channel
|
|
let (announcer, announcement) = Announcement::new();
|
|
|
|
// Create multiple listeners
|
|
let listener1 = announcement.listener();
|
|
let listener2 = announcement.listener();
|
|
|
|
// Spawn tasks that wait for the announcement
|
|
tokio::spawn(async move {
|
|
let value = listener1.listen().await;
|
|
println!("Listener 1 received: {}", value);
|
|
});
|
|
|
|
tokio::spawn(async move {
|
|
let value = listener2.listen().await;
|
|
println!("Listener 2 received: {}", value);
|
|
});
|
|
|
|
// Announce to all listeners at once
|
|
announcer.announce(42).unwrap();
|
|
}
|
|
```
|
|
|
|
## Learning Guide
|
|
|
|
The `examples/` directory contains a progressive learning guide. **Read these in order:**
|
|
|
|
1. **[01_basic_usage.rs](examples/01_basic_usage.rs)** - Start here! Creating channels, announcing, try_listen()
|
|
2. **[02_async_listening.rs](examples/02_async_listening.rs)** - Async operations with listen().await
|
|
3. **[03_multiple_listeners.rs](examples/03_multiple_listeners.rs)** - Broadcasting to multiple listeners
|
|
4. **[04_efficient_broadcasting.rs](examples/04_efficient_broadcasting.rs)** - Using Arc<T> for efficiency
|
|
5. **[05_blocking_operations.rs](examples/05_blocking_operations.rs)** - Blocking operations and deadlock prevention
|
|
|
|
Each example is **heavily documented** with explanations, tips, and best practices. They're meant to be read and learned from, not just run.
|
|
|
|
### Advanced Examples
|
|
|
|
After completing the guide, check out these real-world patterns:
|
|
|
|
- **[shutdown.rs](examples/shutdown.rs)** - Graceful shutdown signal pattern
|
|
- **[config_broadcast.rs](examples/config_broadcast.rs)** - Configuration broadcast to workers
|
|
- **[lazy_init.rs](examples/lazy_init.rs)** - Lazy resource initialization pattern
|
|
|
|
## Usage Examples
|
|
|
|
### Shutdown Signal Pattern
|
|
|
|
Coordinate graceful shutdown of multiple worker tasks:
|
|
|
|
```rust
|
|
use announcement::Announcement;
|
|
|
|
#[tokio::main]
|
|
async fn main() {
|
|
let (shutdown_tx, shutdown_rx) = Announcement::new();
|
|
|
|
// Spawn workers with shutdown listeners
|
|
let worker = tokio::spawn({
|
|
let shutdown = shutdown_rx.listener();
|
|
async move {
|
|
loop {
|
|
tokio::select! {
|
|
_ = shutdown.listen() => {
|
|
println!("Shutting down gracefully...");
|
|
break;
|
|
}
|
|
_ = do_work() => {}
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// Later: broadcast shutdown signal
|
|
shutdown_tx.announce(()).unwrap();
|
|
worker.await.unwrap();
|
|
}
|
|
|
|
async fn do_work() {
|
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
|
}
|
|
```
|
|
|
|
### Configuration Broadcast
|
|
|
|
Share initialized configuration to multiple workers using `Arc` for efficiency:
|
|
|
|
```rust
|
|
use announcement::Announcement;
|
|
use std::sync::Arc;
|
|
|
|
#[derive(Clone, Debug)]
|
|
struct Config {
|
|
database_url: String,
|
|
api_key: String,
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() {
|
|
let (config_tx, config_rx) = Announcement::new();
|
|
|
|
// Workers can start before config is ready
|
|
let worker = tokio::spawn({
|
|
let config_listener = config_rx.listener();
|
|
async move {
|
|
// Wait for config to be ready
|
|
let config = config_listener.listen().await;
|
|
println!("Worker received config: {:?}", config);
|
|
}
|
|
});
|
|
|
|
// Initialize config (takes time)
|
|
let config = Arc::new(Config {
|
|
database_url: "postgresql://localhost/db".to_string(),
|
|
api_key: "secret".to_string(),
|
|
});
|
|
|
|
// Broadcast to all workers (Arc is cloned, not the data)
|
|
config_tx.announce(config).unwrap();
|
|
|
|
worker.await.unwrap();
|
|
}
|
|
```
|
|
|
|
### Lazy Initialization
|
|
|
|
Signal when a resource is ready:
|
|
|
|
```rust
|
|
use announcement::Announcement;
|
|
use std::sync::Arc;
|
|
|
|
struct Database { /* ... */ }
|
|
|
|
impl Database {
|
|
async fn connect() -> Self {
|
|
// Expensive initialization
|
|
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
|
Database { /* ... */ }
|
|
}
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() {
|
|
let (db_tx, db_rx) = Announcement::new();
|
|
|
|
// Services wait for database
|
|
let api_service = tokio::spawn({
|
|
let db = db_rx.listener();
|
|
async move {
|
|
let db = db.listen().await;
|
|
// Use database
|
|
}
|
|
});
|
|
|
|
// Initialize database
|
|
let db = Arc::new(Database::connect().await);
|
|
db_tx.announce(db).unwrap();
|
|
|
|
api_service.await.unwrap();
|
|
}
|
|
```
|
|
|
|
### Non-blocking Check
|
|
|
|
Use `try_listen()` to check without waiting:
|
|
|
|
```rust
|
|
use announcement::Announcement;
|
|
|
|
let (announcer, announcement) = Announcement::new();
|
|
let listener = announcement.listener();
|
|
|
|
// Check if value is ready
|
|
if let Some(value) = listener.try_listen() {
|
|
println!("Already announced: {}", value);
|
|
} else {
|
|
println!("Not announced yet");
|
|
}
|
|
|
|
announcer.announce(42).unwrap();
|
|
|
|
// Now it's available
|
|
assert_eq!(listener.try_listen(), Some(42));
|
|
```
|
|
|
|
## Clone vs Arc: When to Use Each
|
|
|
|
`Announcement<T>` requires `T: Clone`. Each listener clones the value.
|
|
|
|
### Use Direct Values
|
|
|
|
✓ For small types (< 16 bytes):
|
|
```rust
|
|
Announcement::<i32>::new()
|
|
Announcement::<(u64, u64)>::new()
|
|
```
|
|
|
|
✓ For Copy types:
|
|
```rust
|
|
Announcement::<f64>::new()
|
|
```
|
|
|
|
✓ For 1-2 listeners (clone cost amortized)
|
|
|
|
### Use Arc for Efficiency
|
|
|
|
✓ For large types:
|
|
```rust
|
|
Announcement::<Arc<Vec<u8>>>::new() // Instead of Vec<u8>
|
|
```
|
|
|
|
✓ For expensive clones:
|
|
```rust
|
|
Announcement::<Arc<Config>>::new() // Config has String, Vec, HashMap
|
|
```
|
|
|
|
✓ For many listeners (3+):
|
|
```rust
|
|
let (tx, announcement) = Announcement::<Arc<Data>>::new();
|
|
// Create 1000 listeners - only 1 Data allocation!
|
|
```
|
|
|
|
### Memory Savings Example
|
|
|
|
Broadcasting 1MB data to 100 listeners:
|
|
|
|
- **Without Arc**: 100 MB (100 allocations)
|
|
- **With Arc**: 1 MB (1 allocation)
|
|
- **Savings**: 99 MB, 200,000x faster
|
|
|
|
See [example 04](examples/04_efficient_broadcasting.rs) for detailed explanation.
|
|
|
|
## Use Cases
|
|
|
|
- **Shutdown signals** - Notify all tasks to shut down gracefully
|
|
- **Configuration broadcast** - Share initialized config to all workers
|
|
- **Lazy initialization** - Signal when a resource is ready (database, cache, etc.)
|
|
- **Event fanout** - Broadcast a one-time event to multiple subscribers
|
|
- **Barrier synchronization** - Wait for initialization to complete
|
|
- **State transitions** - Signal when application reaches a certain state
|
|
|
|
## Performance
|
|
|
|
**Benchmarked on AMD Ryzen 7 7840U (8C/16T), 64GB RAM, Framework 13 laptop**
|
|
|
|
### Core Operations
|
|
|
|
| Operation | Time | Notes |
|
|
|-----------|------|-------|
|
|
| Channel creation | ~150ns | 2 allocations |
|
|
| Listener creation | ~15ns | 2 Arc clones |
|
|
| Announce (no listeners) | ~30ns | Non-blocking |
|
|
| Announce (N listeners) | ~30ns + wakeup | O(N) wakeup |
|
|
| try_listen() hit | ~8ns | Lock-free read |
|
|
| try_listen() miss | ~5ns | Lock-free read |
|
|
| is_announced() | ~5ns | Lock-free read |
|
|
|
|
### Clone vs Arc Comparison
|
|
|
|
For **large types** (1MB) with **100 listeners**:
|
|
|
|
| Method | Memory | Time | Speedup |
|
|
|--------|--------|------|---------|
|
|
| Direct Clone | 100 MB | ~100ms | 1x |
|
|
| Arc Clone | 1 MB | ~500ns | **200,000x** |
|
|
|
|
**Rule of thumb:** Use `Arc<T>` for:
|
|
- Types > 8 bytes with multiple listeners
|
|
- Expensive-to-clone types (Vec, String, HashMap)
|
|
|
|
### Scalability
|
|
|
|
- **1,000 listeners**: ~15µs announce, ~8µs retrieve all
|
|
- **10,000 listeners**: ~150µs announce, ~80µs retrieve all
|
|
- **100,000 listeners**: ~1.5ms announce, ~800µs retrieve all
|
|
|
|
Linear scaling O(N) for wakeup, constant O(1) for retrieval.
|
|
|
|
## Testing
|
|
|
|
Run tests (recommended - faster):
|
|
```bash
|
|
cargo nextest run
|
|
```
|
|
|
|
Or with standard test runner:
|
|
```bash
|
|
cargo test
|
|
```
|
|
|
|
**Test Coverage:** 507+ tests covering:
|
|
- All primitive types
|
|
- Collections, smart pointers, trait objects
|
|
- Edge cases (ZST, 1MB+ types, recursive structures)
|
|
- Concurrency (1000+ threads)
|
|
- Race conditions (10,000 iterations)
|
|
- Cross-runtime compatibility (tokio, async-std, smol)
|
|
- Property-based testing with proptest
|
|
|
|
## Comparison to Alternatives
|
|
|
|
### vs `tokio::sync::broadcast`
|
|
|
|
| Feature | `announcement` | `tokio::sync::broadcast` |
|
|
|---------|---------------|-------------------------|
|
|
| One-shot | Yes | No (multi-shot) |
|
|
| Runtime-agnostic | Yes | No (tokio only) |
|
|
| Bounded capacity | N/A (single value) | Yes |
|
|
| Lagging handling | N/A | Required |
|
|
| API complexity | Simple | More complex |
|
|
| Use case | One-time events | Continuous streams |
|
|
|
|
**Use `announcement` when:**
|
|
- You need to broadcast once (shutdown, config, initialization)
|
|
- You want runtime independence
|
|
- You want simpler semantics
|
|
|
|
**Use `tokio::sync::broadcast` when:**
|
|
- You need to send multiple values
|
|
- You're already using tokio
|
|
- You need bounded buffering
|
|
|
|
### vs `tokio::sync::oneshot`
|
|
|
|
| Feature | `announcement` | `tokio::sync::oneshot` |
|
|
|---------|---------------|------------------------|
|
|
| Multiple receivers | Yes | No (1-to-1) |
|
|
| Runtime-agnostic | Yes | No (tokio only) |
|
|
| Cloneable receiver | Yes | No |
|
|
| Memory per listener | ~16 bytes | Full channel per pair |
|
|
|
|
**Use `announcement` when:**
|
|
- You need 1-to-many communication
|
|
- You want to create listeners dynamically
|
|
|
|
**Use `tokio::sync::oneshot` when:**
|
|
- You only need 1-to-1 communication
|
|
- You're using tokio
|
|
|
|
### vs Manual `Arc<OnceLock>` + `Event`
|
|
|
|
`announcement` provides a safe, ergonomic wrapper around this pattern with:
|
|
- Type-safe oneshot semantics (can't announce twice)
|
|
- Proper TOCTOU protection in async contexts
|
|
- Clear ownership model
|
|
- Documented best practices
|
|
|
|
## Contributing
|
|
|
|
Contributions are welcome! Please see [CODE_OF_CONDUCT.md](CODE_OF_CONDUCT.md) for community guidelines.
|
|
|
|
## License
|
|
|
|
This project is licensed under the MIT License.
|
|
|
|
Copyright (c) 2025 Stephen Waits <steve@waits.net>
|
|
|
|
Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
of this software and associated documentation files (the "Software"), to deal
|
|
in the Software without restriction, including without limitation the rights
|
|
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
copies of the Software, and to permit persons to whom the Software is
|
|
furnished to do so, subject to the following conditions:
|
|
|
|
The above copyright notice and this permission notice shall be included in all
|
|
copies or substantial portions of the Software.
|
|
|
|
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
|
SOFTWARE.
|
|
|
|
## See Also
|
|
|
|
- [API Documentation](https://docs.rs/announcement)
|
|
- [Examples](examples/)
|
|
- [Changelog](CHANGELOG.md)
|