Skip to content
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

Jump to bottom

feat(transport): Add server graceful shutdown #169

Merged
merged 8 commits into from
Dec 11, 2019
Merged

feat(transport): Add server graceful shutdown #169

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
3501592
Started work on graceful shutdown
xd009642 Dec 9, 2019
70ec64d
Fix issues and respond to feedback
xd009642 Dec 9, 2019
2346ab0
Update tonic/src/transport/server.rs
xd009642 Dec 9, 2019
c7b2c93
Tried a different approach
xd009642 Dec 9, 2019
e4df228
Merge branch 'graceful_shutdown' of github.com:xd009642/tonic into gr…
xd009642 Dec 9, 2019
b0d7c1d
Moved to a Pinned Box
xd009642 Dec 9, 2019
973ab61
Move to serve and serve_with_shutdown
xd009642 Dec 9, 2019
fba82a0
stop trying to be smart, just work
xd009642 Dec 11, 2019
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
79 changes: 78 additions & 1 deletion tonic/src/transport/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,7 +261,6 @@ impl Server {
}
},
);

let svc = MakeSvc {
inner: svc,
interceptor,
Expand All @@ -280,6 +279,71 @@ impl Server {

Ok(())
}

pub(crate) async fn serve_with_shutdown<S, F>(
self,
addr: SocketAddr,
svc: S,
signal: F,
) -> Result<(), super::Error>
where
S: Service<Request<Body>, Response = Response<BoxBody>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
F: Future<Output = ()>,
{
let interceptor = self.interceptor.clone();
let concurrency_limit = self.concurrency_limit;
let init_connection_window_size = self.init_connection_window_size;
let init_stream_window_size = self.init_stream_window_size;
let max_concurrent_streams = self.max_concurrent_streams;
// let timeout = self.timeout.clone();

let incoming = hyper::server::accept::from_stream::<_, _, crate::Error>(
async_stream::try_stream! {
let mut tcp = TcpIncoming::bind(addr)?
.set_nodelay(self.tcp_nodelay)
.set_keepalive(self.tcp_keepalive);

while let Some(stream) = tcp.try_next().await? {
#[cfg(feature = "tls")]
{
if let Some(tls) = &self.tls {
let io = match tls.connect(stream.into_inner()).await {
Ok(io) => io,
Err(error) => {
error!(message = "Unable to accept incoming connection.", %error);
continue
},
};
yield BoxedIo::new(io);
continue;
}
}

yield BoxedIo::new(stream);
}
},
);

let svc = MakeSvc {
inner: svc,
interceptor,
concurrency_limit,
// timeout,
};
hyper::Server::builder(incoming)
.http2_only(true)
.http2_initial_connection_window_size(init_connection_window_size)
.http2_initial_stream_window_size(init_stream_window_size)
.http2_max_concurrent_streams(max_concurrent_streams)
.serve(svc)
.with_graceful_shutdown(signal)
.await
.map_err(map_err)?;

Ok(())
}
}

impl<S> Router<S, Unimplemented> {
Expand Down Expand Up @@ -348,6 +412,19 @@ where
pub async fn serve(self, addr: SocketAddr) -> Result<(), super::Error> {
self.server.serve(addr, self.routes).await
}

/// Consume this [`Server`] creating a future that will execute the server
/// on [`tokio`]'s default executor. And shutdown when the provided signal
/// is received.
///
/// [`Server`]: struct.Server.html
pub async fn serve_with_shutdown<F: Future<Output = ()>>(
self,
addr: SocketAddr,
f: F,
) -> Result<(), super::Error> {
self.server.serve_with_shutdown(addr, self.routes, f).await
}
}

fn map_err(e: impl Into<crate::Error>) -> super::Error {
Expand Down