From 1bbaf5107c62243f1d959b7fa8a8ae59376f2123 Mon Sep 17 00:00:00 2001 From: Jan Berktold Date: Tue, 9 Apr 2024 13:40:42 -0700 Subject: [PATCH 1/6] Add !Send server example --- Cargo.toml | 4 + examples/single_threaded_server.rs | 148 +++++++++++++++++++++++++++++ 2 files changed, 152 insertions(+) create mode 100644 examples/single_threaded_server.rs diff --git a/Cargo.toml b/Cargo.toml index 1937a147..207c6bed 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -113,3 +113,7 @@ required-features = ["server", "http1", "tokio"] [[example]] name = "server_graceful" required-features = ["tokio", "server-graceful", "server-auto"] + +[[example]] +name = "single_threaded_server" +required-features = ["server", "http1", "http2", "tokio"] diff --git a/examples/single_threaded_server.rs b/examples/single_threaded_server.rs new file mode 100644 index 00000000..a8917cbb --- /dev/null +++ b/examples/single_threaded_server.rs @@ -0,0 +1,148 @@ +//! This example runs a server that responds to any request with "Hello, world!". +//! Unlike it's server.rs counter part, it demonstrates using a !Send executor (i.e. gloomio, +//! monoio). + +use std::{convert::Infallible, error::Error}; + +use std::marker::PhantomData; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::thread; + +use bytes::Bytes; +use http::{header::CONTENT_TYPE, Request, Response}; +use http_body_util::{combinators::BoxBody, BodyExt, Full}; +use hyper::{body::Incoming, service::service_fn}; +use hyper_util::{ + rt::{TokioExecutor, TokioIo}, + server::conn::auto::Builder, +}; +use tokio::{net::TcpListener, net::TcpStream, task::JoinSet}; + +/// Function from an incoming request to an outgoing response +/// +/// This function gets turned into a [`hyper::service::Service`] later via +/// [`service_fn`]. Instead of doing this, you could also write a type that +/// implements [`hyper::service::Service`] directly and pass that in place of +/// writing a function like this and calling [`service_fn`]. +async fn handle_request(_request: Request) -> Result>, Infallible> { + let response = Response::builder() + .header(CONTENT_TYPE, "text/plain") + .body(Full::new(Bytes::from("Hello, world!\n"))) + .expect("values provided to the builder should be valid"); + + Ok(response) +} + +async fn upgradable_server() -> Result<(), Box> { + let listen_addr = "127.0.0.1:8000"; + let tcp_listener = TcpListener::bind(listen_addr).await?; + println!("listening on http://{listen_addr}"); + + loop { + let (stream, addr) = match tcp_listener.accept().await { + Ok(x) => x, + Err(e) => { + eprintln!("failed to accept connection: {e}"); + continue; + } + }; + + let serve_connection = async move { + println!("handling a request from {addr}"); + + let result = Builder::new(LocalExec) + .serve_connection(TokioIo::new(stream), service_fn(handle_request)) + .await; + + if let Err(e) = result { + eprintln!("error serving {addr}: {e}"); + } + + println!("handled a request from {addr}"); + }; + + tokio::task::spawn_local(serve_connection); + } +} + +fn main() { + let server = thread::spawn(move || { + // Configure a runtime for the server that runs everything on the current thread + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime"); + + // Combine it with a `LocalSet, which means it can spawn !Send futures... + let local = tokio::task::LocalSet::new(); + local.block_on(&rt, upgradable_server()).unwrap(); + }); + + server.join().unwrap() +} + +// NOTE: This part is only needed for HTTP/2. HTTP/1 doesn't need an executor. +// +// Since the Server needs to spawn some background tasks, we needed +// to configure an Executor that can spawn !Send futures... +#[derive(Clone, Copy, Debug)] +struct LocalExec; + +impl hyper::rt::Executor for LocalExec +where + F: std::future::Future + 'static, // not requiring `Send` +{ + fn execute(&self, fut: F) { + // This will spawn into the currently running `LocalSet`. + tokio::task::spawn_local(fut); + } +} + +struct IOTypeNotSend { + _marker: PhantomData<*const ()>, + stream: TokioIo, +} + +impl IOTypeNotSend { + fn new(stream: TokioIo) -> Self { + Self { + _marker: PhantomData, + stream, + } + } +} + +impl hyper::rt::Write for IOTypeNotSend { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.stream).poll_write(cx, buf) + } + + fn poll_flush( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_flush(cx) + } + + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_shutdown(cx) + } +} + +impl hyper::rt::Read for IOTypeNotSend { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: hyper::rt::ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_read(cx, buf) + } +} From 39073251efb02bf8a72ff880d3a962586d00d0be Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Wed, 4 Nov 2026 00:00:00 +0000 Subject: [PATCH 2/6] nit(examples): remove unused imports `cargo check --features=full --examples` returns a number of warnings. this commit fixes these warnings. Signed-off-by: katelyn martin --- examples/single_threaded_server.rs | 24 +++++++++++------------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/examples/single_threaded_server.rs b/examples/single_threaded_server.rs index a8917cbb..42432a89 100644 --- a/examples/single_threaded_server.rs +++ b/examples/single_threaded_server.rs @@ -2,22 +2,20 @@ //! Unlike it's server.rs counter part, it demonstrates using a !Send executor (i.e. gloomio, //! monoio). -use std::{convert::Infallible, error::Error}; - -use std::marker::PhantomData; -use std::pin::Pin; -use std::task::{Context, Poll}; -use std::thread; - use bytes::Bytes; -use http::{header::CONTENT_TYPE, Request, Response}; -use http_body_util::{combinators::BoxBody, BodyExt, Full}; +use http::{Request, Response, header::CONTENT_TYPE}; +use http_body_util::Full; use hyper::{body::Incoming, service::service_fn}; -use hyper_util::{ - rt::{TokioExecutor, TokioIo}, - server::conn::auto::Builder, +use hyper_util::{rt::TokioIo, server::conn::auto::Builder}; +use std::{ + convert::Infallible, + error::Error, + marker::PhantomData, + pin::Pin, + task::{Context, Poll}, + thread, }; -use tokio::{net::TcpListener, net::TcpStream, task::JoinSet}; +use tokio::{net::TcpListener, net::TcpStream}; /// Function from an incoming request to an outgoing response /// From d4289c690efe960b99a89223c9c6af443113e6b1 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Wed, 4 Nov 2026 00:00:00 +0000 Subject: [PATCH 3/6] refactor(examples): use `#[tokio::main]` macro this macro allows callers to specify the kind of runtime they wish to use. we can use this, instead of laying out the boilerplate to do this by hand, to provide a more concise example. Signed-off-by: katelyn martin --- examples/single_threaded_server.rs | 20 ++------------------ 1 file changed, 2 insertions(+), 18 deletions(-) diff --git a/examples/single_threaded_server.rs b/examples/single_threaded_server.rs index 42432a89..4b8e45ad 100644 --- a/examples/single_threaded_server.rs +++ b/examples/single_threaded_server.rs @@ -13,7 +13,6 @@ use std::{ marker::PhantomData, pin::Pin, task::{Context, Poll}, - thread, }; use tokio::{net::TcpListener, net::TcpStream}; @@ -32,7 +31,8 @@ async fn handle_request(_request: Request) -> Result Result<(), Box> { +#[tokio::main(flavor = "local")] +async fn main() -> Result<(), Box> { let listen_addr = "127.0.0.1:8000"; let tcp_listener = TcpListener::bind(listen_addr).await?; println!("listening on http://{listen_addr}"); @@ -64,22 +64,6 @@ async fn upgradable_server() -> Result<(), Box> { } } -fn main() { - let server = thread::spawn(move || { - // Configure a runtime for the server that runs everything on the current thread - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("build runtime"); - - // Combine it with a `LocalSet, which means it can spawn !Send futures... - let local = tokio::task::LocalSet::new(); - local.block_on(&rt, upgradable_server()).unwrap(); - }); - - server.join().unwrap() -} - // NOTE: This part is only needed for HTTP/2. HTTP/1 doesn't need an executor. // // Since the Server needs to spawn some background tasks, we needed From 003b848ed612566daccff1d12cd5407b3e5d639c Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Wed, 4 Nov 2026 00:00:00 +0000 Subject: [PATCH 4/6] refactor(examples): remove unused `IOTypeNotSend` this type isn't used in the example. we can remove it. Signed-off-by: katelyn martin --- examples/single_threaded_server.rs | 58 ++---------------------------- 1 file changed, 2 insertions(+), 56 deletions(-) diff --git a/examples/single_threaded_server.rs b/examples/single_threaded_server.rs index 4b8e45ad..67c43337 100644 --- a/examples/single_threaded_server.rs +++ b/examples/single_threaded_server.rs @@ -7,14 +7,8 @@ use http::{Request, Response, header::CONTENT_TYPE}; use http_body_util::Full; use hyper::{body::Incoming, service::service_fn}; use hyper_util::{rt::TokioIo, server::conn::auto::Builder}; -use std::{ - convert::Infallible, - error::Error, - marker::PhantomData, - pin::Pin, - task::{Context, Poll}, -}; -use tokio::{net::TcpListener, net::TcpStream}; +use std::{convert::Infallible, error::Error}; +use tokio::net::TcpListener; /// Function from an incoming request to an outgoing response /// @@ -80,51 +74,3 @@ where tokio::task::spawn_local(fut); } } - -struct IOTypeNotSend { - _marker: PhantomData<*const ()>, - stream: TokioIo, -} - -impl IOTypeNotSend { - fn new(stream: TokioIo) -> Self { - Self { - _marker: PhantomData, - stream, - } - } -} - -impl hyper::rt::Write for IOTypeNotSend { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - Pin::new(&mut self.stream).poll_write(cx, buf) - } - - fn poll_flush( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { - Pin::new(&mut self.stream).poll_flush(cx) - } - - fn poll_shutdown( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { - Pin::new(&mut self.stream).poll_shutdown(cx) - } -} - -impl hyper::rt::Read for IOTypeNotSend { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: hyper::rt::ReadBufCursor<'_>, - ) -> Poll> { - Pin::new(&mut self.stream).poll_read(cx, buf) - } -} From b68037008014e14425c374ccb30147dd43e4fa95 Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Wed, 4 Nov 2026 00:00:00 +0000 Subject: [PATCH 5/6] refactor(examples): use `TokioLocalExecutor` in #330, we introduced a new executor that can spawn futures onto a local runtime. this commit replaces the `LocalExec` type in the example with this executor, to further distill this example. Signed-off-by: katelyn martin --- examples/single_threaded_server.rs | 24 +++++------------------- 1 file changed, 5 insertions(+), 19 deletions(-) diff --git a/examples/single_threaded_server.rs b/examples/single_threaded_server.rs index 67c43337..5f0c14c1 100644 --- a/examples/single_threaded_server.rs +++ b/examples/single_threaded_server.rs @@ -6,7 +6,10 @@ use bytes::Bytes; use http::{Request, Response, header::CONTENT_TYPE}; use http_body_util::Full; use hyper::{body::Incoming, service::service_fn}; -use hyper_util::{rt::TokioIo, server::conn::auto::Builder}; +use hyper_util::{ + rt::{TokioIo, tokio::TokioLocalExecutor}, + server::conn::auto::Builder, +}; use std::{convert::Infallible, error::Error}; use tokio::net::TcpListener; @@ -43,7 +46,7 @@ async fn main() -> Result<(), Box> { let serve_connection = async move { println!("handling a request from {addr}"); - let result = Builder::new(LocalExec) + let result = Builder::new(TokioLocalExecutor::new()) .serve_connection(TokioIo::new(stream), service_fn(handle_request)) .await; @@ -57,20 +60,3 @@ async fn main() -> Result<(), Box> { tokio::task::spawn_local(serve_connection); } } - -// NOTE: This part is only needed for HTTP/2. HTTP/1 doesn't need an executor. -// -// Since the Server needs to spawn some background tasks, we needed -// to configure an Executor that can spawn !Send futures... -#[derive(Clone, Copy, Debug)] -struct LocalExec; - -impl hyper::rt::Executor for LocalExec -where - F: std::future::Future + 'static, // not requiring `Send` -{ - fn execute(&self, fut: F) { - // This will spawn into the currently running `LocalSet`. - tokio::task::spawn_local(fut); - } -} From 0d9ed8e95126081b3d5f615bc1b6b88ecfa7b67f Mon Sep 17 00:00:00 2001 From: katelyn martin Date: Wed, 4 Nov 2026 00:00:00 +0000 Subject: [PATCH 6/6] doc(examples): additional context about example Signed-off-by: katelyn martin --- examples/single_threaded_server.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/examples/single_threaded_server.rs b/examples/single_threaded_server.rs index 5f0c14c1..954a18a5 100644 --- a/examples/single_threaded_server.rs +++ b/examples/single_threaded_server.rs @@ -1,6 +1,8 @@ //! This example runs a server that responds to any request with "Hello, world!". -//! Unlike it's server.rs counter part, it demonstrates using a !Send executor (i.e. gloomio, -//! monoio). +//! +//! Unlike its `server.rs` counter part, this server runs on a local single-threaded runtime, +//! and uses a `!Send` executor. This pattern is useful for running hyper servers on asynchronous +//! runtimes like gloomio or monoio. use bytes::Bytes; use http::{Request, Response, header::CONTENT_TYPE};