forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): keep-alive!

Browse Source 
Internals have been shuffled around such that Request and Reponse are
now given only a mutable reference to the stream, instead of being
allowed to consume it. This allows the server to re-use the streams if
keep-alive is true.

A task pool is used, and the number of the threads can currently be
adjusted by using the `listen_threads()` method on Server.

[breaking-change]
This commit is contained in:
Sean McArthur
2014-11-12 15:17:41 -08:00
parent 1f2f93cfea
commit 3cd9b10bcb
12 changed files with 167 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.toml
View File

@@ -16,9 +16,6 @@ git = "https://github.com/hyperium/mime.rs"
[dependencies.unsafe-any]
git = "https://github.com/reem/rust-unsafe-any"
[dependencies.move-acceptor]
git = "https://github.com/reem/rust-move-acceptor"
[dependencies.typeable]
git = "https://github.com/reem/rust-typeable"

17
benches/client.rs
View File

@@ -7,28 +7,25 @@ extern crate test;
use std::fmt::{mod, Show};
use std::io::net::ip::Ipv4Addr;
use hyper::server::{Incoming, Server};
use hyper::server::{Request, Response, Server};
fn listen() -> hyper::server::Listening {
let server = Server::http(Ipv4Addr(127, 0, 0, 1), 0);
server.listen(handle).unwrap()
}
macro_rules! try_continue(
macro_rules! try_return(
($e:expr) => {{
match $e {
Ok(v) => v,
Err(..) => continue
Err(..) => return
}
}})
fn handle(mut incoming: Incoming) {
for conn in incoming {
let (_, res) = try_continue!(conn.open());
let mut res = try_continue!(res.start());
try_continue!(res.write(b"Benchmarking hyper vs others!"))
try_continue!(res.end());
}
fn handle(_: Request, res: Response) {
let mut res = try_return!(res.start());
try_return!(res.write(b"Benchmarking hyper vs others!"))
try_return!(res.end());
}

12
benches/server.rs
View File

@@ -9,6 +9,7 @@ use test::Bencher;
use std::io::net::ip::{SocketAddr, Ipv4Addr};
use http::server::Server;
use hyper::server::{Request, Response};
static PHRASE: &'static [u8] = b"Benchmarking hyper vs others!";
@@ -17,13 +18,10 @@ fn request(url: hyper::Url) {
req.start().unwrap().send().unwrap().read_to_string().unwrap();
}
fn hyper_handle(mut incoming: hyper::server::Incoming) {
for conn in incoming {
let (_, res) = conn.open().unwrap();
let mut res = res.start().unwrap();
res.write(PHRASE).unwrap();
res.end().unwrap();
}
fn hyper_handle(_: Request, res: Response) {
let mut res = res.start().unwrap();
res.write(PHRASE).unwrap();
res.end().unwrap();
}
#[bench]

77
examples/concurrent-server.rs
View File

@@ -1,77 +0,0 @@
#![feature(macro_rules, default_type_params)]
extern crate hyper;
use std::io::util::copy;
use std::io::net::ip::Ipv4Addr;
use std::sync::Arc;
use hyper::{Get, Post};
use hyper::server::{Server, Handler, Incoming, Request, Response};
use hyper::header::common::ContentLength;
use hyper::net::{HttpStream, HttpAcceptor, Fresh};
macro_rules! try_abort(
($e:expr) => {{
match $e {
Ok(v) => v,
Err(..) => return
}
}}
)
trait ConcurrentHandler: Send + Sync {
fn handle(&self, req: Request, res: Response<Fresh>);
}
struct Concurrent<H: ConcurrentHandler> { handler: Arc<H> }
impl<H: ConcurrentHandler> Handler<HttpAcceptor, HttpStream> for Concurrent<H> {
fn handle(self, mut incoming: Incoming) {
for conn in incoming {
let clone = self.handler.clone();
spawn(proc() {
let (req, res) = try_abort!(conn.open());
clone.handle(req, res);
})
}
}
}
struct Echo;
impl ConcurrentHandler for Echo {
fn handle(&self, mut req: Request, mut res: Response<Fresh>) {
match req.uri {
hyper::uri::AbsolutePath(ref path) => match (&req.method, path.as_slice()) {
(&Get, "/") | (&Get, "/echo") => {
let out = b"Try POST /echo";
res.headers_mut().set(ContentLength(out.len()));
let mut res = try_abort!(res.start());
try_abort!(res.write(out));
try_abort!(res.end());
return;
},
(&Post, "/echo") => (), // fall through, fighting mutable borrows
_ => {
*res.status_mut() = hyper::status::NotFound;
try_abort!(res.start().and_then(|res| res.end()));
return;
}
},
_ => {
try_abort!(res.start().and_then(|res| res.end()));
return;
}
}
let mut res = try_abort!(res.start());
try_abort!(copy(&mut req, &mut res));
try_abort!(res.end());
}
}
fn main() {
let server = Server::http(Ipv4Addr(127, 0, 0, 1), 3000);
server.listen(Concurrent { handler: Arc::new(Echo) }).unwrap();
}

19
examples/hello.rs
View File

@@ -1,23 +1,16 @@
extern crate hyper;
use std::sync::TaskPool;
use std::io::net::ip::Ipv4Addr;
use hyper::server::{Request, Response};
static PHRASE: &'static [u8] = b"Hello World!";
fn hyper_handle(mut incoming: hyper::server::Incoming) {
let pool = TaskPool::new(100);
for conn in incoming {
pool.execute(proc() {
let (_, res) = conn.open().unwrap();
let mut res = res.start().unwrap();
res.write(PHRASE).unwrap();
res.end().unwrap();
});
}
fn hello(_: Request, res: Response) {
let mut res = res.start().unwrap();
res.write(PHRASE).unwrap();
res.end().unwrap();
}
fn main() {
hyper::Server::http(Ipv4Addr(127, 0, 0, 1), 3000).listen(hyper_handle).unwrap();
hyper::Server::http(Ipv4Addr(127, 0, 0, 1), 3000).listen(hello).unwrap();
}

58
examples/server.rs
View File

@@ -1,54 +1,52 @@
#![feature(macro_rules)]
#![feature(macro_rules, phase)]
extern crate hyper;
#[phase(plugin, link)] extern crate log;
use std::io::util::copy;
use std::io::net::ip::Ipv4Addr;
use hyper::{Get, Post};
use hyper::header::common::ContentLength;
use hyper::server::{Server, Incoming};
use hyper::server::{Server, Request, Response};
macro_rules! try_continue(
macro_rules! try_return(
($e:expr) => {{
match $e {
Ok(v) => v,
Err(e) => { println!("Error: {}", e); continue; }
Err(e) => { error!("Error: {}", e); return; }
}
}}
)
fn echo(mut incoming: Incoming) {
for conn in incoming {
let (mut req, mut res) = try_continue!(conn.open());
match req.uri {
hyper::uri::AbsolutePath(ref path) => match (&req.method, path.as_slice()) {
(&Get, "/") | (&Get, "/echo") => {
let out = b"Try POST /echo";
fn echo(mut req: Request, mut res: Response) {
match req.uri {
hyper::uri::AbsolutePath(ref path) => match (&req.method, path.as_slice()) {
(&Get, "/") | (&Get, "/echo") => {
let out = b"Try POST /echo";
res.headers_mut().set(ContentLength(out.len()));
let mut res = try_continue!(res.start());
try_continue!(res.write(out));
try_continue!(res.end());
continue;
},
(&Post, "/echo") => (), // fall through, fighting mutable borrows
_ => {
*res.status_mut() = hyper::status::NotFound;
try_continue!(res.start().and_then(|res| res.end()));
continue;
}
res.headers_mut().set(ContentLength(out.len()));
let mut res = try_return!(res.start());
try_return!(res.write(out));
try_return!(res.end());
return;
},
(&Post, "/echo") => (), // fall through, fighting mutable borrows
_ => {
try_continue!(res.start().and_then(|res| res.end()));
continue;
*res.status_mut() = hyper::status::NotFound;
try_return!(res.start().and_then(|res| res.end()));
return;
}
};
},
_ => {
try_return!(res.start().and_then(|res| res.end()));
return;
}
};
let mut res = try_continue!(res.start());
try_continue!(copy(&mut req, &mut res));
try_continue!(res.end());
}
let mut res = try_return!(res.start());
try_return!(copy(&mut req, &mut res));
try_return!(res.end());
}
fn main() {

2
src/header/common/connection.rs
View File

@@ -3,7 +3,7 @@ use std::fmt::{mod, Show};
use std::str::FromStr;
use super::util::{from_comma_delimited, fmt_comma_delimited};
use self::ConnectionOption::{KeepAlive, Close, ConnectionHeader};
pub use self::ConnectionOption::{KeepAlive, Close, ConnectionHeader};
/// The `Connection` header.
#[deriving(Clone, PartialEq, Show)]

12
src/lib.rs
View File

@@ -66,10 +66,10 @@
//!
//! #### Handler + Server
//!
//! A Handler in Hyper just accepts an Iterator of `(Request, Response)` pairs and
//! does whatever it wants with it. This gives Handlers maximum flexibility to decide
//! on concurrency strategy and exactly how they want to distribute the work of
//! dealing with `Request` and `Response.`
//! A `Handler` in Hyper accepts a `Request` and `Response`. This is where
//! user-code can handle each connection. The server accepts connections in a
//! task pool with a customizable number of threads, and passes the Request /
//! Response to the handler.
//!
//! #### Request
//!
@@ -133,7 +133,6 @@ extern crate openssl;
#[phase(plugin,link)] extern crate log;
#[cfg(test)] extern crate test;
extern crate "unsafe-any" as uany;
extern crate "move-acceptor" as macceptor;
extern crate typeable;
extern crate cookie;
@@ -252,7 +251,4 @@ impl FromError<IoError> for HttpError {
fn _assert_send<T: Send>() {
_assert_send::<client::Request<net::Fresh>>();
_assert_send::<client::Response>();
_assert_send::<server::Request>();
_assert_send::<server::Response<net::Fresh>>();
}

13
src/net.rs
View File

@@ -82,6 +82,19 @@ impl Writer for Box<NetworkStream + Send> {
fn flush(&mut self) -> IoResult<()> { (**self).flush() }
}
impl<'a> Reader for &'a mut NetworkStream {
#[inline]
fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> { (**self).read(buf) }
}
impl<'a> Writer for &'a mut NetworkStream {
#[inline]
fn write(&mut self, msg: &[u8]) -> IoResult<()> { (**self).write(msg) }
#[inline]
fn flush(&mut self) -> IoResult<()> { (**self).flush() }
}
impl UncheckedBoxAnyDowncast for Box<NetworkStream + Send> {
unsafe fn downcast_unchecked<T: 'static>(self) -> Box<T> {
let to = *mem::transmute::<&Box<NetworkStream + Send>, &raw::TraitObject>(&self);

137
src/server/mod.rs
View File

@@ -1,17 +1,22 @@
//! HTTP Server
use std::io::{Listener, EndOfFile};
use std::io::{Listener, EndOfFile, BufferedReader, BufferedWriter};
use std::io::net::ip::{IpAddr, Port, SocketAddr};
use std::os;
use std::sync::{Arc, TaskPool};
use std::task::TaskBuilder;
use macceptor::{MoveAcceptor, MoveConnections};
pub use self::request::Request;
pub use self::response::Response;
pub use net::{Fresh, Streaming};
use {HttpResult};
use header::common::Connection;
use header::common::connection::{KeepAlive, Close};
use net::{NetworkListener, NetworkAcceptor, NetworkStream,
HttpAcceptor, HttpListener, HttpStream,
Fresh};
HttpAcceptor, HttpListener, HttpStream};
use version::HttpVersion::{Http10, Http11};
pub mod request;
pub mod response;
@@ -45,12 +50,12 @@ impl Server<HttpListener> {
}
impl<L: NetworkListener<S, A>, S: NetworkStream, A: NetworkAcceptor<S>> Server<L> {
/// Binds to a socket, and starts handling connections.
/// Binds to a socket, and starts handling connections using a task pool.
///
/// This method has unbound type parameters, so can be used when you want to use
/// something other than the provided HttpStream, HttpAcceptor, and HttpListener.
pub fn listen_network<H, S, A, L>(self, handler: H) -> HttpResult<Listening<A>>
where H: Handler<A, S>,
pub fn listen_network<H, S, A, L>(self, handler: H, threads: uint) -> HttpResult<Listening<A>>
where H: Handler,
S: NetworkStream,
A: NetworkAcceptor<S>,
L: NetworkListener<S, A>, {
@@ -61,9 +66,61 @@ impl<L: NetworkListener<S, A>, S: NetworkStream, A: NetworkAcceptor<S>> Server<L
let acceptor = try!(listener.listen());
let captured = acceptor.clone();
let mut captured = acceptor.clone();
TaskBuilder::new().named("hyper acceptor").spawn(proc() {
handler.handle(Incoming { from: captured.move_incoming() });
let handler = Arc::new(handler);
debug!("threads = {}", threads);
let pool = TaskPool::new(threads);
for conn in captured.incoming() {
match conn {
Ok(mut stream) => {
debug!("Incoming stream");
let handler = handler.clone();
pool.execute(proc() {
let addr = match stream.peer_name() {
Ok(addr) => addr,
Err(e) => {
error!("Peer Name error: {}", e);
return;
}
};
let mut rdr = BufferedReader::new(stream.clone());
let mut wrt = BufferedWriter::new(stream);
let mut keep_alive = true;
while keep_alive {
let mut res = Response::new(&mut wrt);
let req = match Request::new(&mut rdr, addr) {
Ok(req) => req,
Err(e) => {
//TODO: send a 400 response
error!("request error: {}", e);
return;
}
};
keep_alive = match (req.version, req.headers.get::<Connection>()) {
(Http10, Some(conn)) if !conn.0.contains(&KeepAlive) => false,
(Http11, Some(conn)) if conn.0.contains(&Close) => false,
_ => true
};
res.version = req.version;
handler.handle(req, res);
debug!("keep_alive = {}", keep_alive);
}
});
},
Err(ref e) if e.kind == EndOfFile => {
debug!("server closed");
break;
},
Err(e) => {
error!("Connection failed: {}", e);
continue;
}
}
}
});
Ok(Listening {
@@ -72,49 +129,16 @@ impl<L: NetworkListener<S, A>, S: NetworkStream, A: NetworkAcceptor<S>> Server<L
})
}
/// Binds to a socket and starts handling connections with the specified number of tasks.
pub fn listen_threads<H: Handler>(self, handler: H, threads: uint) -> HttpResult<Listening<HttpAcceptor>> {
self.listen_network::<H, HttpStream, HttpAcceptor, HttpListener>(handler, threads)
}
/// Binds to a socket and starts handling connections.
pub fn listen<H: Handler<HttpAcceptor, HttpStream>>(self, handler: H) -> HttpResult<Listening<HttpAcceptor>> {
self.listen_network::<H, HttpStream, HttpAcceptor, HttpListener>(handler)
pub fn listen<H: Handler>(self, handler: H) -> HttpResult<Listening<HttpAcceptor>> {
self.listen_threads(handler, os::num_cpus() * 5 / 4)
}
}
/// An iterator over incoming `Connection`s.
pub struct Incoming<A = HttpAcceptor> {
from: MoveConnections<A>
}
impl<S: NetworkStream + 'static, A: NetworkAcceptor<S>> Iterator<Connection<S>> for Incoming<A> {
fn next(&mut self) -> Option<Connection<S>> {
for conn in self.from {
match conn {
Ok(stream) => {
debug!("Incoming stream");
return Some(Connection(stream));
},
Err(ref e) if e.kind == EndOfFile => return None, // server closed
Err(e) => {
error!("Connection failed: {}", e);
continue;
}
}
}
None
}
}
/// An incoming connection. It can be opened to receive a request/response pair.
pub struct Connection<S: Send = HttpStream>(S);
impl<S: NetworkStream + 'static> Connection<S> {
/// Opens the incoming connection, parsing it into a Request/Response pair.
pub fn open(self) -> HttpResult<(Request, Response<Fresh>)> {
let stream = self.0;
let clone = stream.clone();
let req = try!(Request::new(stream));
let mut res = Response::new(clone);
res.version = req.version;
return Ok((req, res))
}
}
/// A listening server, which can later be closed.
@@ -125,10 +149,7 @@ pub struct Listening<A = HttpAcceptor> {
}
impl<A: NetworkAcceptor<S>, S: NetworkStream> Listening<A> {
/// Stop the server from listening to all of its socket addresses.
///
/// If closing any of the servers acceptors fails, this function returns Err
/// and does not close the rest of the acceptors.
/// Stop the server from listening to its socket address.
pub fn close(&mut self) -> HttpResult<()> {
debug!("closing server");
try!(self.acceptor.close());
@@ -137,16 +158,16 @@ impl<A: NetworkAcceptor<S>, S: NetworkStream> Listening<A> {
}
/// A handler that can handle incoming requests for a server.
pub trait Handler<A: NetworkAcceptor<S>, S: NetworkStream>: Send {
pub trait Handler: Sync + Send {
/// Receives a `Request`/`Response` pair, and should perform some action on them.
///
/// This could reading from the request, and writing to the response.
fn handle(self, Incoming<A>);
fn handle(&self, Request, Response<Fresh>);
}
impl<A: NetworkAcceptor<S>, S: NetworkStream> Handler<A, S> for fn(Incoming<A>) {
fn handle(self, incoming: Incoming<A>) {
(self)(incoming)
impl Handler for fn(Request, Response<Fresh>) {
fn handle(&self, req: Request, res: Response<Fresh>) {
(*self)(req, res)
}
}

20
src/server/request.rs
View File

@@ -2,7 +2,7 @@
//!
//! These are requests that a `hyper::Server` receives, and include its method,
//! target URI, headers, and message body.
use std::io::{Reader, BufferedReader, IoResult};
use std::io::IoResult;
use std::io::net::ip::SocketAddr;
use {HttpResult};
@@ -13,11 +13,12 @@ use header::common::ContentLength;
use http::{read_request_line};
use http::HttpReader;
use http::HttpReader::{SizedReader, ChunkedReader};
use net::NetworkStream;
use uri::RequestUri;
pub type InternalReader<'a> = &'a mut Reader + 'a;
/// A request bundles several parts of an incoming `NetworkStream`, given to a `Handler`.
pub struct Request {
pub struct Request<'a> {
/// The IP address of the remote connection.
pub remote_addr: SocketAddr,
/// The `Method`, such as `Get`, `Post`, etc.
@@ -28,18 +29,15 @@ pub struct Request {
pub uri: RequestUri,
/// The version of HTTP for this request.
pub version: HttpVersion,
body: HttpReader<BufferedReader<Box<NetworkStream + Send>>>
body: HttpReader<InternalReader<'a>>
}
impl Request {
impl<'a> Request<'a> {
/// Create a new Request, reading the StartLine and Headers so they are
/// immediately useful.
pub fn new<S: NetworkStream>(mut stream: S) -> HttpResult<Request> {
let remote_addr = try!(stream.peer_name());
debug!("remote addr = {}", remote_addr);
let mut stream = BufferedReader::new(box stream as Box<NetworkStream + Send>);
pub fn new(mut stream: InternalReader<'a>, addr: SocketAddr) -> HttpResult<Request<'a>> {
let (method, uri, version) = try!(read_request_line(&mut stream));
debug!("Request Line: {} {} {}", method, uri, version);
let headers = try!(Headers::from_raw(&mut stream));
@@ -57,7 +55,7 @@ impl Request {
};
Ok(Request {
remote_addr: remote_addr,
remote_addr: addr,
method: method,
uri: uri,
headers: headers,
@@ -67,7 +65,7 @@ impl Request {
}
}
impl Reader for Request {
impl<'a> Reader for Request<'a> {
fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
self.body.read(buf)
}

29
src/server/response.rs
View File

@@ -2,7 +2,7 @@
//!
//! These are responses sent by a `hyper::Server` to clients, after
//! receiving a request.
use std::io::{BufferedWriter, IoResult};
use std::io::IoResult;
use time::now_utc;
@@ -11,23 +11,24 @@ use header::common;
use http::{CR, LF, LINE_ENDING, HttpWriter};
use http::HttpWriter::{ThroughWriter, ChunkedWriter, SizedWriter};
use status;
use net::{NetworkStream, Fresh, Streaming};
use net::{Fresh, Streaming};
use version;
pub type InternalWriter<'a> = &'a mut Writer + 'a;
/// The outgoing half for a Tcp connection, created by a `Server` and given to a `Handler`.
pub struct Response<W> {
pub struct Response<'a, W = Fresh> {
/// The HTTP version of this response.
pub version: version::HttpVersion,
// Stream the Response is writing to, not accessible through UnwrittenResponse
body: HttpWriter<BufferedWriter<Box<NetworkStream + Send>>>,
body: HttpWriter<InternalWriter<'a>>,
// The status code for the request.
status: status::StatusCode,
// The outgoing headers on this response.
headers: header::Headers
}
impl<W> Response<W> {
impl<'a, W> Response<'a, W> {
/// The status of this response.
#[inline]
pub fn status(&self) -> status::StatusCode { self.status }
@@ -37,9 +38,9 @@ impl<W> Response<W> {
/// Construct a Response from its constituent parts.
pub fn construct(version: version::HttpVersion,
body: HttpWriter<BufferedWriter<Box<NetworkStream + Send>>>,
body: HttpWriter<InternalWriter<'a>>,
status: status::StatusCode,
headers: header::Headers) -> Response<Fresh> {
headers: header::Headers) -> Response<'a, Fresh> {
Response {
status: status,
version: version,
@@ -49,25 +50,25 @@ impl<W> Response<W> {
}
/// Deconstruct this Response into its constituent parts.
pub fn deconstruct(self) -> (version::HttpVersion, HttpWriter<BufferedWriter<Box<NetworkStream + Send>>>,
pub fn deconstruct(self) -> (version::HttpVersion, HttpWriter<InternalWriter<'a>>,
status::StatusCode, header::Headers) {
(self.version, self.body, self.status, self.headers)
}
}
impl Response<Fresh> {
impl<'a> Response<'a, Fresh> {
/// Creates a new Response that can be used to write to a network stream.
pub fn new<S: NetworkStream>(stream: S) -> Response<Fresh> {
pub fn new(stream: InternalWriter<'a>) -> Response<'a, Fresh> {
Response {
status: status::StatusCode::Ok,
version: version::HttpVersion::Http11,
headers: header::Headers::new(),
body: ThroughWriter(BufferedWriter::new(box stream as Box<NetworkStream + Send>))
body: ThroughWriter(stream)
}
}
/// Consume this Response<Fresh>, writing the Headers and Status and creating a Response<Streaming>
pub fn start(mut self) -> IoResult<Response<Streaming>> {
pub fn start(mut self) -> IoResult<Response<'a, Streaming>> {
debug!("writing head: {} {}", self.version, self.status);
try!(write!(&mut self.body, "{} {}{}{}", self.version, self.status, CR as char, LF as char));
@@ -133,7 +134,7 @@ impl Response<Fresh> {
pub fn headers_mut(&mut self) -> &mut header::Headers { &mut self.headers }
}
impl Response<Streaming> {
impl<'a> Response<'a, Streaming> {
/// Flushes all writing of a response to the client.
pub fn end(self) -> IoResult<()> {
debug!("ending");
@@ -142,7 +143,7 @@ impl Response<Streaming> {
}
}
impl Writer for Response<Streaming> {
impl<'a> Writer for Response<'a, Streaming> {
fn write(&mut self, msg: &[u8]) -> IoResult<()> {
debug!("write {} bytes", msg.len());
self.body.write(msg)