From 3c4cc3aa9170eeb7c43fe1019497580ec0172afd Mon Sep 17 00:00:00 2001 From: Sebastien Boeuf Date: Mon, 12 Jul 2021 17:28:52 +0200 Subject: [PATCH] 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 is ignored for now. Signed-off-by: Sebastien Boeuf --- src/common/mod.rs | 18 ++++++++++++------ src/connection.rs | 11 ++++++----- src/server.rs | 11 ++++++----- 3 files changed, 24 insertions(+), 16 deletions(-) diff --git a/src/common/mod.rs b/src/common/mod.rs index 329a850..6094f51 100644 --- a/src/common/mod.rs +++ b/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)" ); } diff --git a/src/connection.rs b/src/connection.rs index d16e479..b1d0963 100644 --- a/src/connection.rs +++ b/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 { response_buffer: Option>, } -impl HttpConnection { +impl HttpConnection { /// Creates an empty connection. pub fn new(stream: T) -> Self { Self { @@ -126,10 +127,10 @@ impl HttpConnection { } // 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 HttpConnection { let mut response_buffer_vec: Vec = 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); diff --git a/src/server.rs b/src/server.rs index 881d939..c534ca8 100644 --- a/src/server.rs +++ b/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 { in_flight_response_count: u32, } -impl ClientConnection { +impl ClientConnection { fn new(connection: HttpConnection) -> Self { Self { connection, @@ -113,7 +114,7 @@ impl ClientConnection { // 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 ClientConnection { ))); 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 ClientConnection { 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; }