berkus
/
actix-web
mirror of https://github.com/fafhrd91/actix-web
1
0
Fork 0
Code Issues Releases Wiki Activity

remove trace logs

This commit is contained in:
Rob Ede 2022-01-31 12:13:53 +00:00
parent dfb3faa990
commit 2c2ebb18f6
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
9 changed files with 120 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
actix-http/examples/bench.rs
View File

@ -4,13 +4,16 @@ use actix_http::{HttpService, Request, Response, StatusCode};
use actix_server::Server; use actix_server::Server;
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
static STR: Lazy<String> = Lazy::new(|| "HELLO WORLD ".repeat(100)); static STR: Lazy<String> = Lazy::new(|| "HELLO WORLD ".repeat(20));
#[actix_rt::main] #[actix_rt::main]
async fn main() -> io::Result<()> { async fn main() -> io::Result<()> {
env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
Server::build() Server::build()
.bind("dispatcher-benchmark", ("127.0.0.1", 8080), || { .bind("dispatcher-benchmark", ("127.0.0.1", 8080), || {
HttpService::build() HttpService::build()
.client_timeout(1000)
.finish(|_: Request| async move { .finish(|_: Request| async move {
let mut res = Response::build(StatusCode::OK); let mut res = Response::build(StatusCode::OK);
Ok::<_, Infallible>(res.body(&**STR)) Ok::<_, Infallible>(res.body(&**STR))

25
actix-http/examples/h2spec.rs Normal file
View File

@ -0,0 +1,25 @@
use std::{convert::Infallible, io};
use actix_http::{HttpService, Request, Response, StatusCode};
use actix_server::Server;
use once_cell::sync::Lazy;
static STR: Lazy<String> = Lazy::new(|| "HELLO WORLD ".repeat(100));
#[actix_rt::main]
async fn main() -> io::Result<()> {
env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
Server::build()
.bind("h2spec", ("127.0.0.1", 8080), || {
HttpService::build()
.h2(|_: Request| async move {
let mut res = Response::build(StatusCode::OK);
Ok::<_, Infallible>(res.body(&**STR))
})
.tcp()
})?
.workers(4)
.run()
.await
}

12
actix-http/src/h1/codec.rs
View File

@ -16,7 +16,7 @@ use crate::{
bitflags! { bitflags! {
struct Flags: u8 { struct Flags: u8 {
const HEAD = 0b0000_0001; const HEAD = 0b0000_0001;
const KEEPALIVE_ENABLED = 0b0000_0010; const KEEP_ALIVE_ENABLED = 0b0000_0010;
const STREAM = 0b0000_0100; const STREAM = 0b0000_0100;
} }
} }
@ -52,7 +52,7 @@ impl Codec {
/// `keepalive_enabled` how response `connection` header get generated. /// `keepalive_enabled` how response `connection` header get generated.
pub fn new(config: ServiceConfig) -> Self { pub fn new(config: ServiceConfig) -> Self {
let flags = if config.keep_alive_enabled() { let flags = if config.keep_alive_enabled() {
Flags::KEEPALIVE_ENABLED Flags::KEEP_ALIVE_ENABLED
} else { } else {
Flags::empty() Flags::empty()
}; };
@ -76,14 +76,14 @@ impl Codec {
/// Check if last response is keep-alive. /// Check if last response is keep-alive.
#[inline] #[inline]
pub fn keepalive(&self) -> bool { pub fn keep_alive(&self) -> bool {
self.conn_type == ConnectionType::KeepAlive self.conn_type == ConnectionType::KeepAlive
} }
/// Check if keep-alive enabled on server level. /// Check if keep-alive enabled on server level.
#[inline] #[inline]
pub fn keepalive_enabled(&self) -> bool { pub fn keep_alive_enabled(&self) -> bool {
self.flags.contains(Flags::KEEPALIVE_ENABLED) self.flags.contains(Flags::KEEP_ALIVE_ENABLED)
} }
/// Check last request's message type. /// Check last request's message type.
@ -124,7 +124,7 @@ impl Decoder for Codec {
self.version = head.version; self.version = head.version;
self.conn_type = head.connection_type(); self.conn_type = head.connection_type();
if self.conn_type == ConnectionType::KeepAlive if self.conn_type == ConnectionType::KeepAlive
&& !self.flags.contains(Flags::KEEPALIVE_ENABLED) && !self.flags.contains(Flags::KEEP_ALIVE_ENABLED)
{ {
self.conn_type = ConnectionType::Close self.conn_type = ConnectionType::Close
} }

149
actix-http/src/h1/dispatcher.rs
View File

@ -299,8 +299,6 @@ where
U::Error: fmt::Display, U::Error: fmt::Display,
{ {
fn can_read(&self, cx: &mut Context<'_>) -> bool { fn can_read(&self, cx: &mut Context<'_>) -> bool {
log::trace!("enter InnerDispatcher::can_read");
if self.flags.contains(Flags::READ_DISCONNECT) { if self.flags.contains(Flags::READ_DISCONNECT) {
false false
} else if let Some(ref info) = self.payload { } else if let Some(ref info) = self.payload {
@ -310,10 +308,7 @@ where
} }
} }
/// If checked is set to true, delay disconnect until all tasks have finished.
fn client_disconnected(self: Pin<&mut Self>) { fn client_disconnected(self: Pin<&mut Self>) {
log::trace!("enter InnerDispatcher::client_disconnect");
let this = self.project(); let this = self.project();
this.flags this.flags
@ -325,8 +320,6 @@ where
} }
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> { fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
log::trace!("enter InnerDispatcher::poll_flush");
let InnerDispatcherProj { io, write_buf, .. } = self.project(); let InnerDispatcherProj { io, write_buf, .. } = self.project();
let mut io = Pin::new(io.as_mut().unwrap()); let mut io = Pin::new(io.as_mut().unwrap());
@ -336,17 +329,13 @@ where
while written < len { while written < len {
match io.as_mut().poll_write(cx, &write_buf[written..])? { match io.as_mut().poll_write(cx, &write_buf[written..])? {
Poll::Ready(0) => { Poll::Ready(0) => {
log::trace!("write zero error"); log::error!("write zero; closing");
return Poll::Ready(Err(io::Error::new(io::ErrorKind::WriteZero, ""))); return Poll::Ready(Err(io::Error::new(io::ErrorKind::WriteZero, "")));
} }
Poll::Ready(n) => { Poll::Ready(n) => written += n,
log::trace!(" written {} bytes", n);
written += n
}
Poll::Pending => { Poll::Pending => {
log::trace!(" pending");
write_buf.advance(written); write_buf.advance(written);
return Poll::Pending; return Poll::Pending;
} }
@ -365,8 +354,6 @@ where
res: Response<()>, res: Response<()>,
body: &impl MessageBody, body: &impl MessageBody,
) -> Result<BodySize, DispatchError> { ) -> Result<BodySize, DispatchError> {
log::trace!("enter InnerDispatcher::send_response_inner");
let this = self.project(); let this = self.project();
let size = body.size(); let size = body.size();
@ -381,13 +368,7 @@ where
DispatchError::Io(err) DispatchError::Io(err)
})?; })?;
let conn_keep_alive = this.codec.keepalive(); this.flags.set(Flags::KEEP_ALIVE, this.codec.keep_alive());
this.flags.set(Flags::KEEP_ALIVE, conn_keep_alive);
if !conn_keep_alive {
log::trace!("clearing keep-alive timer");
this.ka_timer.clear(line!());
}
Ok(size) Ok(size)
} }
@ -397,8 +378,6 @@ where
res: Response<()>, res: Response<()>,
body: B, body: B,
) -> Result<(), DispatchError> { ) -> Result<(), DispatchError> {
log::trace!("enter InnerDispatcher::send_response");
let size = self.as_mut().send_response_inner(res, &body)?; let size = self.as_mut().send_response_inner(res, &body)?;
let mut this = self.project(); let mut this = self.project();
this.state.set(match size { this.state.set(match size {
@ -417,8 +396,6 @@ where
res: Response<()>, res: Response<()>,
body: BoxBody, body: BoxBody,
) -> Result<(), DispatchError> { ) -> Result<(), DispatchError> {
log::trace!("enter InnerDispatcher::send_error_response");
let size = self.as_mut().send_response_inner(res, &body)?; let size = self.as_mut().send_response_inner(res, &body)?;
let mut this = self.project(); let mut this = self.project();
this.state.set(match size { this.state.set(match size {
@ -433,8 +410,6 @@ where
} }
fn send_continue(self: Pin<&mut Self>) { fn send_continue(self: Pin<&mut Self>) {
log::trace!("enter InnerDispatcher::send_continue");
self.project() self.project()
.write_buf .write_buf
.extend_from_slice(b"HTTP/1.1 100 Continue\r\n\r\n"); .extend_from_slice(b"HTTP/1.1 100 Continue\r\n\r\n");
@ -445,8 +420,6 @@ where
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Result<PollResponse, DispatchError> { ) -> Result<PollResponse, DispatchError> {
'res: loop { 'res: loop {
log::trace!("enter InnerDispatcher::poll_response loop iteration");
let mut this = self.as_mut().project(); let mut this = self.as_mut().project();
match this.state.as_mut().project() { match this.state.as_mut().project() {
// no future is in InnerDispatcher state; pop next message // no future is in InnerDispatcher state; pop next message
@ -455,12 +428,10 @@ where
Some(DispatcherMessage::Item(req)) => { Some(DispatcherMessage::Item(req)) => {
// Handle `EXPECT: 100-Continue` header // Handle `EXPECT: 100-Continue` header
if req.head().expect() { if req.head().expect() {
log::trace!(" passing request to expect handler");
// set InnerDispatcher state and continue loop to poll it // set InnerDispatcher state and continue loop to poll it
let fut = this.flow.expect.call(req); let fut = this.flow.expect.call(req);
this.state.set(State::ExpectCall { fut }); this.state.set(State::ExpectCall { fut });
} else { } else {
log::trace!(" passing request to service handler");
// set InnerDispatcher state and continue loop to poll it // set InnerDispatcher state and continue loop to poll it
let fut = this.flow.service.call(req); let fut = this.flow.service.call(req);
this.state.set(State::ServiceCall { fut }); this.state.set(State::ServiceCall { fut });
@ -469,7 +440,6 @@ where
// handle error message // handle error message
Some(DispatcherMessage::Error(res)) => { Some(DispatcherMessage::Error(res)) => {
log::trace!(" handling dispatcher error message");
// send_response would update InnerDispatcher state to SendPayload or None // send_response would update InnerDispatcher state to SendPayload or None
// (If response body is empty) // (If response body is empty)
// continue loop to poll it // continue loop to poll it
@ -478,31 +448,23 @@ where
// return with upgrade request and poll it exclusively // return with upgrade request and poll it exclusively
Some(DispatcherMessage::Upgrade(req)) => { Some(DispatcherMessage::Upgrade(req)) => {
// return upgrade return Ok(PollResponse::Upgrade(req))
return Ok(PollResponse::Upgrade(req));
} }
// all messages are dealt with // all messages are dealt with
None => { None => return Ok(PollResponse::DoNothing),
log::trace!("all messages handled");
return Ok(PollResponse::DoNothing);
}
}, },
StateProj::ServiceCall { fut } => { StateProj::ServiceCall { fut } => {
log::trace!(" calling request handler service");
match fut.poll(cx) { match fut.poll(cx) {
// service call resolved. send response. // service call resolved. send response.
Poll::Ready(Ok(res)) => { Poll::Ready(Ok(res)) => {
log::trace!(" ok");
let (res, body) = res.into().replace_body(()); let (res, body) = res.into().replace_body(());
self.as_mut().send_response(res, body)?; self.as_mut().send_response(res, body)?;
} }
// send service call error as response // send service call error as response
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
log::trace!(" error");
let res: Response<BoxBody> = err.into(); let res: Response<BoxBody> = err.into();
let (res, body) = res.replace_body(()); let (res, body) = res.replace_body(());
self.as_mut().send_error_response(res, body)?; self.as_mut().send_error_response(res, body)?;
@ -511,7 +473,6 @@ where
// service call pending and could be waiting for more chunk messages // service call pending and could be waiting for more chunk messages
// (pipeline message limit and/or payload can_read limit) // (pipeline message limit and/or payload can_read limit)
Poll::Pending => { Poll::Pending => {
log::trace!(" pending");
// no new message is decoded and no new payload is fed // no new message is decoded and no new payload is fed
// nothing to do except waiting for new incoming data from client // nothing to do except waiting for new incoming data from client
if !self.as_mut().poll_request(cx)? { if !self.as_mut().poll_request(cx)? {
@ -523,8 +484,6 @@ where
} }
StateProj::SendPayload { mut body } => { StateProj::SendPayload { mut body } => {
log::trace!("sending payload");
// keep populate writer buffer until buffer size limit hit, // keep populate writer buffer until buffer size limit hit,
// get blocked or finished. // get blocked or finished.
while this.write_buf.len() < super::payload::MAX_BUFFER_SIZE { while this.write_buf.len() < super::payload::MAX_BUFFER_SIZE {
@ -553,14 +512,13 @@ where
Poll::Pending => return Ok(PollResponse::DoNothing), Poll::Pending => return Ok(PollResponse::DoNothing),
} }
} }
// buffer is beyond max size.
// return and try to write the whole buffer to io stream. // buffer is beyond max size
// return and try to write the whole buffer to I/O stream.
return Ok(PollResponse::DrainWriteBuf); return Ok(PollResponse::DrainWriteBuf);
} }
StateProj::SendErrorPayload { mut body } => { StateProj::SendErrorPayload { mut body } => {
log::trace!(" sending error payload");
// TODO: de-dupe impl with SendPayload // TODO: de-dupe impl with SendPayload
// keep populate writer buffer until buffer size limit hit, // keep populate writer buffer until buffer size limit hit,
@ -713,13 +671,11 @@ where
/// Process one incoming request. /// Process one incoming request.
/// ///
/// Boolean in return type indicates if any meaningful work was done. /// Returns true if any meaningful work was done.
fn poll_request( fn poll_request(
mut self: Pin<&mut Self>, mut self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Result<bool, DispatchError> { ) -> Result<bool, DispatchError> {
log::trace!("enter InnerDispatcher::poll_request");
let pipeline_queue_full = self.messages.len() >= MAX_PIPELINED_MESSAGES; let pipeline_queue_full = self.messages.len() >= MAX_PIPELINED_MESSAGES;
let can_not_read = !self.can_read(cx); let can_not_read = !self.can_read(cx);
@ -733,16 +689,13 @@ where
let mut updated = false; let mut updated = false;
loop { loop {
log::trace!("attempt to decode frame");
match this.codec.decode(this.read_buf) { match this.codec.decode(this.read_buf) {
Ok(Some(msg)) => { Ok(Some(msg)) => {
updated = true; updated = true;
log::trace!("found full frame (head)");
match msg { match msg {
Message::Item(mut req) => { Message::Item(mut req) => {
// head timer only applies to first request on connection
this.head_timer.clear(line!()); this.head_timer.clear(line!());
req.head_mut().peer_addr = *this.peer_addr; req.head_mut().peer_addr = *this.peer_addr;
@ -815,10 +768,7 @@ where
// decode is partial and buffer is not full yet // decode is partial and buffer is not full yet
// break and wait for more read // break and wait for more read
Ok(None) => { Ok(None) => break,
log::trace!("found partial frame");
break;
}
Err(ParseError::Io(err)) => { Err(ParseError::Io(err)) => {
log::trace!("io error: {}", &err); log::trace!("io error: {}", &err);
@ -925,7 +875,6 @@ where
if let Some(deadline) = this.config.client_disconnect_deadline() { if let Some(deadline) = this.config.client_disconnect_deadline() {
// start shutdown timeout if enabled // start shutdown timeout if enabled
log::trace!("starting disconnect timer");
this.shutdown_timer this.shutdown_timer
.set_and_init(cx, sleep_until(deadline), line!()); .set_and_init(cx, sleep_until(deadline), line!());
} else { } else {
@ -960,10 +909,10 @@ where
} }
/// Poll head, keep-alive, and disconnect timer. /// Poll head, keep-alive, and disconnect timer.
fn poll_timer(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Result<(), DispatchError> { fn poll_timers(
log::trace!("enter InnerDispatcher::poll_timer"); mut self: Pin<&mut Self>,
trace_timer_states(&self.head_timer, &self.ka_timer, &self.shutdown_timer); cx: &mut Context<'_>,
) -> Result<(), DispatchError> {
self.as_mut().poll_head_timer(cx)?; self.as_mut().poll_head_timer(cx)?;
self.as_mut().poll_ka_timer(cx)?; self.as_mut().poll_ka_timer(cx)?;
self.as_mut().poll_shutdown_timer(cx)?; self.as_mut().poll_shutdown_timer(cx)?;
@ -981,13 +930,9 @@ where
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Result<bool, DispatchError> { ) -> Result<bool, DispatchError> {
log::trace!("enter InnerDispatcher::read_available");
log::trace!(" reading from a {}", core::any::type_name::<T>());
let this = self.project(); let this = self.project();
if this.flags.contains(Flags::READ_DISCONNECT) { if this.flags.contains(Flags::READ_DISCONNECT) {
log::trace!(" read DC");
return Ok(false); return Ok(false);
}; };
@ -1043,12 +988,9 @@ where
match actix_codec::poll_read_buf(io.as_mut(), cx, this.read_buf) { match actix_codec::poll_read_buf(io.as_mut(), cx, this.read_buf) {
Poll::Ready(Ok(n)) => { Poll::Ready(Ok(n)) => {
log::trace!(" read {} bytes", n);
this.flags.remove(Flags::FINISHED); this.flags.remove(Flags::FINISHED);
if n == 0 { if n == 0 {
log::trace!(" signalling should_disconnect");
return Ok(true); return Ok(true);
} }
@ -1056,13 +998,10 @@ where
} }
Poll::Pending => { Poll::Pending => {
log::trace!(" read pending");
return Ok(false); return Ok(false);
} }
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
log::trace!(" read err: {:?}", &err);
return match err.kind() { return match err.kind() {
// convert WouldBlock error to the same as Pending return // convert WouldBlock error to the same as Pending return
io::ErrorKind::WouldBlock => Ok(false), io::ErrorKind::WouldBlock => Ok(false),
@ -1111,9 +1050,6 @@ where
#[inline] #[inline]
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
log::trace!(target: "actix*", "");
log::trace!("enter Dispatcher::poll");
let this = self.as_mut().project(); let this = self.as_mut().project();
#[cfg(test)] #[cfg(test)]
@ -1122,10 +1058,22 @@ where
} }
match this.inner.project() { match this.inner.project() {
DispatcherStateProj::Normal { mut inner } => { DispatcherStateProj::Upgrade { fut: upgrade } => upgrade.poll(cx).map_err(|err| {
log::trace!("current flags: {:?}", &inner.flags); log::error!("Upgrade handler error: {}", err);
DispatchError::Upgrade
}),
inner.as_mut().poll_timer(cx)?; DispatcherStateProj::Normal { mut inner } => {
log::trace!("start flags: {:?}", &inner.flags);
trace_timer_states(
"start",
&inner.head_timer,
&inner.ka_timer,
&inner.shutdown_timer,
);
inner.as_mut().poll_timers(cx)?;
let poll = if inner.flags.contains(Flags::SHUTDOWN) { let poll = if inner.flags.contains(Flags::SHUTDOWN) {
if inner.flags.contains(Flags::WRITE_DISCONNECT) { if inner.flags.contains(Flags::WRITE_DISCONNECT) {
@ -1141,12 +1089,15 @@ where
// read from I/O stream and fill read buffer // read from I/O stream and fill read buffer
let should_disconnect = inner.as_mut().read_available(cx)?; let should_disconnect = inner.as_mut().read_available(cx)?;
// after reading something from stream, clear keep-alive timer
if !inner.read_buf.is_empty() && inner.flags.contains(Flags::KEEP_ALIVE) {
inner.as_mut().project().ka_timer.clear(line!());
}
if !inner.flags.contains(Flags::STARTED) { if !inner.flags.contains(Flags::STARTED) {
log::trace!("set started flag");
inner.as_mut().project().flags.insert(Flags::STARTED); inner.as_mut().project().flags.insert(Flags::STARTED);
if let Some(deadline) = inner.config.client_request_deadline() { if let Some(deadline) = inner.config.client_request_deadline() {
log::trace!("start head timer");
inner.as_mut().project().head_timer.set_and_init( inner.as_mut().project().head_timer.set_and_init(
cx, cx,
sleep_until(deadline), sleep_until(deadline),
@ -1158,7 +1109,6 @@ where
inner.as_mut().poll_request(cx)?; inner.as_mut().poll_request(cx)?;
if should_disconnect { if should_disconnect {
log::trace!("should_disconnect = true");
// I/O stream should to be closed // I/O stream should to be closed
let inner = inner.as_mut().project(); let inner = inner.as_mut().project();
inner.flags.insert(Flags::READ_DISCONNECT); inner.flags.insert(Flags::READ_DISCONNECT);
@ -1175,11 +1125,10 @@ where
PollResponse::DoNothing => { PollResponse::DoNothing => {
if inner.flags.contains(Flags::FINISHED | Flags::KEEP_ALIVE) { if inner.flags.contains(Flags::FINISHED | Flags::KEEP_ALIVE) {
if let Some(deadline) = inner.config.keep_alive_timer() { if let Some(timer) = inner.config.keep_alive_timer() {
log::trace!("setting keep-alive timer");
inner.as_mut().project().ka_timer.set_and_init( inner.as_mut().project().ka_timer.set_and_init(
cx, cx,
deadline, timer,
line!(), line!(),
); );
} }
@ -1206,7 +1155,6 @@ where
// TODO: what? is WouldBlock good or bad? // TODO: what? is WouldBlock good or bad?
// want to find a reference for this macOS behavior // want to find a reference for this macOS behavior
if inner.as_mut().poll_flush(cx)?.is_pending() || !drain { if inner.as_mut().poll_flush(cx)?.is_pending() || !drain {
log::trace!("break out of poll_response loop after poll_flush");
break; break;
} }
} }
@ -1228,10 +1176,8 @@ where
// keep-alive and stream errors // keep-alive and stream errors
if state_is_none && inner_p.write_buf.is_empty() { if state_is_none && inner_p.write_buf.is_empty() {
log::trace!("state is None and write buf is empty");
if let Some(err) = inner_p.error.take() { if let Some(err) = inner_p.error.take() {
log::trace!("stream error {}", &err); log::error!("stream error: {}", &err);
return Poll::Ready(Err(err)); return Poll::Ready(Err(err));
} }
@ -1239,10 +1185,6 @@ where
if inner_p.flags.contains(Flags::FINISHED) if inner_p.flags.contains(Flags::FINISHED)
&& !inner_p.flags.contains(Flags::KEEP_ALIVE) && !inner_p.flags.contains(Flags::KEEP_ALIVE)
{ {
log::trace!(
"start shutdown because keep-alive is disabled or opted \
out for this connection"
);
inner_p.flags.remove(Flags::FINISHED); inner_p.flags.remove(Flags::FINISHED);
inner_p.flags.insert(Flags::SHUTDOWN); inner_p.flags.insert(Flags::SHUTDOWN);
return self.poll(cx); return self.poll(cx);
@ -1250,14 +1192,12 @@ where
// disconnect if shutdown // disconnect if shutdown
if inner_p.flags.contains(Flags::SHUTDOWN) { if inner_p.flags.contains(Flags::SHUTDOWN) {
log::trace!("shutdown from shutdown flag");
return self.poll(cx); return self.poll(cx);
} }
} }
log::trace!("dispatcher going to sleep; wait for next event");
trace_timer_states( trace_timer_states(
"end",
inner_p.head_timer, inner_p.head_timer,
inner_p.ka_timer, inner_p.ka_timer,
inner_p.shutdown_timer, inner_p.shutdown_timer,
@ -1266,25 +1206,22 @@ where
Poll::Pending Poll::Pending
}; };
log::trace!("current flags: {:?}", &inner.flags); log::trace!("end flags: {:?}", &inner.flags);
poll poll
} }
DispatcherStateProj::Upgrade { fut: upgrade } => upgrade.poll(cx).map_err(|err| {
log::error!("Upgrade handler error: {}", err);
DispatchError::Upgrade
}),
} }
} }
} }
#[allow(dead_code)]
fn trace_timer_states( fn trace_timer_states(
label: &str,
head_timer: &TimerState, head_timer: &TimerState,
ka_timer: &TimerState, ka_timer: &TimerState,
shutdown_timer: &TimerState, shutdown_timer: &TimerState,
) { ) {
log::trace!("timers:"); log::trace!("{} timers:", label);
if head_timer.is_enabled() { if head_timer.is_enabled() {
log::trace!(" head {}", &head_timer); log::trace!(" head {}", &head_timer);

10
actix-http/src/h1/timer.rs
View File

@ -23,8 +23,8 @@ impl TimerState {
} }
pub(super) fn set(&mut self, timer: Sleep, line: u32) { pub(super) fn set(&mut self, timer: Sleep, line: u32) {
if !self.is_enabled() { if matches!(self, Self::Disabled) {
log::warn!("setting disabled timer from line {}", line); log::trace!("setting disabled timer from line {}", line);
} }
*self = Self::Active { *self = Self::Active {
@ -38,12 +38,12 @@ impl TimerState {
} }
pub(super) fn clear(&mut self, line: u32) { pub(super) fn clear(&mut self, line: u32) {
if !self.is_enabled() { if matches!(self, Self::Disabled) {
log::warn!("trying to clear a disabled timer from line {}", line); log::trace!("trying to clear a disabled timer from line {}", line);
} }
if matches!(self, Self::Inactive) { if matches!(self, Self::Inactive) {
log::warn!("trying to clear an inactive timer from line {}", line); log::trace!("trying to clear an inactive timer from line {}", line);
} }
*self = Self::Inactive; *self = Self::Inactive;

10
actix-http/src/h2/mod.rs
View File

@ -15,17 +15,17 @@ use h2::{
RecvStream, RecvStream,
}; };
use crate::{
config::ServiceConfig,
error::{DispatchError, PayloadError},
};
mod dispatcher; mod dispatcher;
mod service; mod service;
pub use self::dispatcher::Dispatcher; pub use self::dispatcher::Dispatcher;
pub use self::service::H2Service; pub use self::service::H2Service;
use crate::{
config::ServiceConfig,
error::{DispatchError, PayloadError},
};
/// HTTP/2 peer stream. /// HTTP/2 peer stream.
pub struct Payload { pub struct Payload {
stream: RecvStream, stream: RecvStream,

8
actix-http/tests/test_client.rs
View File

@ -31,7 +31,7 @@ const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
Hello World Hello World Hello World Hello World Hello World"; Hello World Hello World Hello World Hello World Hello World";
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_v2() { async fn h1_v2() {
let srv = test_server(move || { let srv = test_server(move || {
HttpService::build() HttpService::build()
.finish(|_| future::ok::<_, Infallible>(Response::ok().set_body(STR))) .finish(|_| future::ok::<_, Infallible>(Response::ok().set_body(STR)))
@ -59,7 +59,7 @@ async fn test_h1_v2() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_connection_close() { async fn connection_close() {
let srv = test_server(move || { let srv = test_server(move || {
HttpService::build() HttpService::build()
.finish(|_| future::ok::<_, Infallible>(Response::ok().set_body(STR))) .finish(|_| future::ok::<_, Infallible>(Response::ok().set_body(STR)))
@ -73,7 +73,7 @@ async fn test_connection_close() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_with_query_parameter() { async fn with_query_parameter() {
let srv = test_server(move || { let srv = test_server(move || {
HttpService::build() HttpService::build()
.finish(|req: Request| async move { .finish(|req: Request| async move {
@ -104,7 +104,7 @@ impl From<ExpectFailed> for Response<BoxBody> {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_expect() { async fn h1_expect() {
let srv = test_server(move || { let srv = test_server(move || {
HttpService::build() HttpService::build()
.expect(|req: Request| async { .expect(|req: Request| async {

50
actix-http/tests/test_server.rs
View File

@ -22,7 +22,7 @@ use futures_util::{
use regex::Regex; use regex::Regex;
#[actix_rt::test] #[actix_rt::test]
async fn test_h1() { async fn h1_basic() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.keep_alive(KeepAlive::Disabled) .keep_alive(KeepAlive::Disabled)
@ -43,7 +43,7 @@ async fn test_h1() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_2() { async fn h1_2() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.keep_alive(KeepAlive::Disabled) .keep_alive(KeepAlive::Disabled)
@ -75,7 +75,7 @@ impl From<ExpectFailed> for Response<BoxBody> {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_expect_continue() { async fn expect_continue() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.expect(fn_service(|req: Request| { .expect(fn_service(|req: Request| {
@ -106,7 +106,7 @@ async fn test_expect_continue() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_expect_continue_h1() { async fn expect_continue_h1() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.expect(fn_service(|req: Request| { .expect(fn_service(|req: Request| {
@ -139,7 +139,7 @@ async fn test_expect_continue_h1() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_chunked_payload() { async fn chunked_payload() {
let chunk_sizes = vec![32768, 32, 32768]; let chunk_sizes = vec![32768, 32, 32768];
let total_size: usize = chunk_sizes.iter().sum(); let total_size: usize = chunk_sizes.iter().sum();
@ -233,7 +233,7 @@ async fn slow_request_408() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http1_malformed_request() { async fn http1_malformed_request() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok())) .h1(|_| ok::<_, Infallible>(Response::ok()))
@ -251,7 +251,7 @@ async fn test_http1_malformed_request() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http1_keepalive() { async fn http1_keepalive() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok())) .h1(|_| ok::<_, Infallible>(Response::ok()))
@ -274,7 +274,7 @@ async fn test_http1_keepalive() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http1_keepalive_timeout() { async fn http1_keepalive_timeout() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.keep_alive(1) .keep_alive(1)
@ -300,7 +300,7 @@ async fn test_http1_keepalive_timeout() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http1_keepalive_close() { async fn http1_keepalive_close() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok())) .h1(|_| ok::<_, Infallible>(Response::ok()))
@ -322,7 +322,7 @@ async fn test_http1_keepalive_close() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http10_keepalive_default_close() { async fn http10_keepalive_default_close() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok())) .h1(|_| ok::<_, Infallible>(Response::ok()))
@ -344,7 +344,7 @@ async fn test_http10_keepalive_default_close() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http10_keepalive() { async fn http10_keepalive() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok())) .h1(|_| ok::<_, Infallible>(Response::ok()))
@ -373,7 +373,7 @@ async fn test_http10_keepalive() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_http1_keepalive_disabled() { async fn http1_keepalive_disabled() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.keep_alive(KeepAlive::Disabled) .keep_alive(KeepAlive::Disabled)
@ -396,7 +396,7 @@ async fn test_http1_keepalive_disabled() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_content_length() { async fn content_length() {
use actix_http::{ use actix_http::{
header::{HeaderName, HeaderValue}, header::{HeaderName, HeaderValue},
StatusCode, StatusCode,
@ -445,7 +445,7 @@ async fn test_content_length() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_headers() { async fn h1_headers() {
let data = STR.repeat(10); let data = STR.repeat(10);
let data2 = data.clone(); let data2 = data.clone();
@ -511,7 +511,7 @@ const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
Hello World Hello World Hello World Hello World Hello World"; Hello World Hello World Hello World Hello World Hello World";
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_body() { async fn h1_body() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR))) .h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR)))
@ -530,7 +530,7 @@ async fn test_h1_body() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_head_empty() { async fn h1_head_empty() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR))) .h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR)))
@ -557,7 +557,7 @@ async fn test_h1_head_empty() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_head_binary() { async fn h1_head_binary() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR))) .h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR)))
@ -584,7 +584,7 @@ async fn test_h1_head_binary() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_head_binary2() { async fn h1_head_binary2() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR))) .h1(|_| ok::<_, Infallible>(Response::ok().set_body(STR)))
@ -607,7 +607,7 @@ async fn test_h1_head_binary2() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_body_length() { async fn h1_body_length() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| { .h1(|_| {
@ -631,7 +631,7 @@ async fn test_h1_body_length() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_body_chunked_explicit() { async fn h1_body_chunked_explicit() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| { .h1(|_| {
@ -668,7 +668,7 @@ async fn test_h1_body_chunked_explicit() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_body_chunked_implicit() { async fn h1_body_chunked_implicit() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| { .h1(|_| {
@ -699,7 +699,7 @@ async fn test_h1_body_chunked_implicit() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_response_http_error_handling() { async fn h1_response_http_error_handling() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(fn_service(|_| { .h1(fn_service(|_| {
@ -738,7 +738,7 @@ impl From<BadRequest> for Response<BoxBody> {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_service_error() { async fn h1_service_error() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.h1(|_| err::<Response<()>, _>(BadRequest)) .h1(|_| err::<Response<()>, _>(BadRequest))
@ -757,7 +757,7 @@ async fn test_h1_service_error() {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_h1_on_connect() { async fn h1_on_connect() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.on_connect_ext(|_, data| { .on_connect_ext(|_, data| {
@ -780,7 +780,7 @@ async fn test_h1_on_connect() {
/// Tests compliance with 304 Not Modified spec in RFC 7232 §4.1. /// Tests compliance with 304 Not Modified spec in RFC 7232 §4.1.
/// https://datatracker.ietf.org/doc/html/rfc7232#section-4.1 /// https://datatracker.ietf.org/doc/html/rfc7232#section-4.1
#[actix_rt::test] #[actix_rt::test]
async fn test_not_modified_spec_h1() { async fn not_modified_spec_h1() {
// TODO: this test needing a few seconds to complete reveals some weirdness with either the // TODO: this test needing a few seconds to complete reveals some weirdness with either the
// dispatcher or the client, though similar hangs occur on other tests in this file, only // dispatcher or the client, though similar hangs occur on other tests in this file, only
// succeeding, it seems, because of the keepalive timer // succeeding, it seems, because of the keepalive timer

2
actix-http/tests/test_ws.rs
View File

@ -109,7 +109,7 @@ async fn service(msg: Frame) -> Result<Message, Error> {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_simple() { async fn simple() {
let mut srv = test_server(|| { let mut srv = test_server(|| {
HttpService::build() HttpService::build()
.upgrade(fn_factory(|| async { .upgrade(fn_factory(|| async {