mill_io/lib.rs
1//! # Mill-IO
2//! A lightweight, production-ready event loop library for Rust that provides efficient non-blocking I/O management
3//! without relying on heavyweight async runtimes like Tokio.
4//! Mill-IO is a modular, reactor-based event loop built on top of [`mio`], offering cross-platform polling,
5//! configurable thread pool integration, and object pooling for high-performance applications that need
6//! fine-grained control over their I/O operations.
7//! ## Core Philosophy
8//! Mill-IO was designed for applications that require:
9//! - **Predictable performance** with minimal runtime overhead
10//! - **Runtime-agnostic architecture** that doesn't force async/await patterns
11//! - **Direct control** over concurrency and resource management
12//! - **Minimal dependencies** for reduced attack surface and faster builds
13//! ## Features
14//! - **Runtime-agnostic**: No dependency on Tokio or other async runtimes
15//! - **Cross-platform**: Leverages mio's polling abstraction (epoll, kqueue, IOCP)
16//! - **Thread pool integration**: Configurable worker threads for handling I/O events
17//! - **Object pooling**: Reduces allocation overhead for frequent operations
18//! - **Clean API**: Simple registration and handler interface
19//! - **Thread-safe**: Lock-free operations in hot paths
20//! ## Architecture Overview
21//! ```text
22//! +-------------+ +--------------+ +-------------+
23//! | EventLoop |----| Reactor |----| PollHandle |
24//! +-------------+ +--------------+ +-------------+
25//! |
26//! |
27//! +--------------+ +-------------+
28//! | ThreadPool |----| Workers |
29//! +--------------+ +-------------+
30//! ```
31//! ## Quick Start
32//!
33//! ```rust,no_run
34//! use mill_io::{EventLoop, EventHandler};
35//! use mio::{net::TcpListener, Interest, Token, event::Event};
36//! use std::net::SocketAddr;
37//!
38//! struct EchoHandler;
39//!
40//! impl EventHandler for EchoHandler {
41//! fn handle_event(&self, event: &Event) {
42//! println!("Received event: {:?}", event);
43//! // Handle incoming connections and data
44//! }
45//! }
46//!
47//! fn main() -> Result<(), Box<dyn std::error::Error>> {
48//! // Create event loop with default configuration
49//! let event_loop = EventLoop::default();
50//!
51//! // Bind to localhost
52//! let addr: SocketAddr = "127.0.0.1:8080".parse()?;
53//! let mut listener = TcpListener::bind(addr)?;
54//!
55//! // Register the listener with a handler
56//! event_loop.register(
57//! &mut listener,
58//! Token(1),
59//! Interest::READABLE,
60//! EchoHandler
61//! )?;
62//!
63//! println!("Server listening on 127.0.0.1:8080");
64//!
65//! // Start the event loop (blocks until stopped)
66//! event_loop.run()?;
67//!
68//! Ok(())
69//! }
70//! ```
71//!
72//! ```rust,no_run
73//! use mill_io::EventLoop;
74//!
75//! fn main() -> Result<(), Box<dyn std::error::Error>> {
76//! let event_loop = EventLoop::new(
77//! 8, // 8 worker threads
78//! 1024, // Handle up to 1024 events per poll
79//! 100 // 100ms poll timeout
80//! )?;
81//! Ok(())
82//! }
83//! ```
84//!
85//! - [`EventLoop`]: Main entry point for registering I/O sources and running the event loop
86//! - [`EventHandler`]: Trait for implementing custom event handling logic
87//! - [`reactor`]: Core reactor implementation managing the event loop lifecycle
88//! - [`thread_pool`]: Configurable thread pool for distributing work
89//! - [`poll`]: Cross-platform polling abstraction and handler registry
90//! - [`error`]: Error types and result handling
91//!
92//! For comprehensive examples and architectural details, see the [README](../README.md)
93//! and [Architecture Guide](../docs/Arch.md).
94
95#![cfg_attr(feature = "unstable-mpmc", feature(mpmc_channel))]
96
97use mio::{Interest, Token};
98pub mod error;
99pub mod handler;
100pub mod object_pool;
101pub mod poll;
102pub mod reactor;
103pub mod thread_pool;
104
105pub use handler::EventHandler;
106pub use mio::event::Event;
107pub use object_pool::{ObjectPool, PooledObject};
108pub use thread_pool::{ComputePoolMetrics, TaskPriority};
109
110use crate::{error::Result, reactor::ReactorOptions};
111
112/// A convenient prelude module that re-exports commonly used types and traits.
113///
114/// This module provides a convenient way to import the most commonly used items from mill-io:
115///
116/// ```rust
117/// use mill_io::prelude::*;
118/// ```
119///
120/// This brings into scope:
121/// - [`EventHandler`] - Trait for implementing event handling logic
122/// - [`ObjectPool`] and [`PooledObject`] - Object pooling utilities
123/// - [`reactor::Reactor`] - Core reactor implementation (advanced usage)
124/// - [`thread_pool::ThreadPool`] - Thread pool implementation (advanced usage)
125pub mod prelude {
126 pub use crate::handler::EventHandler;
127 pub use crate::object_pool::{ObjectPool, PooledObject};
128 pub use crate::reactor::{self, Reactor};
129 pub use crate::thread_pool::{self, ComputePoolMetrics, TaskPriority, ThreadPool};
130}
131
132/// The main event loop structure for registering I/O sources and handling events.
133///
134/// `EventLoop` is the primary interface for Mill-IO, providing a simple API for:
135/// - Registering I/O sources (sockets, files, etc.) with event handlers
136/// - Starting and stopping the event loop
137/// - Managing the underlying reactor and thread pool
138///
139/// The event loop uses a reactor pattern internally, where I/O events are detected
140/// by the polling mechanism and dispatched to registered handlers via a thread pool.
141///
142/// ## Example
143///
144/// Basic usage with default configuration:
145///
146/// ```rust,no_run
147/// use mill_io::{EventLoop, EventHandler};
148/// use mio::{net::TcpListener, Interest, Token, event::Event};
149/// use std::net::SocketAddr;
150///
151/// struct MyHandler;
152/// impl EventHandler for MyHandler {
153/// fn handle_event(&self, event: &Event) {
154/// println!("Event received: {:?}", event);
155/// }
156/// }
157///
158/// let event_loop = EventLoop::default();
159/// let addr: SocketAddr = "127.0.0.1:0".parse()?;
160/// let mut listener = TcpListener::bind(addr)?;
161///
162/// event_loop.register(&mut listener, Token(0), Interest::READABLE, MyHandler)?;
163/// event_loop.run()?; // Blocks until stopped
164/// # Ok::<(), Box<dyn std::error::Error>>(())
165/// ```
166///
167/// Custom configuration:
168///
169/// ```rust,no_run
170/// use mill_io::EventLoop;
171///
172/// let event_loop = EventLoop::new(
173/// 4, // 4 worker threads
174/// 512, // Buffer for 512 events per poll
175/// 50 // 50ms poll timeout
176/// )?;
177/// # Ok::<(), Box<dyn std::error::Error>>(())
178/// ```
179pub struct EventLoop {
180 reactor: reactor::Reactor,
181}
182
183impl Default for EventLoop {
184 /// Creates a new `EventLoop` with default configuration.
185 ///
186 /// The default configuration uses:
187 /// - Number of worker threads equal to available CPU cores, falling back to 4 threads if CPU detection fails ([`thread_pool::DEFAULT_POOL_CAPACITY`])
188 /// - 1024 events capacity ([`reactor::DEFAULT_EVENTS_CAPACITY`])
189 /// - 150ms poll timeout ([`reactor::DEFAULT_POLL_TIMEOUT_MS`])
190 ///
191 /// # Panics
192 ///
193 /// Panics if the reactor cannot be initialized with default settings.
194 fn default() -> Self {
195 let reactor = reactor::Reactor::default();
196 Self { reactor }
197 }
198}
199
200impl EventLoop {
201 /// Creates a new `EventLoop` with custom configuration.
202 ///
203 /// ## Arguments
204 /// * `workers` - Number of worker threads in the thread pool (recommended: num_cpus)
205 /// * `events_capacity` - Maximum number of events to poll per iteration (typical: 512-4096)
206 /// * `poll_timeout_ms` - Poll timeout in milliseconds (balance between latency and CPU usage)
207 ///
208 /// ## Errors
209 ///
210 /// Returns an error if:
211 /// - The reactor cannot be initialized
212 /// - The thread pool cannot be created
213 /// - The polling mechanism fails to initialize
214 ///
215 /// ## Example
216 ///
217 /// ```rust,no_run
218 /// use mill_io::EventLoop;
219 ///
220 /// // High-throughput configuration
221 /// let event_loop = EventLoop::new(8, 2048, 50)?;
222 ///
223 /// // Low-latency configuration
224 /// let event_loop = EventLoop::new(2, 256, 10)?;
225 /// # Ok::<(), Box<dyn std::error::Error>>(())
226 /// ```
227 pub fn new(workers: usize, events_capacity: usize, poll_timeout_ms: u64) -> Result<Self> {
228 let reactor = reactor::Reactor::new(workers, events_capacity, poll_timeout_ms)?;
229 Ok(Self { reactor })
230 }
231
232 /// Creates a new EventLoop optimized for low latency.
233 ///
234 /// This mode:
235 /// - Executes handlers directly on the reactor thread (no thread pool dispatch)
236 /// - Uses thread-local buffer pools (no lock contention)
237 /// - Best for I/O-bound workloads with fast handlers
238 ///
239 /// WARNING: Slow handlers will block all I/O processing!
240 pub fn new_low_latency(events_capacity: usize, poll_timeout_ms: u64) -> Result<Self> {
241 let reactor = reactor::Reactor::new_with_options(
242 1, // Minimal pool for compute tasks
243 events_capacity,
244 poll_timeout_ms,
245 ReactorOptions {
246 direct_dispatch: true,
247 },
248 )?;
249 Ok(Self { reactor })
250 }
251
252 /// Registers an I/O source with the event loop and associates it with a handler.
253 ///
254 /// This method registers an I/O source (such as a TCP listener or socket) with the event loop.
255 /// When events occur on the source, the provided handler will be invoked on a worker thread.
256 ///
257 ///
258 /// ## Arguments
259 /// * `source` - The I/O source to register (e.g., [`mio::net::TcpListener`])
260 /// * `token` - Unique token for identifying events from this source
261 /// * `interests` - I/O events to listen for ([`mio::Interest::READABLE`], [`mio::Interest::WRITABLE`])
262 /// * `handler` - Event handler that will process events from this source
263 ///
264 /// ## Errors
265 ///
266 /// Returns an error if:
267 /// - The token is already in use
268 /// - The source cannot be registered with the underlying poll mechanism
269 /// - The handler registry is full
270 ///
271 /// ## Example
272 ///
273 /// ```rust,no_run
274 /// use mill_io::{EventLoop, EventHandler};
275 /// use mio::{net::TcpListener, Interest, Token, event::Event};
276 /// use std::net::SocketAddr;
277 ///
278 /// struct ConnectionHandler;
279 /// impl EventHandler for ConnectionHandler {
280 /// fn handle_event(&self, event: &Event) {
281 /// // Handle new connections
282 /// }
283 /// }
284 /// fn main() -> Result<(), Box<dyn std::error::Error>> {
285 /// let event_loop = EventLoop::default();
286 /// let addr: SocketAddr = "0.0.0.0:8080".parse()?;
287 /// let mut listener = TcpListener::bind(addr)?;
288 ///
289 /// event_loop.register(
290 /// &mut listener,
291 /// Token(0),
292 /// Interest::READABLE,
293 /// ConnectionHandler
294 /// )?;
295 /// Ok(())
296 /// }
297 /// ```
298 pub fn register<H, S>(
299 &self,
300 source: &mut S,
301 token: Token,
302 interests: Interest,
303 handler: H,
304 ) -> Result<()>
305 where
306 H: EventHandler + Send + Sync + 'static,
307 S: mio::event::Source + ?Sized,
308 {
309 self.reactor
310 .poll_handle
311 .register(source, token, interests, handler)
312 }
313
314 /// Deregisters an I/O source from the event loop.
315 ///
316 /// Removes the source from the polling mechanism and clears its associated handler.
317 /// After deregistration, no more events will be delivered for this source.
318 ///
319 /// ## #Arguments
320 /// * `source` - The I/O source to deregister
321 /// * `token` - Token associated with the source during registration
322 ///
323 /// ## Error
324 ///
325 /// Returns an error if:
326 /// - The source is not currently registered
327 /// - The deregistration fails at the OS level
328 /// - The token is invalid
329 ///
330 /// ## Example
331 ///
332 /// ```rust,no_run
333 /// use mill_io::{EventLoop, EventHandler};
334 /// use mio::{net::TcpListener, Interest, Token, event::Event};
335 /// use std::net::SocketAddr;
336 ///
337 /// struct Handler;
338 /// impl EventHandler for Handler {
339 /// fn handle_event(&self, _: &Event) {}
340 /// }
341 /// fn main() -> Result<(), Box<dyn std::error::Error>> {
342 ///
343 /// let event_loop = EventLoop::default();
344 /// let addr: SocketAddr = "127.0.0.1:0".parse()?;
345 /// let mut listener = TcpListener::bind(addr)?;
346 /// let token = Token(0);
347 ///
348 /// // Register
349 /// event_loop.register(&mut listener, token, Interest::READABLE, Handler)?;
350 ///
351 /// // Later, deregister
352 /// event_loop.deregister(&mut listener, token)?;
353 /// Ok(())
354 /// }
355 /// ```
356 pub fn deregister<S>(&self, source: &mut S, token: Token) -> Result<()>
357 where
358 S: mio::event::Source + ?Sized,
359 {
360 self.reactor.poll_handle.deregister(source, token)
361 }
362
363 /// Runs the event loop, blocking the current thread and dispatching events.
364 ///
365 /// This method starts the reactor's main loop, which will:
366 /// 1. Poll for I/O events using the configured timeout
367 /// 2. Dispatch events to registered handlers via the thread pool
368 /// 3. Continue until [`stop()`](Self::stop) is called or an error occurs
369 ///
370 /// The method blocks the calling thread and will only return when the event loop
371 /// is stopped or encounters a fatal error.
372 ///
373 /// ## Errors
374 ///
375 /// Returns an error if:
376 /// - The polling mechanism fails
377 /// - The thread pool encounters a fatal error
378 /// - System resources are exhausted
379 ///
380 /// ## Example
381 ///
382 /// ```rust,no_run
383 /// use mill_io::EventLoop;
384 ///
385 /// let event_loop = EventLoop::default();
386 /// // Register some handlers first...
387 /// event_loop.run()
388 /// # ; Ok::<(), Box<dyn std::error::Error>>(())
389 /// ```
390 pub fn run(&self) -> Result<()> {
391 self.reactor.run()
392 }
393
394 /// Submits a CPU-intensive task to the compute thread pool with default (Normal) priority.
395 ///
396 /// This method allows offloading heavy computations (e.g., cryptography, image processing)
397 /// to a dedicated thread pool, preventing the I/O event loop from being blocked.
398 ///
399 /// ## Arguments
400 /// * `task` - The closure to execute
401 ///
402 /// ## Example
403 ///
404 /// ```rust,no_run
405 /// use mill_io::EventLoop;
406 ///
407 /// let event_loop = EventLoop::default();
408 ///
409 /// event_loop.spawn_compute(|| {
410 /// // Heavy computation here
411 /// let result = 2 + 2;
412 /// println!("Computed: {}", result);
413 /// });
414 /// ```
415 pub fn spawn_compute<F>(&self, task: F)
416 where
417 F: FnOnce() + Send + 'static,
418 {
419 self.reactor.spawn_compute(task, TaskPriority::Normal);
420 }
421
422 /// Submits a CPU-intensive task to the compute thread pool with a specific priority.
423 ///
424 /// ## Arguments
425 /// * `task` - The closure to execute
426 /// * `priority` - The priority of the task
427 ///
428 /// ## Example
429 ///
430 /// ```rust,no_run
431 /// use mill_io::{EventLoop, TaskPriority};
432 ///
433 /// let event_loop = EventLoop::default();
434 ///
435 /// event_loop.spawn_compute_with_priority(|| {
436 /// // Heavy computation here
437 /// }, TaskPriority::High);
438 /// ```
439 pub fn spawn_compute_with_priority<F>(&self, task: F, priority: TaskPriority)
440 where
441 F: FnOnce() + Send + 'static,
442 {
443 self.reactor.spawn_compute(task, priority);
444 }
445
446 /// Returns metrics for the compute-intensive thread pool.
447 pub fn get_compute_metrics(&self) -> std::sync::Arc<ComputePoolMetrics> {
448 self.reactor.get_compute_metrics()
449 }
450
451 /// Signals the event loop to stop gracefully.
452 ///
453 /// This method initiates a graceful shutdown of the event loop. It sends a shutdown
454 /// signal to the reactor, which will cause the main loop to exit after finishing
455 /// the current polling cycle.
456 ///
457 /// This method is non-blocking and returns immediately. The actual shutdown happens
458 /// asynchronously, and [`run()`](Self::run) will return once the shutdown is complete.
459 ///
460 /// # Thread Safety
461 ///
462 /// This method is thread-safe and can be called from any thread, making it suitable
463 /// for use in signal handlers or from other threads.
464 ///
465 /// ## Example
466 ///
467 /// ```rust,no_run
468 /// use mill_io::EventLoop;
469 /// use std::thread;
470 /// use std::sync::Arc;
471 ///
472 /// let event_loop = Arc::new(EventLoop::default());
473 /// let event_loop_clone = Arc::clone(&event_loop);
474 ///
475 /// // Start event loop in background thread
476 /// let handle = thread::spawn(move || {
477 /// // In a real application, you would handle the result properly
478 /// let _ = event_loop_clone.run();
479 /// });
480 ///
481 /// // Stop after some time
482 /// thread::sleep(std::time::Duration::from_secs(1));
483 /// event_loop.stop();
484 ///
485 /// // Wait for shutdown
486 /// let _ = handle.join();
487 /// ```
488 pub fn stop(&self) {
489 let shutdown_handler = self.reactor.get_shutdown_handle();
490 shutdown_handler.shutdown();
491 }
492}