Expect stream to implement ScmSocket trait

The stream used by the Connection structure is expected to implement
both Read and Write traits. Since we want the server to be able to
receive file descriptors through control messages mechanism, this patch
extends the expectations regarding the stream by adding ScmSocket to the
list of traits.

Since the stream is a UnixStream structure, and since vmm-sys-util
already provides an ScmSocket implementation for UnixStream, extending
the list of traits is very straightforward.

Relying on the newly added trait, the server now reads incoming bytes
through recv_with_fd() function, which replaces the former call to
read(). This change has no intent of modifying the former behavior from
the read(), which is why the returned Option<File> is ignored for now.

Signed-off-by: Sebastien Boeuf <[email protected]>

Author: Sebastien Boeuf

src/common/mod.rs

@@ -117,8 +117,10 @@ pub enum ConnectionError {
     InvalidWrite,
     /// The request parsing has failed.
     ParseError(RequestError),
-    /// Could not perform a stream operation successfully.
-    StreamError(std::io::Error),
+    /// Could not perform a read operation from stream successfully.
+    StreamReadError(vmm_sys_util::errno::Error),
+    /// Could not perform a write operation to stream successfully.
+    StreamWriteError(std::io::Error),
 }
 
 impl Display for ConnectionError {
@@ -127,7 +129,8 @@ impl Display for ConnectionError {
             Self::ConnectionClosed => write!(f, "Connection closed."),
             Self::InvalidWrite => write!(f, "Invalid write attempt."),
             Self::ParseError(inner) => write!(f, "Parsing error: {}", inner),
-            Self::StreamError(inner) => write!(f, "Stream error: {}", inner),
+            Self::StreamReadError(inner) => write!(f, "Reading stream error: {}", inner),
+            Self::StreamWriteError(inner) => write!(f, "Writing stream error: {}", inner),
         }
     }
 }
@@ -322,7 +325,10 @@ mod tests {
             match (self, other) {
                 (ParseError(ref e), ParseError(ref other_e)) => e.eq(other_e),
                 (ConnectionClosed, ConnectionClosed) => true,
-                (StreamError(ref e), StreamError(ref other_e)) => {
+                (StreamReadError(ref e), StreamReadError(ref other_e)) => {
+                    format!("{}", e).eq(&format!("{}", other_e))
+                }
+                (StreamWriteError(ref e), StreamWriteError(ref other_e)) => {
                     format!("{}", e).eq(&format!("{}", other_e))
                 }
                 (InvalidWrite, InvalidWrite) => true,
@@ -489,9 +495,9 @@ mod tests {
         assert_eq!(
             format!(
                 "{}",
-                ConnectionError::StreamError(std::io::Error::from_raw_os_error(11))
+                ConnectionError::StreamWriteError(std::io::Error::from_raw_os_error(11))
             ),
-            "Stream error: Resource temporarily unavailable (os error 11)"
+            "Writing stream error: Resource temporarily unavailable (os error 11)"
         );
     }
 

src/connection.rs

@@ -10,6 +10,7 @@ pub use crate::common::{ConnectionError, HttpHeaderError, RequestError};
 use crate::headers::Headers;
 use crate::request::{find, Request, RequestLine};
 use crate::response::{Response, StatusCode};
+use vmm_sys_util::sock_ctrl_msg::ScmSocket;
 
 const BUFFER_SIZE: usize = 1024;
 
@@ -51,7 +52,7 @@ pub struct HttpConnection<T> {
     response_buffer: Option<Vec<u8>>,
 }
 
-impl<T: Read + Write> HttpConnection<T> {
+impl<T: Read + Write + ScmSocket> HttpConnection<T> {
     /// Creates an empty connection.
     pub fn new(stream: T) -> Self {
         Self {
@@ -126,10 +127,10 @@ impl<T: Read + Write> HttpConnection<T> {
         }
         // Append new bytes to what we already have in the buffer.
         // The slice access is safe, the index is checked above.
-        let bytes_read = self
+        let (bytes_read, _) = self
             .stream
-            .read(&mut self.buffer[self.read_cursor..])
-            .map_err(ConnectionError::StreamError)?;
+            .recv_with_fd(&mut self.buffer[self.read_cursor..])
+            .map_err(ConnectionError::StreamReadError)?;
 
         // If the read returned 0 then the client has closed the connection.
         if bytes_read == 0 {
@@ -392,7 +393,7 @@ impl<T: Read + Write> HttpConnection<T> {
                 let mut response_buffer_vec: Vec<u8> = Vec::new();
                 response
                     .write_all(&mut response_buffer_vec)
-                    .map_err(ConnectionError::StreamError)?;
+                    .map_err(ConnectionError::StreamWriteError)?;
                 self.response_buffer = Some(response_buffer_vec);
             } else {
                 return Err(ConnectionError::InvalidWrite);

src/server.rs

@@ -1,6 +1,7 @@
 // Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::collections::HashMap;
 use std::io::{Read, Write};
 use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
 use std::os::unix::net::{UnixListener, UnixStream};
@@ -11,7 +12,7 @@ pub use crate::common::{ConnectionError, RequestError, ServerError};
 use crate::connection::HttpConnection;
 use crate::request::Request;
 use crate::response::{Response, StatusCode};
-use std::collections::HashMap;
+use vmm_sys_util::sock_ctrl_msg::ScmSocket;
 
 use vmm_sys_util::epoll;
 
@@ -92,7 +93,7 @@ struct ClientConnection<T> {
     in_flight_response_count: u32,
 }
 
-impl<T: Read + Write> ClientConnection<T> {
+impl<T: Read + Write + ScmSocket> ClientConnection<T> {
     fn new(connection: HttpConnection<T>) -> Self {
         Self {
             connection,
@@ -113,7 +114,7 @@ impl<T: Read + Write> ClientConnection<T> {
                 // safe to drop.
                 return Ok(vec![]);
             }
-            Err(ConnectionError::StreamError(inner)) => {
+            Err(ConnectionError::StreamReadError(inner)) => {
                 // Reading from the connection failed.
                 // We should try to write an error message regardless.
                 let mut internal_error_response =
@@ -134,7 +135,7 @@ impl<T: Read + Write> ClientConnection<T> {
                 )));
                 self.connection.enqueue_response(error_response);
             }
-            Err(ConnectionError::InvalidWrite) => {
+            Err(ConnectionError::InvalidWrite) | Err(ConnectionError::StreamWriteError(_)) => {
                 // This is unreachable because `HttpConnection::try_read()` cannot return this error variant.
                 unreachable!();
             }
@@ -161,7 +162,7 @@ impl<T: Read + Write> ClientConnection<T> {
     fn write(&mut self) -> Result<()> {
         // The stream is available for writing.
         match self.connection.try_write() {
-            Err(ConnectionError::ConnectionClosed) | Err(ConnectionError::StreamError(_)) => {
+            Err(ConnectionError::ConnectionClosed) | Err(ConnectionError::StreamWriteError(_)) => {
                 // Writing to the stream failed so it will be removed.
                 self.state = ClientConnectionState::Closed;
             }