Skip to content

Upgrade Tokio to v1.x and Hyper to v0.14 #3804

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 8, 2025
Merged

Upgrade Tokio to v1.x and Hyper to v0.14 #3804

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
70c8831
update of tokio and related dependencies to 1.x
Mar 2, 2025
fcf37e7
update to hyper 0.14
Mar 2, 2025
2d778c6
fixes to http connector for tests
Mar 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
547 changes: 186 additions & 361 deletions Cargo.lock
View file
Open in desktop

Large diffs are not rendered by default.

16 changes: 9 additions & 7 deletions api/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ edition = "2018"

[dependencies]
easy-jsonrpc-mw = "0.5.4"
hyper = "0.13"
hyper = { version = "0.14", features = ["full"] }
lazy_static = "1"
regex = "1"
ring = "0.16"
Expand All @@ -20,15 +20,17 @@ serde_derive = "1"
serde_json = "1"
thiserror = "1"
log = "0.4"
tokio = { version = "0.2", features = ["full"] }
tokio-rustls = "0.13"
tokio = { version = "1", features = ["full"] }
tokio-rustls = "0.23"
http = "0.2"
hyper-rustls = "0.20"
hyper-timeout = "0.3"
hyper-rustls = "0.23"
hyper-timeout = "0.4"
futures = "0.3"
rustls = "0.17"
rustls = "0.20"
rustls-pemfile = "1.0"
async-stream = "0.3"
url = "2.1"
bytes = "0.5"
bytes = "1"

grin_core = { path = "../core", version = "5.4.0-alpha.0" }
grin_chain = { path = "../chain", version = "5.4.0-alpha.0" }
Expand Down
12 changes: 8 additions & 4 deletions api/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,12 @@ where
}

async fn send_request_async(req: Request<Body>, timeout: TimeOut) -> Result<String, Error> {
let https = hyper_rustls::HttpsConnector::new();
let https = hyper_rustls::HttpsConnectorBuilder::new()
.with_native_roots()
.https_or_http()
.enable_http1()
.build();

let (connect, read, write) = (
Some(timeout.connect),
Some(timeout.read),
Expand All @@ -242,16 +247,15 @@ async fn send_request_async(req: Request<Body>, timeout: TimeOut) -> Result<Stri
.into());
}

let raw = body::to_bytes(resp)
let raw = body::to_bytes(resp.into_body())
.await
.map_err(|e| Error::RequestError(format!("Cannot read response body: {}", e)))?;

Ok(String::from_utf8_lossy(&raw).to_string())
}

pub fn send_request(req: Request<Body>, timeout: TimeOut) -> Result<String, Error> {
let mut rt = Builder::new()
.basic_scheduler()
let rt = Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| Error::RequestError(format!("{}", e)))?;
Expand Down
6 changes: 3 additions & 3 deletions api/src/json_rpc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ pub enum Error {
/// Json error
Json(serde_json::Error),
/// Client error
Hyper(hyper::error::Error),
Hyper(hyper::Error),
/// Error response
Rpc(RpcError),
/// Response to a request did not have the expected nonce
Expand All @@ -120,8 +120,8 @@ impl From<serde_json::Error> for Error {
}
}

impl From<hyper::error::Error> for Error {
fn from(e: hyper::error::Error) -> Error {
impl From<hyper::Error> for Error {
fn from(e: hyper::Error) -> Error {
Error::Hyper(e)
}
}
Expand Down
49 changes: 33 additions & 16 deletions api/src/rest.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,17 @@
use crate::router::{Handler, HandlerObj, ResponseFuture, Router, RouterError};
use crate::web::response;
use futures::channel::oneshot;
use futures::TryStreamExt;
use hyper::server::accept;
use hyper::service::make_service_fn;
use hyper::{Body, Request, Server, StatusCode};
use rustls::internal::pemfile;
use rustls_pemfile as pemfile;
use std::convert::Infallible;
use std::fs::File;
use std::net::SocketAddr;
use std::sync::Arc;
use std::{io, thread};
use tokio::net::TcpListener;
use tokio::runtime::Runtime;
use tokio::stream::StreamExt;
use tokio_rustls::TlsAcceptor;

/// Errors that can be returned by an ApiEndpoint implementation.
Expand Down Expand Up @@ -83,8 +81,10 @@ impl TLSConfig {
})?;
let mut reader = io::BufReader::new(certfile);

pemfile::certs(&mut reader)
.map_err(|_| Error::Internal("failed to load certificate".to_string()))
let certs = pemfile::certs(&mut reader)
.map_err(|_| Error::Internal("failed to load certificate".to_string()))?;

Ok(certs.into_iter().map(rustls::Certificate).collect())
}

