perf(server): try to read from socket at keep-alive

In most situations, this should reduce the number of task wake ups by 1
per request, which can help if reading the request was small.

src/http/conn.rs

@@ -228,6 +228,16 @@ where I: AsyncRead + AsyncWrite,
         }
 
         if !self.io.is_read_blocked() {
+            if self.io.read_buf().is_empty() {
+                match self.io.read_from_io() {
+                    Ok(Async::Ready(_)) => (),
+                    Ok(Async::NotReady) => return,
+                    Err(e) => {
+                        trace!("maybe_notify read_from_io error: {}", e);
+                        self.state.close();
+                    }
+                }
+            }
             if let Some(ref task) = self.state.read_task {
                 task.notify();
             }

src/http/io.rs

@@ -92,9 +92,13 @@ impl<T: AsyncRead + AsyncWrite> Buffered<T> {
         }
     }
 
-    fn read_from_io(&mut self) -> Poll<usize, io::Error> {
+    pub fn read_from_io(&mut self) -> Poll<usize, io::Error> {
         use bytes::BufMut;
-        // TODO: Investigate if we still need these unsafe blocks
+        self.read_blocked = false;
+        //TODO: use io.read_buf(), so we don't have to zero memory
+        //Reason this doesn't use it yet is because benchmarks show the
+        //slightest **decrease** in performance. Switching should be done
+        //when it doesn't cost anything.
         if self.read_buf.remaining_mut() < INIT_BUFFER_SIZE {
             self.read_buf.reserve(INIT_BUFFER_SIZE);
             unsafe { // Zero out unused memory
@@ -103,13 +107,11 @@ impl<T: AsyncRead + AsyncWrite> Buffered<T> {
                 ptr::write_bytes(buf.as_mut_ptr(), 0, len);
             }
         }
-        self.read_blocked = false;
-        unsafe { // Can we use AsyncRead::read_buf instead?
+        unsafe {
             let n = match self.io.read(self.read_buf.bytes_mut()) {
                 Ok(n) => n,
                 Err(e) => {
                     if e.kind() == io::ErrorKind::WouldBlock {
-                        // TODO: Push this out, ideally, into http::Conn.
                         self.read_blocked = true;
                         return Ok(Async::NotReady);
                     }