Skip to main content

net_io/
lib.rs

1//! # net_io
2//!
3//! The network IO substrate for the `-net` and `-io` crates: the entrance and
4//! exit every service shares. It owns listeners and sockets, the life of each
5//! connection, the limits and timeouts that keep a peer from holding what it
6//! should not, graceful drain, and the seams a connection gate and a TLS engine
7//! plug into. It is protocol-free and has no runtime dependencies.
8//!
9//! ## Quick start
10//!
11//! ```no_run
12//! fn main() -> Result<(), net_io::NetError> {
13//!     // An echo server: one handler call per connection on a bounded worker pool,
14//!     // with every default limit (idle, read, write, minimum rate, caps) applied.
15//!     let server = net_io::serve("127.0.0.1:7000", |conn| {
16//!         let mut buf = [0u8; 4096];
17//!         loop {
18//!             let n = conn.recv(&mut buf)?;
19//!             if n == 0 {
20//!                 return Ok(()); // the peer closed its side
21//!             }
22//!             conn.send_all(&buf[..n])?;
23//!         }
24//!     })?;
25//!
26//!     // A client.
27//!     let mut conn = net_io::connect("127.0.0.1:7000")?;
28//!     conn.send_all(b"ping")?;
29//!     let mut reply = [0u8; 4];
30//!     conn.recv_exact(&mut reply)?;
31//!     assert_eq!(&reply, b"ping");
32//!
33//!     // Stop: close the door, drain open connections, join the workers.
34//!     server.handle().request_stop();
35//!     let report = server.wait_for_stop();
36//!     println!("{} closed gracefully, {} killed", report.graceful, report.killed);
37//!     Ok(())
38//! }
39//! ```
40//!
41//! Everything else (several listeners, a gate, events, the control line) goes
42//! through [`ServerBuilder`]. The full reference with every item, its errors,
43//! and its cost is `docs/API.md` in the
44//! [repository](https://github.com/jamesgober/net-io).
45//!
46//! ## Status
47//!
48//! `0.5.0`: the reactor on epoll, kqueue, and IOCP, and the blocking pool, on
49//! Linux, macOS, FreeBSD, and Windows. TCP, Unix domain sockets (AF_UNIX on
50//! Windows), Windows named pipes, and UDP; the PROXY protocol, the TLS seam,
51//! systemd and FastCGI inheritance, listener hand-off on Unix and Windows, and
52//! the public [`Driver`] interface. Performance work and the remaining P2 items
53//! come in `0.6.0`.
54
55#![doc(html_root_url = "https://docs.rs/net-io")]
56#![cfg_attr(docsrs, feature(doc_cfg))]
57#![deny(unsafe_code)]
58#![deny(unsafe_op_in_unsafe_fn)]
59#![deny(missing_docs)]
60#![cfg_attr(
61    not(test),
62    deny(
63        clippy::unwrap_used,
64        clippy::expect_used,
65        clippy::panic,
66        clippy::todo,
67        clippy::unimplemented,
68        clippy::dbg_macro,
69        clippy::print_stdout,
70        clippy::print_stderr
71    )
72)]
73
74mod accept;
75mod addr;
76mod buf;
77mod completer;
78mod control;
79mod crc32c;
80mod driver;
81mod error;
82mod events;
83mod gate;
84mod handoff;
85mod id;
86mod info;
87mod inherit;
88mod limits;
89mod link;
90#[cfg(feature = "local")]
91mod local;
92#[cfg(all(windows, feature = "local"))]
93mod local_windows;
94mod opts;
95#[cfg(all(windows, feature = "local"))]
96mod pipe;
97mod proxy;
98#[cfg(feature = "proxy")]
99mod proxy_parse;
100mod rate;
101mod server;
102mod session;
103mod sys;
104mod tcp;
105#[cfg(test)]
106mod testing;
107mod time;
108mod tls;
109mod trusted;
110#[cfg(feature = "udp")]
111mod udp;
112#[cfg_attr(not(test), allow(dead_code))]
113mod wheel;
114
115/// Entry points for net-io's fuzz targets (`fuzz/`), compiled only under
116/// cargo-fuzz (`--cfg fuzzing`). Not part of the public API and not covered by semver.
117#[cfg(fuzzing)]
118#[doc(hidden)]
119pub mod __fuzz {
120    /// The hand-off manifest decoder.
121    pub fn manifest(data: &[u8]) {
122        crate::handoff::fuzz_manifest(data);
123    }
124
125    /// The systemd environment reader.
126    pub fn systemd_env(data: &[u8]) {
127        crate::inherit::fuzz_systemd_env(data);
128    }
129}
130
131/// Fault injection for net-io's own tests, compiled only with `--cfg net_io_shim`.
132/// Not part of the public API and not covered by semver.
133#[cfg(net_io_shim)]
134#[doc(hidden)]
135pub mod __shim {
136    pub use crate::sys::shim::{Fault, Op, clear, counted, program, start_counting, stop_counting};
137}
138
139use std::net::ToSocketAddrs;
140
141pub use addr::{AddrError, Cidr, PrefixKey, PrefixPolicy, normalize, normalize_socket};
142pub use buf::{Buffer, BufferPool, PoolStats};
143pub use completer::{Completer, PostError};
144pub use control::{ControlLine, ControlState, NoControl, SimpleControl};
145#[cfg(feature = "reactor")]
146pub use driver::new_reactor_driver;
147pub use driver::{
148    BorrowedSocket, Completion, Completions, Driver, DriverKind, DriverWaker, Operation,
149    OperationKind, Rejected, SendChunks, Token,
150};
151pub use error::{
152    CloseReason, ConfigReason, Direction, ErrorClass, KillReason, LimitKind, NetError,
153    OsErrorClass, RefusalCause, TimeoutKind,
154};
155pub use events::{CoalescedKind, Event, EventSink, NoopSink};
156pub use gate::{AdmitStage, Admitted, AllowAll, Denial, Gate, KillHandle, Refusal, WriteCheck};
157pub use handoff::{HandoffError, HandoffPeer, ListenerHandle};
158#[cfg(all(windows, feature = "local"))]
159pub use handoff::{SharedListeners, receive_shared, share_listeners};
160#[cfg(all(unix, feature = "local"))]
161pub use handoff::{recv_listeners, send_listeners};
162pub use id::{ConnectionId, ListenerId, WorkerId};
163pub use info::{
164    ConnectionInfo, Endpoint, GateTag, Lane, Origin, PeerCredentials, PipeName, Transport, UnixPath,
165};
166#[cfg(unix)]
167pub use inherit::fastcgi_listener;
168pub use inherit::{InheritError, InheritedListener, systemd, systemd_named};
169pub use limits::{Limits, LimitsBuilder, MinRate, Pressure, SizeClass};
170#[cfg(all(unix, feature = "local"))]
171pub use local::{Descriptors, UnixConnection, UnixListener};
172#[cfg(feature = "local")]
173pub use local::{LocalConnection, LocalListener};
174#[cfg(all(windows, feature = "local"))]
175pub use local_windows::{UnixConnection, UnixListener};
176pub use opts::{Keepalive, Linger, ListenOptions, LocalOptions, SocketOptions, TcpOptions};
177#[cfg(all(windows, feature = "local"))]
178pub use pipe::{PipeConnection, PipeListener, PipeOptions, PipeSecurity};
179pub use proxy::{ProxyError, ProxyFacts, ProxyMode, ProxyVersions};
180#[cfg(feature = "proxy")]
181pub use proxy_parse::{
182    ProxyAddress, ProxyCommand, ProxyHeader, ProxyParser, ProxyProgress, ProxySsl, ProxyTransport,
183    Tlv, Tlvs,
184};
185pub use server::{
186    AdoptSpec, BlockingHandler, DoorState, DrainReport, ListenerSpec, NoGate, NoWork, Server,
187    ServerBuilder, ServerHandle, Sessions, Stats, Workload,
188};
189pub use session::{
190    Input, Output, OutputFull, Phase, Session, SessionContext, SessionFactory, Step, SwapError,
191};
192pub use tcp::{ConnectOptions, ListenerInfo, TcpConnection, TcpListener};
193pub use time::{Clock, ManualClock, MonotonicClock, monotonic_micros};
194pub use tls::{
195    NoTls, NoTlsSession, TlsAcceptor, TlsCloseKind, TlsError, TlsEvent, TlsFailure, TlsInfo,
196    TlsProgress, TlsSession,
197};
198pub use trusted::TrustedProxies;
199#[cfg(feature = "udp")]
200pub use udp::{Datagram, UdpBudget, UdpOptions, UdpSocket, UdpStats};
201
202/// Crate version string, populated by Cargo at build time.
203///
204/// ```
205/// assert!(!net_io::VERSION.is_empty());
206/// ```
207pub const VERSION: &str = env!("CARGO_PKG_VERSION");
208
209/// Binds a TCP listener with safe defaults: backlog 1,024 (capped by the OS);
210/// `SO_REUSEADDR` on Unix; `SO_EXCLUSIVEADDRUSE` on Windows; IPv6-only for IPv6
211/// addresses; close-on-exec. The first resolved address that binds wins.
212///
213/// # Errors
214///
215/// `AddrInUse`, `PermissionDenied`, `AddrNotAvailable`, `NoAddresses`, `Bind`.
216///
217/// # Examples
218///
219/// ```
220/// let listener = net_io::listen("127.0.0.1:0")?;
221/// let addr = listener.local_addr()?;
222/// println!("listening on {addr}, backlog {:?}", listener.info().backlog);
223/// # Ok::<(), net_io::NetError>(())
224/// ```
225pub fn listen<A: ToSocketAddrs>(addr: A) -> Result<TcpListener, NetError> {
226    TcpListener::bind(addr)
227}
228
229/// Connects with a total timeout of 10 s across every resolved address,
230/// `TCP_NODELAY` on, and default limits (read and write timeouts 30 s).
231///
232/// # Errors
233///
234/// `ConnectTimeout`, `ConnectionRefused`, `HostUnreachable`, `NetworkUnreachable`, `NoAddresses`.
235///
236/// # Examples
237///
238/// ```no_run
239/// let mut conn = net_io::connect("example.com:80")?;
240/// conn.send_all(b"HEAD / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n")?;
241/// let mut head = [0u8; 512];
242/// let n = conn.recv(&mut head)?;
243/// println!("{}", String::from_utf8_lossy(&head[..n]));
244/// # Ok::<(), net_io::NetError>(())
245/// ```
246pub fn connect<A: ToSocketAddrs>(addr: A) -> Result<TcpConnection, NetError> {
247    TcpConnection::connect(addr)
248}
249
250/// Starts a server on the blocking driver that runs `handler` once per
251/// connection on a pool worker, with default limits and [`AllowAll`] as the
252/// gate; the door is open on return. `Ok(())` closes the connection
253/// gracefully; `Err` closes it abortively.
254///
255/// Keep the returned `Server`: dropping it stops the server. A server facing
256/// the internet should set a [`Gate`] through [`ServerBuilder`].
257///
258/// # Errors
259///
260/// As [`listen`]; `InvalidConfig` when the worker count exceeds the descriptor limit.
261///
262/// # Examples
263///
264/// ```no_run
265/// use std::io::{BufRead, BufReader, Write};
266///
267/// // Handlers can use std::io traits: TcpConnection implements Read and Write.
268/// let server = net_io::serve("127.0.0.1:7001", |conn| {
269///     let mut line = String::new();
270///     BufReader::new(&mut *conn).read_line(&mut line)?;
271///     conn.write_all(line.to_uppercase().as_bytes())?;
272///     Ok(())
273/// })?;
274/// server.wait_for_stop();
275/// # Ok::<(), net_io::NetError>(())
276/// ```
277pub fn serve<A, F>(addr: A, handler: F) -> Result<Server, NetError>
278where
279    A: ToSocketAddrs,
280    F: Fn(&mut TcpConnection) -> Result<(), NetError> + Send + Sync + 'static,
281{
282    let server = ServerBuilder::new()
283        .listener(ListenerSpec::tcp(addr))
284        .driver(DriverKind::Blocking)
285        .handler(handler)
286        .gate(AllowAll)
287        .build()?;
288    server.open_door()?;
289    Ok(server)
290}
291
292/// Starts a server that drives sans-IO [`Session`]s, with default limits and
293/// [`AllowAll`] as the gate; the door is open on return. Any
294/// `Fn(&ConnectionInfo) -> S` is a factory.
295///
296/// # Errors
297///
298/// As [`serve`].
299///
300/// # Examples
301///
302/// ```no_run
303/// use net_io::{Input, Session, SessionContext, Step};
304///
305/// struct Echo;
306/// impl Session for Echo {
307///     fn on_data(&mut self, cx: &mut SessionContext<'_>, input: &mut Input) -> Step {
308///         let taken = cx.output().push_partial(input.bytes());
309///         input.consume(taken);
310///         Step::Write
311///     }
312/// }
313///
314/// let server = net_io::serve_sessions("127.0.0.1:7002", |_info: &net_io::ConnectionInfo| Echo)?;
315/// server.wait_for_stop();
316/// # Ok::<(), net_io::NetError>(())
317/// ```
318pub fn serve_sessions<A, F>(addr: A, factory: F) -> Result<Server, NetError>
319where
320    A: ToSocketAddrs,
321    F: SessionFactory,
322{
323    let server = ServerBuilder::new()
324        .listener(ListenerSpec::tcp(addr))
325        .sessions(factory)
326        .gate(AllowAll)
327        .build()?;
328    server.open_door()?;
329    Ok(server)
330}
331
332/// Opens a local IPC listener: on Unix, a Unix domain socket at the path `name`
333/// with mode `0600` (only the owner may connect; [`LocalOptions`] and
334/// `UnixListener::bind_with` for more). On Windows, the named pipe
335/// `\\\\.\\pipe\\<name>`, owner only, remote clients rejected.
336///
337/// # Errors
338///
339/// `PathTooLong`, `StaleSocketInUse` (a live listener owns the path, or it is
340/// not a socket), `PermissionDenied`, `NotFound`, `Unsupported`, `Os`.
341///
342/// ```no_run
343/// let listener = net_io::listen_local("/run/myapp/control.sock")?;
344/// let (mut conn, info) = listener.accept()?;
345/// println!("peer process {:?}", info.peer_credentials.and_then(|c| c.pid));
346/// conn.send_all(b"ok\n")?;
347/// # Ok::<(), net_io::NetError>(())
348/// ```
349#[cfg(feature = "local")]
350pub fn listen_local(name: &str) -> Result<LocalListener, NetError> {
351    local::listen(name)
352}
353
354/// Connects to a local IPC listener (see [`listen_local`]); waits up to 10 s
355/// when its backlog is full.
356///
357/// # Errors
358///
359/// `NotFound` (no socket, or nobody listening), `PermissionDenied`,
360/// `TimedOut { kind: Connect }`, `PathTooLong`, `Unsupported`, `Os`.
361///
362/// ```no_run
363/// let mut conn = net_io::connect_local("/run/myapp/control.sock")?;
364/// conn.send_all(b"status\n")?;
365/// # Ok::<(), net_io::NetError>(())
366/// ```
367#[cfg(feature = "local")]
368pub fn connect_local(name: &str) -> Result<LocalConnection, NetError> {
369    local::connect(name)
370}
371
372#[cfg(test)]
373mod tests {
374    use super::VERSION;
375
376    #[test]
377    fn test_version_matches_manifest() {
378        assert_eq!(VERSION, env!("CARGO_PKG_VERSION"));
379        assert_eq!(VERSION.split('.').count(), 3, "expected MAJOR.MINOR.PATCH");
380    }
381}