use std::fs::File;
use std::io::IoSliceMut;
use std::mem;
use base::AsRawDescriptor;
use base::RawDescriptor;
use zerocopy::AsBytes;
use zerocopy::FromBytes;
use crate::connection::Req;
use crate::message::FrontendReq;
use crate::message::*;
use crate::sys::PlatformConnection;
use crate::Error;
use crate::Result;
pub trait Listener: Sized {
fn accept(&mut self) -> Result<Option<Connection<FrontendReq>>>;
fn set_nonblocking(&self, block: bool) -> Result<()>;
}
fn advance_slices_mut(bufs: &mut &mut [&mut [u8]], mut count: usize) {
use std::mem::take;
let mut idx = 0;
for b in bufs.iter() {
if count < b.len() {
break;
}
count -= b.len();
idx += 1;
}
*bufs = &mut take(bufs)[idx..];
if !bufs.is_empty() {
let slice = take(&mut bufs[0]);
let (_, remaining) = slice.split_at_mut(count);
bufs[0] = remaining;
}
}
pub struct Connection<R: Req>(
pub(crate) PlatformConnection,
pub(crate) std::marker::PhantomData<R>,
pub(crate) std::marker::PhantomData<std::cell::Cell<()>>,
);
impl<R: Req> Connection<R> {
pub fn send_header_only_message(
&self,
hdr: &VhostUserMsgHeader<R>,
fds: Option<&[RawDescriptor]>,
) -> Result<()> {
self.0.send_message(hdr.as_bytes(), &[], &[], fds)
}
pub fn send_message<T: AsBytes>(
&self,
hdr: &VhostUserMsgHeader<R>,
body: &T,
fds: Option<&[RawDescriptor]>,
) -> Result<()> {
self.0
.send_message(hdr.as_bytes(), body.as_bytes(), &[], fds)
}
pub fn send_message_with_payload<T: Sized + AsBytes>(
&self,
hdr: &VhostUserMsgHeader<R>,
body: &T,
payload: &[u8],
fds: Option<&[RawDescriptor]>,
) -> Result<()> {
self.0
.send_message(hdr.as_bytes(), body.as_bytes(), payload, fds)
}
fn recv_into_bufs_all(&self, mut bufs: &mut [&mut [u8]]) -> Result<Vec<File>> {
let mut first_read = true;
let mut rfds = Vec::new();
advance_slices_mut(&mut bufs, 0);
while !bufs.is_empty() {
let mut slices: Vec<IoSliceMut> = bufs.iter_mut().map(|b| IoSliceMut::new(b)).collect();
let res = self.0.recv_into_bufs(&mut slices, true);
match res {
Ok((0, _)) => return Err(Error::PartialMessage),
Ok((n, fds)) => {
if first_read {
first_read = false;
if let Some(fds) = fds {
rfds = fds;
}
}
advance_slices_mut(&mut bufs, n);
}
Err(e) => match e {
Error::SocketRetry(_) => {}
_ => return Err(e),
},
}
}
Ok(rfds)
}
pub fn recv_header(&self) -> Result<(VhostUserMsgHeader<R>, Vec<File>)> {
let mut hdr = VhostUserMsgHeader::default();
let files = self.recv_into_bufs_all(&mut [hdr.as_bytes_mut()])?;
if !hdr.is_valid() {
return Err(Error::InvalidMessage);
}
Ok((hdr, files))
}
pub fn recv_body_bytes(&self, hdr: &VhostUserMsgHeader<R>) -> Result<Vec<u8>> {
let mut body = vec![0; hdr.get_size().try_into().unwrap()];
let files = self.recv_into_bufs_all(&mut [&mut body[..]])?;
if !files.is_empty() {
return Err(Error::InvalidMessage);
}
Ok(body)
}
pub fn recv_message<T: AsBytes + FromBytes + VhostUserMsgValidator>(
&self,
) -> Result<(VhostUserMsgHeader<R>, T, Vec<File>)> {
let mut hdr = VhostUserMsgHeader::default();
let mut body = T::new_zeroed();
let mut slices = [hdr.as_bytes_mut(), body.as_bytes_mut()];
let files = self.recv_into_bufs_all(&mut slices)?;
if !hdr.is_valid() || !body.is_valid() {
return Err(Error::InvalidMessage);
}
Ok((hdr, body, files))
}
pub fn recv_message_with_payload<T: AsBytes + FromBytes + VhostUserMsgValidator>(
&self,
) -> Result<(VhostUserMsgHeader<R>, T, Vec<u8>, Vec<File>)> {
let (hdr, files) = self.recv_header()?;
let mut body = T::new_zeroed();
let payload_size = hdr.get_size() as usize - mem::size_of::<T>();
let mut buf: Vec<u8> = vec![0; payload_size];
let mut slices = [body.as_bytes_mut(), buf.as_bytes_mut()];
let more_files = self.recv_into_bufs_all(&mut slices)?;
if !body.is_valid() || !more_files.is_empty() {
return Err(Error::InvalidMessage);
}
Ok((hdr, body, buf, files))
}
}
impl<R: Req> AsRawDescriptor for Connection<R> {
fn as_raw_descriptor(&self) -> RawDescriptor {
self.0.as_raw_descriptor()
}
}
#[cfg(test)]
pub(crate) mod tests {
use std::io::Read;
use std::io::Seek;
use std::io::SeekFrom;
use std::io::Write;
use tempfile::tempfile;
use super::*;
use crate::message::VhostUserEmptyMessage;
use crate::message::VhostUserU64;
#[test]
fn send_header_only() {
let (client_connection, server_connection) = Connection::pair().unwrap();
let hdr1 = VhostUserMsgHeader::new(FrontendReq::GET_FEATURES, 0, 0);
client_connection
.send_header_only_message(&hdr1, None)
.unwrap();
let (hdr2, _, files) = server_connection
.recv_message::<VhostUserEmptyMessage>()
.unwrap();
assert_eq!(hdr1, hdr2);
assert!(files.is_empty());
}
#[test]
fn send_data() {
let (client_connection, server_connection) = Connection::pair().unwrap();
let hdr1 = VhostUserMsgHeader::new(FrontendReq::SET_FEATURES, 0, 8);
client_connection
.send_message(&hdr1, &VhostUserU64::new(0xf00dbeefdeadf00d), None)
.unwrap();
let (hdr2, body, files) = server_connection.recv_message::<VhostUserU64>().unwrap();
assert_eq!(hdr1, hdr2);
let value = body.value;
assert_eq!(value, 0xf00dbeefdeadf00d);
assert!(files.is_empty());
}
#[test]
fn send_fd() {
let (client_connection, server_connection) = Connection::pair().unwrap();
let mut fd = tempfile().unwrap();
write!(fd, "test").unwrap();
let hdr1 = VhostUserMsgHeader::new(FrontendReq::SET_MEM_TABLE, 0, 0);
client_connection
.send_header_only_message(&hdr1, Some(&[fd.as_raw_descriptor()]))
.unwrap();
let (hdr2, _, files) = server_connection
.recv_message::<VhostUserEmptyMessage>()
.unwrap();
assert_eq!(hdr1, hdr2);
assert_eq!(files.len(), 1);
let mut file = &files[0];
let mut content = String::new();
file.seek(SeekFrom::Start(0)).unwrap();
file.read_to_string(&mut content).unwrap();
assert_eq!(content, "test");
}
#[test]
fn test_advance_slices_mut() {
let mut buf1 = [1; 8];
let mut buf2 = [2; 16];
let mut buf3 = [3; 8];
let mut bufs = &mut [&mut buf1[..], &mut buf2[..], &mut buf3[..]][..];
advance_slices_mut(&mut bufs, 10);
assert_eq!(bufs[0], [2; 14].as_ref());
assert_eq!(bufs[1], [3; 8].as_ref());
}
}