unix only test

move reuse_address before bind

CHANGES.md

@@ -1,5 +1,14 @@
 # Changes
 
+## 0.4.3 (2018-03-03)
+
+* Fix request body read bug
+
+* Fix segmentation fault #79
+
+* Set reuse address before bind #90
+
+
 ## 0.4.2 (2018-03-02)
 
 * Better naming for websockets implementation

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-web"
-version = "0.4.2"
+version = "0.4.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix web is a small, pragmatic, extremely fast, web framework for Rust."
 readme = "README.md"

src/client/parser.rs

@@ -2,7 +2,7 @@ use std::mem;
 use httparse;
 use http::{Version, HttpTryFrom, HeaderMap, StatusCode};
 use http::header::{self, HeaderName, HeaderValue};
-use bytes::{Bytes, BytesMut, BufMut};
+use bytes::{Bytes, BytesMut};
 use futures::{Poll, Async};
 
 use error::{ParseError, PayloadError};
@@ -37,7 +37,7 @@ impl HttpResponseParser {
         where T: IoStream
     {
         // if buf is empty parse_message will always return NotReady, let's avoid that
-        let read = if buf.is_empty() {
+        if buf.is_empty() {
             match utils::read_from_io(io, buf) {
                 Ok(Async::Ready(0)) =>
                     return Err(HttpResponseParserError::Disconnect),
@@ -47,13 +47,12 @@ impl HttpResponseParser {
                 Err(err) =>
                     return Err(HttpResponseParserError::Error(err.into()))
             }
-            false
+        }
-        } else {
-            true
-        };
 
         loop {
-            match HttpResponseParser::parse_message(buf).map_err(HttpResponseParserError::Error)? {
+            match HttpResponseParser::parse_message(buf)
+                .map_err(HttpResponseParserError::Error)?
+            {
                 Async::Ready((msg, decoder)) => {
                     self.decoder = decoder;
                     return Ok(Async::Ready(msg));
@@ -62,7 +61,6 @@ impl HttpResponseParser {
                     if buf.capacity() >= MAX_BUFFER_SIZE {
                         return Err(HttpResponseParserError::Error(ParseError::TooLarge));
                     }
-                    if read || buf.remaining_mut() == 0 {
                     match utils::read_from_io(io, buf) {
                         Ok(Async::Ready(0)) =>
                             return Err(HttpResponseParserError::Disconnect),
@@ -71,9 +69,6 @@ impl HttpResponseParser {
                         Err(err) =>
                             return Err(HttpResponseParserError::Error(err.into())),
                     }
-                    } else {
-                        return Ok(Async::NotReady)
-                    }
                 },
             }
         }
@@ -84,25 +79,34 @@ impl HttpResponseParser {
         where T: IoStream
     {
         if self.decoder.is_some() {
+            loop {
                 // read payload
-            match utils::read_from_io(io, buf) {
+                let not_ready = match utils::read_from_io(io, buf) {
                     Ok(Async::Ready(0)) => {
                         if buf.is_empty() {
                             return Err(PayloadError::Incomplete)
                         }
+                        true
                     }
                     Err(err) => return Err(err.into()),
-                _ => (),
+                    Ok(Async::NotReady) => true,
-            }
+                    _ => false,
+                };
 
                 match self.decoder.as_mut().unwrap().decode(buf) {
-                Ok(Async::Ready(Some(b))) => Ok(Async::Ready(Some(b))),
+                    Ok(Async::Ready(Some(b))) =>
+                        return Ok(Async::Ready(Some(b))),
                     Ok(Async::Ready(None)) => {
                         self.decoder.take();
-                    Ok(Async::Ready(None))
+                        return Ok(Async::Ready(None))
+                    }
+                    Ok(Async::NotReady) => {
+                        if not_ready {
+                            return Ok(Async::NotReady)
+                        }
+                    }
+                    Err(err) => return Err(err.into()),
                 }
-                Ok(Async::NotReady) => Ok(Async::NotReady),
-                Err(err) => Err(err.into()),
             }
         } else {
             Ok(Async::Ready(None))

src/server/channel.rs

@@ -18,15 +18,6 @@ enum HttpProtocol<T: IoStream, H: 'static> {
     Unknown(Rc<WorkerSettings<H>>, Option<SocketAddr>, T, BytesMut),
 }
 
-impl<T: IoStream, H: 'static> HttpProtocol<T, H> {
-    fn is_unknown(&self) -> bool {
-        match *self {
-            HttpProtocol::Unknown(_, _, _, _) => true,
-            _ => false
-        }
-    }
-}
-
 enum ProtocolKind {
     Http1,
     Http2,
@@ -44,15 +35,14 @@ impl<T, H> HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'static
                       io: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H>
     {
         settings.add_channel();
+
         if http2 {
             HttpChannel {
-                node: None,
+                node: None, proto: Some(HttpProtocol::H2(
-                proto: Some(HttpProtocol::H2(
                     h2::Http2::new(settings, io, peer, Bytes::new()))) }
         } else {
             HttpChannel {
-                node: None,
+                node: None, proto: Some(HttpProtocol::Unknown(
-                proto: Some(HttpProtocol::Unknown(
                     settings, peer, io, BytesMut::with_capacity(4096))) }
         }
     }
@@ -78,15 +68,18 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
     type Error = ();
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        if !self.proto.as_ref().map(|p| p.is_unknown()).unwrap_or(false) && self.node.is_none() {
+        if !self.node.is_none() {
-            self.node = Some(Node::new(self));
+            let el = self as *mut _;
-            match self.proto {
+            self.node = Some(Node::new(el));
+            let _ = match self.proto {
                 Some(HttpProtocol::H1(ref mut h1)) =>
-                    h1.settings().head().insert(self.node.as_ref().unwrap()),
+                    self.node.as_ref().map(|n| h1.settings().head().insert(n)),
                 Some(HttpProtocol::H2(ref mut h2)) =>
-                    h2.settings().head().insert(self.node.as_ref().unwrap()),
+                    self.node.as_ref().map(|n| h2.settings().head().insert(n)),
-                _ => (),
+                Some(HttpProtocol::Unknown(ref mut settings, _, _, _)) =>
-            }
+                    self.node.as_ref().map(|n| settings.head().insert(n)),
+                None => unreachable!(),
+            };
         }
 
         let kind = match self.proto {
@@ -95,7 +88,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
                 match result {
                     Ok(Async::Ready(())) | Err(_) => {
                         h1.settings().remove_channel();
-                        self.node.as_ref().unwrap().remove();
+                        self.node.as_mut().map(|n| n.remove());
                     },
                     _ => (),
                 }
@@ -106,7 +99,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
                 match result {
                     Ok(Async::Ready(())) | Err(_) => {
                         h2.settings().remove_channel();
-                        self.node.as_ref().unwrap().remove();
+                        self.node.as_mut().map(|n| n.remove());
                     },
                     _ => (),
                 }
@@ -117,6 +110,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
                     Ok(Async::Ready(0)) | Err(_) => {
                         debug!("Ignored premature client disconnection");
                         settings.remove_channel();
+                        self.node.as_mut().map(|n| n.remove());
                         return Err(())
                     },
                     _ => (),
@@ -163,11 +157,11 @@ pub(crate) struct Node<T>
 
 impl<T> Node<T>
 {
-    fn new(el: &mut T) -> Self {
+    fn new(el: *mut T) -> Self {
         Node {
             next: None,
             prev: None,
-            element: el as *mut _,
+            element: el,
         }
     }
 
@@ -186,13 +180,14 @@ impl<T> Node<T>
         }
     }
 
-    fn remove(&self) {
+    fn remove(&mut self) {
-        #[allow(mutable_transmutes)]
         unsafe {
-            if let Some(ref prev) = self.prev {
+            self.element = ptr::null_mut();
-                let p: &mut Node<()> = mem::transmute(prev.as_ref().unwrap());
+            let next = self.next.take();
-                let slf: &mut Node<T> = mem::transmute(self);
+            let mut prev = self.prev.take();
-                p.next = slf.next.take();
+
+            if let Some(ref mut prev) = prev {
+                prev.as_mut().unwrap().next = next;
             }
         }
     }

src/server/h1.rs

@@ -10,7 +10,7 @@ use actix::Arbiter;
 use httparse;
 use http::{Uri, Method, Version, HttpTryFrom, HeaderMap};
 use http::header::{self, HeaderName, HeaderValue};
-use bytes::{Bytes, BytesMut, BufMut};
+use bytes::{Bytes, BytesMut};
 use futures::{Future, Poll, Async};
 use tokio_core::reactor::Timeout;
 
@@ -402,42 +402,51 @@ impl Reader {
         // read payload
         let done = {
             if let Some(ref mut payload) = self.payload {
-                match utils::read_from_io(io, buf) {
+                'buf: loop {
+                    let not_ready = match utils::read_from_io(io, buf) {
                         Ok(Async::Ready(0)) => {
                             payload.tx.set_error(PayloadError::Incomplete);
 
                             // http channel should not deal with payload errors
                             return Err(ReaderError::Payload)
                         },
+                        Ok(Async::NotReady) => true,
                         Err(err) => {
                             payload.tx.set_error(err.into());
 
                             // http channel should not deal with payload errors
                             return Err(ReaderError::Payload)
                         }
-                    _ => (),
+                        _ => false,
-                }
+                    };
                     loop {
                         match payload.decoder.decode(buf) {
                             Ok(Async::Ready(Some(bytes))) => {
                                 payload.tx.feed_data(bytes);
                                 if payload.decoder.is_eof() {
                                     payload.tx.feed_eof();
-                                break true
+                                    break 'buf true
                                 }
                             },
                             Ok(Async::Ready(None)) => {
                                 payload.tx.feed_eof();
-                            break true
+                                break 'buf true
+                            },
+                            Ok(Async::NotReady) => {
+                                // if buffer is full then
+                                // socket still can contain more data
+                                if not_ready {
+                                    return Ok(Async::NotReady)
+                                }
+                                continue 'buf
                             },
-                        Ok(Async::NotReady) =>
-                            return Ok(Async::NotReady),
                             Err(err) => {
                                 payload.tx.set_error(err.into());
                                 return Err(ReaderError::Payload)
                             }
                         }
                     }
+                }
             } else {
                 false
             }
@@ -445,16 +454,13 @@ impl Reader {
         if done { self.payload = None }
 
         // if buf is empty parse_message will always return NotReady, let's avoid that
-        let read = if buf.is_empty() {
+        if buf.is_empty() {
             match utils::read_from_io(io, buf) {
                 Ok(Async::Ready(0)) => return Err(ReaderError::Disconnect),
                 Ok(Async::Ready(_)) => (),
                 Ok(Async::NotReady) => return Ok(Async::NotReady),
                 Err(err) => return Err(ReaderError::Error(err.into()))
             }
-            false
-        } else {
-            true
         };
 
         loop {
@@ -470,11 +476,10 @@ impl Reader {
                     return Ok(Async::Ready(msg));
                 },
                 Async::NotReady => {
-                    if buf.capacity() >= MAX_BUFFER_SIZE {
+                    if buf.len() >= MAX_BUFFER_SIZE {
                         error!("MAX_BUFFER_SIZE unprocessed data reached, closing");
                         return Err(ReaderError::Error(ParseError::TooLarge));
                     }
-                    if read || buf.remaining_mut() == 0 {
                     match utils::read_from_io(io, buf) {
                         Ok(Async::Ready(0)) => {
                             debug!("Ignored premature client disconnection");
@@ -484,9 +489,6 @@ impl Reader {
                         Ok(Async::NotReady) => return Ok(Async::NotReady),
                         Err(err) => return Err(ReaderError::Error(err.into())),
                     }
-                    } else {
-                        return Ok(Async::NotReady)
-                    }
                 },
             }
         }

src/server/settings.rs

@@ -63,7 +63,7 @@ pub(crate) struct WorkerSettings<H> {
     bytes: Rc<SharedBytesPool>,
     messages: Rc<helpers::SharedMessagePool>,
     channels: Cell<usize>,
-    node: Node<()>,
+    node: Box<Node<()>>,
 }
 
 impl<H> WorkerSettings<H> {
@@ -75,7 +75,7 @@ impl<H> WorkerSettings<H> {
             bytes: Rc::new(SharedBytesPool::new()),
             messages: Rc::new(helpers::SharedMessagePool::new()),
             channels: Cell::new(0),
-            node: Node::head(),
+            node: Box::new(Node::head()),
         }
     }
 

src/server/srv.rs

@@ -697,7 +697,7 @@ fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::T
         net::SocketAddr::V4(_) => TcpBuilder::new_v4()?,
         net::SocketAddr::V6(_) => TcpBuilder::new_v6()?,
     };
-    builder.bind(addr)?;
     builder.reuse_address(true)?;
+    builder.bind(addr)?;
     Ok(builder.listen(backlog)?)
 }

tests/test_server.rs

@@ -94,6 +94,7 @@ fn test_start() {
 }
 
 #[test]
+#[cfg(unix)]
 fn test_shutdown() {
     let _ = test::TestServer::unused_addr();
     let (tx, rx) = mpsc::channel();