nuclei/net/send_file.rs
1//! `send_file`: a range of a file sent to a stream.
2//!
3//! Where the system has a call for it (`sendfile(2)`, or `TransmitFile` on
4//! Windows), the bytes go from the page cache to the socket in the kernel
5//! and never enter user space; the
6//! call reads the file, so a page that is not cached blocks the thread that
7//! makes it, and it is made on the blocking pool, a chunk at a time, never
8//! on a shard. A full socket is waited for through the stream's own
9//! readiness. Where the system has no such call, the range is read at its
10//! offset on the blocking pool and written with `write_owned`.
11//!
12//! Why the blocking pool rather than the nonblocking flags: `sendfile(2)`
13//! takes no flag that keeps it off the disk, and a nonblocking socket only
14//! stops it at the socket.
15//!
16//! The calls run on the stream's and the file's own descriptors, each job
17//! holding a lease on both for as long as it runs: a chunk still on its way
18//! when the future is dropped keeps them open until it ends, and no call
19//! needs a descriptor of its own.
20
21use std::io;
22use std::net::TcpStream;
23#[cfg(unix)]
24use std::os::unix::net::UnixStream;
25#[cfg(windows)]
26use std::sync::Arc;
27
28use crate::fs::File;
29use crate::io::Handle;
30#[cfg(unix)]
31use crate::io::Interest;
32
33/// The most one call of the blocking pool sends, by default (see
34/// [`Builder::send_file_chunk`](crate::runtime::Builder::send_file_chunk)):
35/// what a stall on a cold page holds a pool thread for is bounded, and so is
36/// a cancelled call's bytes that still land.
37pub(crate) const SEND_FILE_CHUNK: u64 = 1 << 20;
38
39/// A range of `file` that `len` bytes from `offset` name, or the error of a
40/// range no file can have.
41fn range(offset: u64, len: u64) -> io::Result<u64> {
42 offset.checked_add(len).ok_or_else(|| {
43 io::Error::new(
44 io::ErrorKind::InvalidInput,
45 "the range of the file runs past what an offset can name",
46 )
47 })
48}
49
50/// Reads up to `buf.len()` bytes of `file` at `offset`, without moving any
51/// cursor.
52fn read_at(file: &std::fs::File, buf: &mut [u8], offset: u64) -> io::Result<usize> {
53 #[cfg(unix)]
54 return std::os::unix::fs::FileExt::read_at(file, buf, offset);
55 #[cfg(windows)]
56 return std::os::windows::fs::FileExt::seek_read(file, buf, offset);
57}
58
59macro_rules! send_file {
60 ($($ty:ty),* $(,)?) => {$(
61 impl Handle<$ty> {
62 /// Sends up to `len` bytes of `file`, from `offset`, to the
63 /// stream, and returns how many it sent: fewer than `len` when
64 /// the file ends first.
65 ///
66 /// Where the system has `sendfile(2)` (Linux, Android, macOS,
67 /// FreeBSD) or `TransmitFile` (Windows, through the completion
68 /// port) the bytes go from the page cache to the socket in the
69 /// kernel and never enter user space; elsewhere (and on a Windows
70 /// provider without `TransmitFile`) they are read at `offset` and
71 /// written, which works the same and copies. The
72 /// file's own cursor does not move. Bytes written to the stream
73 /// before the call go out first; do not write to the stream
74 /// while the call runs, or its bytes may land between the
75 /// call's.
76 ///
77 /// If the future is dropped before it completes, the stream has
78 /// got a prefix of the range, in order and whole, and a chunk of
79 /// at most the runtime's
80 /// [`send_file_chunk`](crate::runtime::Builder::send_file_chunk)
81 /// (one MiB by default) may still be on its way.
82 ///
83 /// # Errors
84 ///
85 /// An error the stream or the file met (also the error of an
86 /// earlier write behind), `InvalidInput` for a range past what
87 /// an offset can name, and `file` not being a regular file.
88 /// The bytes sent before an error are not counted.
89 pub async fn send_file(&self, file: &File, offset: u64, len: u64) -> io::Result<u64> {
90 range(offset, len)?;
91 core::future::poll_fn(|cx| self.source().poll_flush(cx)).await?;
92 if len == 0 {
93 return Ok(0);
94 }
95 let chunk = crate::rt::runtime::io_knobs::current().send_file_chunk;
96 #[cfg(unix)]
97 if crate::io::sys::SENDS_FILES {
98 return self.send_file_in_kernel(file, offset, len, chunk).await;
99 }
100 #[cfg(windows)]
101 if crate::io::sys::transmit_file_available() {
102 return self.send_file_transmitted(file, offset, len, chunk).await;
103 }
104 self.send_file_by_copy(file, offset, len, chunk).await
105 }
106
107 #[cfg(unix)]
108 async fn send_file_in_kernel(
109 &self,
110 file: &File,
111 offset: u64,
112 len: u64,
113 chunk: u64,
114 ) -> io::Result<u64> {
115 let sock = self.source().lease()?;
116 let from = file.source().lease()?;
117 let mut sent = 0u64;
118 while sent < len {
119 let want = (len - sent).min(chunk) as usize;
120 let at = offset + sent;
121 let (out, src) = (sock.clone(), from.clone());
122 // The socket's `WouldBlock` is an answer, not a failure
123 // to count: it rides out inside the `Ok`.
124 let step = crate::rt::blocking::call(move || {
125 Ok(crate::io::sys::send_file(out.raw(), src.raw(), at, want))
126 })
127 .await?;
128 match step {
129 Ok(0) => break,
130 Ok(n) => {
131 crate::rt::counters::tally::wrote(n);
132 sent += n as u64;
133 }
134 Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
135 let raw = sock.raw();
136 self.async_io(Interest::WRITABLE, || crate::io::sys::poll_writable(raw))
137 .await?;
138 }
139 Err(err) => {
140 crate::rt::counters::io_error(&err);
141 return Err(err);
142 }
143 }
144 }
145 Ok(sent)
146 }
147
148 /// `TransmitFile` on the completion port, a chunk at a time. The range is cut to
149 /// the file first, as the call sends exactly what it is asked for or fails.
150 #[cfg(windows)]
151 async fn send_file_transmitted(
152 &self,
153 file: &File,
154 offset: u64,
155 len: u64,
156 chunk: u64,
157 ) -> io::Result<u64> {
158 use std::os::windows::io::AsRawHandle;
159 let raw = file.get_ref().as_raw_handle() as usize as crate::io::raw::Raw;
160 // A file whose sharing forbids another reader is sent by copy, which reads at
161 // the offsets through the handle itself.
162 let Ok(handle) = crate::io::sys::reopen_overlapped(raw) else {
163 return self.send_file_by_copy(file, offset, len, chunk).await;
164 };
165 let handle = Arc::new(handle);
166 let len = len.min(file.get_ref().metadata()?.len().saturating_sub(offset));
167 let mut sent = 0u64;
168 while sent < len {
169 let want = (len - sent).min(chunk) as u32;
170 let wrote = self
171 .source()
172 .send_file_range(Arc::clone(&handle), offset + sent, want)
173 .await?;
174 sent += wrote as u64;
175 }
176 Ok(sent)
177 }
178
179 async fn send_file_by_copy(
180 &self,
181 file: &File,
182 offset: u64,
183 len: u64,
184 chunk: u64,
185 ) -> io::Result<u64> {
186 let from = file.source().lease()?;
187 let mut buf = vec![0u8; chunk.min(len) as usize];
188 let mut sent = 0u64;
189 while sent < len {
190 let want = (len - sent).min(buf.len() as u64) as usize;
191 let at = offset + sent;
192 let from = from.clone();
193 let (read, back) = crate::rt::blocking::call(move || {
194 let read = read_at(&from.file(), &mut buf[..want], at);
195 Ok((read, buf))
196 })
197 .await?;
198 buf = back;
199 let n = read?;
200 crate::rt::counters::tally::read(n);
201 if n == 0 {
202 break;
203 }
204 buf.truncate(n);
205 let (wrote, back) = self.write_owned(buf).await;
206 buf = back;
207 wrote?;
208 buf.resize(chunk.min(len) as usize, 0);
209 sent += n as u64;
210 }
211 Ok(sent)
212 }
213 }
214 )*};
215}
216
217send_file!(TcpStream);
218#[cfg(unix)]
219send_file!(UnixStream);