fn load_private_key(&self) -> Result<rustls::PrivateKey, Error> {
Expand All @@ -97,15 +97,19 @@ impl TLSConfig {
if keys.len() != 1 {
return Err(Error::Internal("expected a single private key".to_string()));
}
Ok(keys[0].clone())
Ok(rustls::PrivateKey(keys[0].clone()))
}

pub fn build_server_config(&self) -> Result<Arc<rustls::ServerConfig>, Error> {
let certs = self.load_certs()?;
let key = self.load_private_key()?;
let mut cfg = rustls::ServerConfig::new(rustls::NoClientAuth::new());
cfg.set_single_cert(certs, key)

let cfg = rustls::ServerConfig::builder()
.with_safe_defaults()
.with_no_client_auth()
.with_single_cert(certs, key)
.map_err(|e| Error::Internal(format!("set single certificate failed {}", e)))?;

Ok(Arc::new(cfg))
}
}
Expand Down Expand Up @@ -175,7 +179,7 @@ impl ApiServer {
server.await
};

let mut rt = Runtime::new()
let rt = Runtime::new()
.map_err(|e| eprintln!("HTTP API server error: {}", e))
.unwrap();
if let Err(e) = rt.block_on(server) {
Expand Down Expand Up @@ -214,13 +218,26 @@ impl ApiServer {
.name("apis".to_string())
.spawn(move || {
let server = async move {
let mut listener = TcpListener::bind(&addr).await.expect("failed to bind");
let listener = listener
.incoming()
.and_then(move |s| acceptor.accept(s))
.filter(|r| r.is_ok());
let listener = TcpListener::bind(&addr).await.expect("failed to bind");

let tls_stream = async_stream::stream! {
loop {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why does this need a loop that was not needed before?

Copy link
Member Author

@yeastplume yeastplume Mar 8, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Without the loop it would handle a single connection before stopping, this is the common pattern in this particular version of hyper (and beyond, I think, though we'll get to updating hyper to the latest in a future PR). An example here https://github.com/hs-CN/async-rustls-stream/blob/ec4298493353d3fae6f86e106d07543b9f1bea56/examples/server_client.rs#L38

let (socket, _addr) = match listener.accept().await {
Ok(conn) => conn,
Err(e) => {
eprintln!("Error accepting connection: {}", e);
continue;
}
};

match acceptor.accept(socket).await {
Ok(stream) => yield Ok::<_, std::io::Error>(stream),
Err(_) => continue,
}
}
};

let server = Server::builder(accept::from_stream(listener))
let server = Server::builder(accept::from_stream(tls_stream))
.serve(make_service_fn(move |_| {
let router = router.clone();
async move { Ok::<_, Infallible>(router) }
Expand All @@ -232,7 +249,7 @@ impl ApiServer {
server.await
};

let mut rt = Runtime::new()
let rt = Runtime::new()
.map_err(|e| eprintln!("HTTP API server error: {}", e))
.unwrap();
if let Err(e) = rt.block_on(server) {
Expand Down
5 changes: 3 additions & 2 deletions api/src/web.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
use crate::rest::*;
use crate::router::ResponseFuture;
use bytes::Buf;
use futures::future::ok;
use hyper::body;
use hyper::{Body, Request, Response, StatusCode};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::Debug;
use std::io::Cursor;
use url::form_urlencoded;

/// Parse request body
Expand All @@ -18,7 +18,8 @@ where
.await
.map_err(|e| Error::RequestError(format!("Failed to read request: {}", e)))?;

serde_json::from_reader(raw.bytes())
let cursor = Cursor::new(raw);
serde_json::from_reader(cursor)
.map_err(|e| Error::RequestError(format!("Invalid request body: {}", e)))
}

Expand Down
4 changes: 2 additions & 2 deletions core/src/core/pmmr/pmmr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ where
B: 'a + Backend<T>,
{
/// Build a new prunable Merkle Mountain Range using the provided backend.
pub fn new(backend: &'a mut B) -> PMMR<'_, T, B> {
pub fn new(backend: &'a mut B) -> PMMR<'a, T, B> {
PMMR {
backend,
size: 0,
Expand All @@ -198,7 +198,7 @@ where

/// Build a new prunable Merkle Mountain Range pre-initialized until
/// size with the provided backend.
pub fn at(backend: &'a mut B, size: u64) -> PMMR<'_, T, B> {
pub fn at(backend: &'a mut B, size: u64) -> PMMR<'a, T, B> {
PMMR {
backend,
size,
Expand Down
4 changes: 2 additions & 2 deletions core/src/core/pmmr/readonly_pmmr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ where
B: 'a + Backend<T>,
{
/// Build a new readonly PMMR.
pub fn new(backend: &'a B) -> ReadonlyPMMR<'_, T, B> {
pub fn new(backend: &'a B) -> ReadonlyPMMR<'a, T, B> {
ReadonlyPMMR {
backend,
size: 0,
Expand All @@ -51,7 +51,7 @@ where

/// Build a new readonly PMMR pre-initialized to
/// size with the provided backend.
pub fn at(backend: &'a B, size: u64) -> ReadonlyPMMR<'_, T, B> {
pub fn at(backend: &'a B, size: u64) -> ReadonlyPMMR<'a, T, B> {
ReadonlyPMMR {
backend,
size,
Expand Down
4 changes: 2 additions & 2 deletions core/src/core/pmmr/rewindable_pmmr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ where
B: 'a + Backend<T>,
{
/// Build a new readonly PMMR.
pub fn new(backend: &'a B) -> RewindablePMMR<'_, T, B> {
pub fn new(backend: &'a B) -> RewindablePMMR<'a, T, B> {
RewindablePMMR {
backend,
last_pos: 0,
Expand All @@ -50,7 +50,7 @@ where

/// Build a new readonly PMMR pre-initialized to
/// last_pos with the provided backend.
pub fn at(backend: &'a B, last_pos: u64) -> RewindablePMMR<'_, T, B> {
pub fn at(backend: &'a B, last_pos: u64) -> RewindablePMMR<'a, T, B> {
RewindablePMMR {
backend,
last_pos,
Expand Down
10 changes: 6 additions & 4 deletions servers/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@ workspace = ".."
edition = "2018"

[dependencies]
hyper = "0.13"
hyper-rustls = "0.20"
hyper = { version = "0.14", features = ["full"] }
hyper-rustls = "0.23"
fs2 = "0.4"
futures = "0.3"
http = "0.2"
Expand All @@ -22,8 +22,10 @@ log = "0.4"
serde_derive = "1"
serde_json = "1"
chrono = "0.4.11"
tokio = {version = "0.2", features = ["full"] }
tokio-util = { version = "0.2", features = ["codec"] }
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.7", features = ["codec"] }
async-stream = "0.3"
rustls = "0.20"
walkdir = "2.3.1"

grin_api = { path = "../api", version = "5.4.0-alpha.0" }
Expand Down
12 changes: 8 additions & 4 deletions servers/src/common/hooks.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,7 +220,12 @@ impl WebHook {
nthreads, timeout
);

let https = HttpsConnector::new();
let https = hyper_rustls::HttpsConnectorBuilder::new()
.with_native_roots()
.https_only()
.enable_http1()
.build();

let client = Client::builder()
.pool_idle_timeout(keep_alive)
.build::<_, hyper::Body>(https);
Expand All @@ -231,10 +236,9 @@ impl WebHook {
header_received_url,
block_accepted_url,
client,
runtime: Builder::new()
.threaded_scheduler()
runtime: Builder::new_multi_thread()
.enable_all()
.core_threads(nthreads as usize)
.worker_threads(nthreads as usize)
.build()
.unwrap(),
}
Expand Down
Loading