nuclei/net/mod.rs
1//! Sockets: TCP, UDP and Unix domain, as [`Handle`]s.
2//!
3//! Each type keeps std's own socket inside; the handle adds the async
4//! operations (`accept`, `connect`, `send`, `recv`, `peek`, `send_to`,
5//! `recv_from`, `peek_from`) and, for the streams, `AsyncRead` and
6//! `AsyncWrite` on the handle and on a shared reference to it, so two tasks
7//! can read and write one stream at once. A stream's `read_owned` and
8//! `write_owned` take the caller's buffer (any [`IoBuf`](crate::io::IoBuf)
9//! or [`IoBufMut`](crate::io::IoBufMut)) and give it back, so the driver
10//! reads into it and writes from it with no copy, and `read_vectored_owned`
11//! and `write_vectored_owned` do the same with several buffers in one call;
12//! a TCP stream's `write_zero_copy` asks the kernel not to copy it either,
13//! and gives it back through a [`Released`] once the kernel let go of it.
14//! `send_file` sends a range of a file to a stream.
15//!
16//! A stream ends with [`shut`](TcpStream::shut), std's shutdown made once
17//! the bytes written behind reached the kernel, or for TCP with
18//! [`abort`](TcpStream::abort), which resets the connection now.
19//! [`TcpSocket`] and [`UdpSocketBuilder`] set a socket up before it listens,
20//! connects or binds, and [`TcpKeepalive`] times a TCP connection's
21//! keepalive probes.
22//!
23//! An address is anything [`ToSocketAddrs`] takes: numbers are used at
24//! once; a name (`"localhost:80"`) is looked up on the blocking pool, so no
25//! shard waits on the resolver. [`lookup_host`] looks one up alone.
26//!
27//! ```
28//! use nuclei::net::{TcpListener, TcpStream};
29//!
30//! # nuclei::block_on(async {
31//! let listener = TcpListener::bind("127.0.0.1:0").await?;
32//! let addr = listener.get_ref().local_addr()?;
33//! let client = TcpStream::connect(addr).await?;
34//! let (server, _) = listener.accept().await?;
35//! client.send(b"hi").await?;
36//! let mut buf = [0u8; 2];
37//! assert_eq!(server.recv(&mut buf).await?, 2);
38//! # std::io::Result::Ok(())
39//! # }).unwrap();
40//! ```
41
42mod addr;
43mod close;
44pub(crate) mod opt;
45mod released;
46pub(crate) mod send_file;
47mod socket;
48mod sockopt;
49mod split;
50mod stream;
51mod tcp;
52mod tcp_options;
53mod udp;
54mod udp_socket;
55#[cfg(unix)]
56pub mod unix;
57#[cfg(unix)]
58mod unix_socket;
59
60pub use self::addr::{ToSocketAddrs, lookup_host};
61pub use self::released::Released;
62pub use self::socket::TcpSocket;
63#[cfg(any(target_os = "linux", target_os = "android"))]
64pub use self::sockopt::DeviceName;
65pub use self::split::{OwnedReadHalf, OwnedWriteHalf, ReadHalf, ReuniteError, WriteHalf};
66pub use self::tcp::TcpStreamGenerator;
67pub use self::tcp_options::TcpKeepalive;
68pub use self::udp_socket::UdpSocketBuilder;
69#[cfg(unix)]
70pub use self::unix_socket::UnixSocket;
71
72use std::io;
73
74use crate::io::Handle;
75
76/// A TCP listener: std's, with an async `accept`.
77pub type TcpListener = Handle<std::net::TcpListener>;
78/// A TCP stream: std's, with async reads and writes. A few of std's calls
79/// reached through `Deref` go around the handle (`shutdown` among them): see
80/// [`Handle`].
81pub type TcpStream = Handle<std::net::TcpStream>;
82/// A UDP socket: std's, with async sends and receives. Its
83/// [`connect`](UdpSocket::connect) makes a socket;
84/// `socket.connect(addr)` on one is std's own (see there).
85pub type UdpSocket = Handle<std::net::UdpSocket>;
86/// A Unix stream listener: std's, with an async `accept`.
87#[cfg(unix)]
88pub type UnixListener = Handle<std::os::unix::net::UnixListener>;
89/// A Unix stream: std's, with async reads and writes. A few of std's calls
90/// reached through `Deref` go around the handle (`shutdown` among them): see
91/// [`Handle`].
92#[cfg(unix)]
93pub type UnixStream = Handle<std::os::unix::net::UnixStream>;
94/// A Unix datagram socket: std's, with async sends and receives.
95#[cfg(unix)]
96pub type UnixDatagram = Handle<std::os::unix::net::UnixDatagram>;
97
98/// A port nobody listens on, so a connect to it is refused. On Linux a
99/// socket bound to it and not listening holds it, so no test running
100/// meanwhile is handed the port; macOS drops a connect to such a socket
101/// rather than refusing it, and hands out ports in turn, so there (and on
102/// Windows) the port is bound and freed again. The socket, if any, is the
103/// caller's to keep while it connects.
104#[cfg(test)]
105pub(crate) fn refused_addr() -> (Option<TcpSocket>, std::net::SocketAddr) {
106 if cfg!(any(target_os = "linux", target_os = "android")) {
107 let socket = TcpSocket::new_v4().unwrap();
108 socket.bind("127.0.0.1:0".parse().unwrap()).unwrap();
109 let addr = socket.local_addr().unwrap();
110 return (Some(socket), addr);
111 }
112 let freed = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
113 (None, freed.local_addr().unwrap())
114}
115
116/// One poll of `poll`, outside the task's budget, with no waker kept: what
117/// a `try_` call answers, `WouldBlock` for one that would wait (which still
118/// sets the driver going, so a later call finds what came).
119fn at_once<T>(
120 poll: impl FnOnce(&mut core::task::Context<'_>) -> core::task::Poll<io::Result<T>>,
121) -> io::Result<T> {
122 crate::coop::outside(|| {
123 let mut cx = core::task::Context::from_waker(core::task::Waker::noop());
124 match poll(&mut cx) {
125 core::task::Poll::Ready(result) => result,
126 core::task::Poll::Pending => Err(io::ErrorKind::WouldBlock.into()),
127 }
128 })
129}
130
131/// What `interest` asks for of `source`, once some of it is ready.
132fn poll_ready(
133 source: &std::sync::Arc<crate::io::Source>,
134 cx: &mut core::task::Context<'_>,
135 interest: crate::io::Interest,
136) -> core::task::Poll<io::Result<crate::io::Ready>> {
137 use crate::io::Ready;
138 use core::task::Poll;
139 let mut ready = Ready::EMPTY;
140 if interest.is_readable()
141 && let Poll::Ready(closed) = source.poll_read_ready(cx)
142 {
143 ready = ready | Ready::READABLE;
144 if closed? {
145 ready = ready | Ready::READ_CLOSED;
146 }
147 }
148 if interest.is_writable()
149 && let Poll::Ready(closed) = source.poll_write_ready(cx)
150 {
151 ready = ready | Ready::WRITABLE;
152 if closed? {
153 ready = ready | Ready::WRITE_CLOSED;
154 }
155 }
156 if ready.is_empty() {
157 Poll::Pending
158 } else {
159 Poll::Ready(Ok(ready))
160 }
161}
162
163/// A socket's calls of its caller's own: `try_io` and `async_io`.
164macro_rules! caller_io {
165 ($($ty:ty),* $(,)?) => {$(
166 #[cfg(unix)]
167 impl Handle<$ty> {
168 /// Runs `f`, a call of the caller's own on the socket (through
169 /// [`get_ref`](Handle::get_ref) or its descriptor), if the socket
170 /// may be ready for `interest`, and clears that readiness if `f`
171 /// answers `WouldBlock`. The side then takes its caller's calls
172 /// only, until the handle's own next read or write on it. Nothing
173 /// is reordered: bytes the handle wrote behind go out first, and
174 /// data it read ahead must be read before.
175 ///
176 /// # Errors
177 ///
178 /// `WouldBlock` when none of `interest` may be ready (or bytes
179 /// written behind are still going out), `ResourceBusy` when the
180 /// handle holds data it read ahead, and what `f` answers.
181 pub fn try_io<R>(
182 &self,
183 interest: crate::io::Interest,
184 f: impl FnOnce() -> io::Result<R>,
185 ) -> io::Result<R> {
186 crate::io::caller::try_io(self.source(), crate::io::caller::sides(interest), f)
187 }
188
189 /// Runs `f`, a call of the caller's own on the socket, each time
190 /// it may be ready for `interest`, until `f` answers other than
191 /// `WouldBlock`, clearing the readiness each time it does; as
192 /// [`try_io`](Self::try_io), bytes written behind go out first.
193 ///
194 /// # Errors
195 ///
196 /// `ResourceBusy` when the handle holds data it read ahead, and
197 /// what `f` answers.
198 pub async fn async_io<R>(
199 &self,
200 interest: crate::io::Interest,
201 f: impl FnMut() -> io::Result<R>,
202 ) -> io::Result<R> {
203 crate::io::caller::async_io(self.source(), crate::io::caller::sides(interest), f)
204 .await
205 }
206 }
207 )*};
208}
209
210caller_io!(std::net::TcpStream, std::net::UdpSocket);
211#[cfg(unix)]
212caller_io!(
213 std::os::unix::net::UnixStream,
214 std::os::unix::net::UnixDatagram
215);
216
217/// The error when every address failed (the last one's), or none was given.
218fn none_worked(last: Option<io::Error>) -> io::Error {
219 last.unwrap_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "no address was given"))
220}
221
222#[cfg(test)]
223mod tests;