gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.34.13_cleaup_code

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-26 14:14:41 +01:00
1 parent ec46a96a29
commit 5c2e50ec06
87 files changed
+4890 -1338
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/server/http.rs
+10 -1
View File
@@ -153,7 +153,16 @@ fn log_protocol_error(
tls_sni: Option<String>,
) {
log_protocol_error_with_identity(
state, kind, detail, peer, local, transport, http_version, status, None, None,
state,
kind,
detail,
peer,
local,
transport,
http_version,
status,
None,
None,
tls_sni,
);
}
src/server/http1_connection.rs
-1
View File
@@ -205,4 +205,3 @@ async fn connection<I>(
},
}
}
src/server/listeners.rs
+3 -3
View File
@@ -201,12 +201,12 @@ fn h2_headers_frame<B>(request: &Request<B>) -> std::result::Result<Bytes, &'sta
((length >> 16) & 0xff) as u8,
((length >> 8) & 0xff) as u8,
(length & 0xff) as u8,
0x1,
0x5,
0x1,
0x5,
0,
0,
0,
1,
1,
]);
frame.extend_from_slice(&block);
Ok(Bytes::from(frame))
src/server/sendfile.rs
+207 -58
View File
@@ -5,11 +5,14 @@ use std::{
io,
os::fd::{AsRawFd, RawFd},
pin::Pin,
sync::{Arc, Mutex},
sync::atomic::Ordering,
sync::{Arc, Mutex},
task::{Context, Poll},
};
use tokio::{io::{AsyncRead, AsyncWrite, Interest, ReadBuf}, net::TcpStream};
use tokio::{
io::{AsyncRead, AsyncWrite, Interest, ReadBuf},
net::TcpStream,
};
const TAG_LEN: usize = 32;
const FRAME_LEN: usize = 64 * 1024;
@@ -54,24 +57,37 @@ impl SendfileSlots {
return None;
}
let mut registry = self.0.lock().unwrap_or_else(|e| e.into_inner());
if registry.slots.len() >= MAX_PENDING { return None; }
if registry.slots.len() >= MAX_PENDING {
return None;
}
let mut tag = [0u8; TAG_LEN];
tag[0] = 0;
tag[0] = 0;
tag[1..].copy_from_slice(&rand::random::<[u8; TAG_LEN - 1]>());
if registry.slots.iter().any(|slot| slot.tag == tag) { return None; }
if registry.slots.iter().any(|slot| slot.tag == tag) {
return None;
}
let mut data = vec![0u8; FRAME_LEN];
for chunk in data.chunks_exact_mut(TAG_LEN) { chunk.copy_from_slice(&tag); }
for chunk in data.chunks_exact_mut(TAG_LEN) {
chunk.copy_from_slice(&tag);
}
registry.slots.push(Slot {
tag, file: candidate.file, length: candidate.length, written: 0,
tag,
file: candidate.file,
length: candidate.length,
written: 0,
});
Some(SendfileBody {
bytes: Bytes::from(data), remaining: candidate.length, web_metrics,
bytes: Bytes::from(data),
remaining: candidate.length,
web_metrics,
})
}
fn plan(&self, buf: &[u8]) -> WritePlan {
let registry = self.0.lock().unwrap_or_else(|e| e.into_inner());
if registry.slots.is_empty() { return WritePlan::Plain(buf.len()); }
if registry.slots.is_empty() {
return WritePlan::Plain(buf.len());
}
if let Some(tag) = registry.active {
if let Some(slot) = registry.slots.iter().find(|s| s.tag == tag) {
return plan_file(slot, buf);
@@ -79,12 +95,16 @@ impl SendfileSlots {
return WritePlan::Invalid;
}
for start in 0..buf.len() {
if buf[start] != 0 { continue; }
if buf[start] != 0 {
continue;
}
for slot in &registry.slots {
let available = buf.len() - start;
let compare = available.min(TAG_LEN);
if buf[start..start + compare] == slot.tag[..compare] {
if start > 0 { return WritePlan::Plain(start); }
if start > 0 {
return WritePlan::Plain(start);
}
return plan_file(slot, buf);
}
}
@@ -107,36 +127,58 @@ impl SendfileSlots {
enum WritePlan {
Plain(usize),
File { tag: [u8; TAG_LEN], fd: RawFd, file: Arc<File>, offset: u64, count: usize },
File {
tag: [u8; TAG_LEN],
fd: RawFd,
file: Arc<File>,
offset: u64,
count: usize,
},
Invalid,
}
fn plan_file(slot: &Slot, buf: &[u8]) -> WritePlan {
let expected = (slot.length - slot.written).min(buf.len() as u64) as usize;
let matched = buf[..expected].iter().enumerate()
let matched = buf[..expected]
.iter()
.enumerate()
.take_while(|(n, byte)| **byte == slot.tag[(slot.written as usize + n) % TAG_LEN])
.count();
if matched == 0 { return WritePlan::Invalid; }
if matched == 0 {
return WritePlan::Invalid;
}
WritePlan::File {
tag: slot.tag, fd: slot.file.as_raw_fd(), file: slot.file.clone(),
offset: slot.written, count: matched,
tag: slot.tag,
fd: slot.file.as_raw_fd(),
file: slot.file.clone(),
offset: slot.written,
count: matched,
}
}
impl HttpBody for SendfileBody {
type Data = Bytes;
type Error = io::Error;
fn poll_frame(mut self: Pin<&mut Self>, _cx: &mut Context<'_>)
-> Poll<Option<Result<Frame<Bytes>, io::Error>>>
{
if self.remaining == 0 { return Poll::Ready(None); }
fn poll_frame(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Bytes>, io::Error>>> {
if self.remaining == 0 {
return Poll::Ready(None);
}
let len = self.remaining.min(FRAME_LEN as u64) as usize;
self.remaining -= len as u64;
self.web_metrics.response_bytes.fetch_add(len as u64, Ordering::Relaxed);
self.web_metrics
.response_bytes
.fetch_add(len as u64, Ordering::Relaxed);
Poll::Ready(Some(Ok(Frame::data(self.bytes.slice(0..len)))))
}
fn size_hint(&self) -> SizeHint { SizeHint::with_exact(self.remaining) }
fn is_end_stream(&self) -> bool { self.remaining == 0 }
fn size_hint(&self) -> SizeHint {
SizeHint::with_exact(self.remaining)
}
fn is_end_stream(&self) -> bool {
self.remaining == 0
}
}
pub(crate) struct SendfileIo {
@@ -146,22 +188,46 @@ pub(crate) struct SendfileIo {
impl SendfileIo {
pub fn new(stream: TcpStream) -> (Self, SendfileSlots) {
let slots = SendfileSlots::default();
(Self { stream, slots: slots.clone() }, slots)
(
Self {
stream,
slots: slots.clone(),
},
slots,
)
}
}
impl AsyncRead for SendfileIo {
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().stream).poll_read(cx, buf)
}
}
impl AsyncWrite for SendfileIo {
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
if buf.is_empty() { return Poll::Ready(Ok(0)); }
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
if buf.is_empty() {
return Poll::Ready(Ok(0));
}
match self.slots.plan(buf) {
WritePlan::Plain(count) => Pin::new(&mut self.stream).poll_write(cx, &buf[..count]),
WritePlan::Invalid => Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData,
"sendfile placeholder did not match its registered descriptor"))),
WritePlan::File { tag, fd, file, offset, count } => {
WritePlan::Invalid => Poll::Ready(Err(io::Error::new(
io::ErrorKind::InvalidData,
"sendfile placeholder did not match its registered descriptor",
))),
WritePlan::File {
tag,
fd,
file,
offset,
count,
} => {
let _keep_open = file;
let socket: RawFd = self.stream.as_raw_fd();
let mut at: libc::off_t = offset as libc::off_t;
@@ -171,13 +237,22 @@ impl AsyncWrite for SendfileIo {
Poll::Ready(Ok(())) => {
let result = self.stream.try_io(Interest::WRITABLE, || {
let n = unsafe { libc::sendfile(socket, fd, &mut at, count) };
if n < 0 { Err(io::Error::last_os_error()) }
else if n == 0 { Err(io::Error::new(io::ErrorKind::UnexpectedEof,
"static source truncated during sendfile")) }
else { Ok(n as usize) }
if n < 0 {
Err(io::Error::last_os_error())
} else if n == 0 {
Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"static source truncated during sendfile",
))
} else {
Ok(n as usize)
}
});
match result {
Ok(n) => { self.slots.commit(tag, n); Poll::Ready(Ok(n)) },
Ok(n) => {
self.slots.commit(tag, n);
Poll::Ready(Ok(n))
}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
Err(e) => Poll::Ready(Err(e)),
}
@@ -192,9 +267,22 @@ impl AsyncWrite for SendfileIo {
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.stream).poll_shutdown(cx)
}
fn is_write_vectored(&self) -> bool { self.stream.is_write_vectored() }
fn poll_write_vectored(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>]) -> Poll<io::Result<usize>> {
if self.slots.0.lock().unwrap_or_else(|e| e.into_inner()).slots.is_empty() {
fn is_write_vectored(&self) -> bool {
self.stream.is_write_vectored()
}
fn poll_write_vectored(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[io::IoSlice<'_>],
) -> Poll<io::Result<usize>> {
if self
.slots
.0
.lock()
.unwrap_or_else(|e| e.into_inner())
.slots
.is_empty()
{
return Pin::new(&mut self.stream).poll_write_vectored(cx, bufs);
}
match bufs.iter().find(|part| !part.is_empty()) {
@@ -208,8 +296,8 @@ impl AsyncWrite for SendfileIo {
mod tests {
use super::*;
use http_body::Body;
use tempfile::tempfile;
use std::io::Write;
use tempfile::tempfile;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[test]
@@ -218,19 +306,42 @@ mod tests {
file.write_all(&vec![42u8; 100_000]).unwrap();
let slots = SendfileSlots::default();
let counter = Arc::new(crate::web_server::metrics::PhpMetrics::default());
let mut body = slots.create_body(SendfileCandidate { file: Arc::new(file), length: 100_000 }, counter.clone()).unwrap();
let first = match Pin::new(&mut body).poll_frame(&mut Context::from_waker(std::task::Waker::noop())) {
Poll::Ready(Some(Ok(frame))) => frame.into_data().unwrap(), _ => panic!("frame missing"),
let mut body = slots
.create_body(
SendfileCandidate {
file: Arc::new(file),
length: 100_000,
},
counter.clone(),
)
.unwrap();
let first = match Pin::new(&mut body)
.poll_frame(&mut Context::from_waker(std::task::Waker::noop()))
{
Poll::Ready(Some(Ok(frame))) => frame.into_data().unwrap(),
_ => panic!("frame missing"),
};
assert_eq!(first.len(), FRAME_LEN);
match slots.plan(&first) { WritePlan::File { count, .. } => assert_eq!(count, FRAME_LEN), _ => panic!("missing kernel file plan") }
match slots.plan(&first) {
WritePlan::File { count, .. } => assert_eq!(count, FRAME_LEN),
_ => panic!("missing kernel file plan"),
}
let tag = slots.0.lock().unwrap().slots[0].tag;
slots.commit(tag, FRAME_LEN);
let second = match Pin::new(&mut body).poll_frame(&mut Context::from_waker(std::task::Waker::noop())) {
Poll::Ready(Some(Ok(frame))) => frame.into_data().unwrap(), _ => panic!("frame missing"),
let second = match Pin::new(&mut body)
.poll_frame(&mut Context::from_waker(std::task::Waker::noop()))
{
Poll::Ready(Some(Ok(frame))) => frame.into_data().unwrap(),
_ => panic!("frame missing"),
};
assert_eq!(second.len(), 34_464);
match slots.plan(&second) { WritePlan::File { count, offset, .. } => { assert_eq!(offset, 65_536); assert_eq!(count, 34_464); }, _ => panic!("second plan missing") }
match slots.plan(&second) {
WritePlan::File { count, offset, .. } => {
assert_eq!(offset, 65_536);
assert_eq!(count, 34_464);
}
_ => panic!("second plan missing"),
}
slots.commit(tag, second.len());
assert!(slots.0.lock().unwrap().slots.is_empty());
assert!(body.is_end_stream());
@@ -245,36 +356,74 @@ mod tests {
let (tcp, _) = listener.accept().await.unwrap();
let (mut socket, slots) = SendfileIo::new(tcp);
for round in 0..2 {
let expected = if round == 0 { vec![0, 1, 2, 3].repeat(25_000) }
else { vec![9u8; 100_003] };
let expected = if round == 0 {
vec![0, 1, 2, 3].repeat(25_000)
} else {
vec![9u8; 100_003]
};
let mut file = tempfile().unwrap();
file.write_all(&expected).unwrap();
let web = Arc::new(crate::web_server::metrics::PhpMetrics::default());
let body = slots.create_body(SendfileCandidate {
file: Arc::new(file), length: expected.len() as u64,
}, web.clone()).unwrap();
let first = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", expected.len());
let body = slots
.create_body(
SendfileCandidate {
file: Arc::new(file),
length: expected.len() as u64,
},
web.clone(),
)
.unwrap();
let first = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n",
expected.len()
);
let mut body = body;
let first_frame = http_body_util::BodyExt::frame(&mut body).await.unwrap().unwrap().into_data().unwrap();
let first_frame = http_body_util::BodyExt::frame(&mut body)
.await
.unwrap()
.unwrap()
.into_data()
.unwrap();
let mut joined = first.into_bytes();
joined.extend_from_slice(&first_frame);
socket.write_all(&joined).await.unwrap();
while let Some(frame) = http_body_util::BodyExt::frame(&mut body).await {
socket.write_all(&frame.unwrap().into_data().unwrap()).await.unwrap();
socket
.write_all(&frame.unwrap().into_data().unwrap())
.await
.unwrap();
}
assert_eq!(web.response_bytes.load(Ordering::Relaxed), expected.len() as u64);
assert_eq!(
web.response_bytes.load(Ordering::Relaxed),
expected.len() as u64
);
}
socket.shutdown().await.unwrap();
});
let mut client = tokio::net::TcpStream::connect(address).await.unwrap();
let mut actual = Vec::new();
tokio::time::timeout(std::time::Duration::from_secs(10), client.read_to_end(&mut actual)).await.unwrap().unwrap();
tokio::time::timeout(
std::time::Duration::from_secs(10),
client.read_to_end(&mut actual),
)
.await
.unwrap()
.unwrap();
server.await.unwrap();
let mut wanted = Vec::new();
for expected in [vec![0, 1, 2, 3].repeat(25_000), vec![9u8; 100_003]] {
wanted.extend_from_slice(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", expected.len()).as_bytes());
wanted.extend_from_slice(
format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n",
expected.len()
)
.as_bytes(),
);
wanted.extend_from_slice(&expected);
}
assert_eq!(actual, wanted, "kernel sendfile must never leak placeholder markers");
assert_eq!(
actual, wanted,
"kernel sendfile must never leak placeholder markers"
);
}
}