pxcloud/vhost-device
5
0
Fork 0
mirror of https://github.com/rust-vmm/vhost-device.git synced 2026-01-08 20:57:35 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

vsock: remove "vsock_" prefix where is redundant

Browse Source 
In several places the "vsock_" prefix doesn't add much and makes
names longer, let's remove it where it's not needed.

Suggested-by: Viresh Kumar <viresh.kumar@linaro.org>
Signed-off-by: Stefano Garzarella <sgarzare@redhat.com>
This commit is contained in:
Stefano Garzarella 2022-10-12 13:09:42 +02:00
parent c452d1c2f9
commit a76f9ebe16
5 changed files with 184 additions and 195 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
vsock/src/main.rs
View File

@ -48,33 +48,33 @@ impl TryFrom<VsockArgs> for VsockConfig {
/// This is the public API through which an external program starts the
/// vhost-user-vsock backend server.
pub(crate) fn start_backend_server(vsock_config: VsockConfig) {
pub(crate) fn start_backend_server(config: VsockConfig) {
loop {
let vsock_backend = Arc::new(RwLock::new(
VhostUserVsockBackend::new(vsock_config.clone()).unwrap(),
let backend = Arc::new(RwLock::new(
VhostUserVsockBackend::new(config.clone()).unwrap(),
));
let listener = Listener::new(vsock_config.get_socket_path(), true).unwrap();
let listener = Listener::new(config.get_socket_path(), true).unwrap();
let mut vsock_daemon = VhostUserDaemon::new(
let mut daemon = VhostUserDaemon::new(
String::from("vhost-user-vsock"),
vsock_backend.clone(),
backend.clone(),
GuestMemoryAtomic::new(GuestMemoryMmap::new()),
)
.unwrap();
let mut vring_workers = vsock_daemon.get_epoll_handlers();
let mut vring_workers = daemon.get_epoll_handlers();
for thread in vsock_backend.read().unwrap().threads.iter() {
for thread in backend.read().unwrap().threads.iter() {
thread
.lock()
.unwrap()
.set_vring_worker(Some(vring_workers.remove(0)));
}
vsock_daemon.start(listener).unwrap();
daemon.start(listener).unwrap();
match vsock_daemon.wait() {
match daemon.wait() {
Ok(()) => {
info!("Stopping cleanly");
}
@ -87,15 +87,15 @@ pub(crate) fn start_backend_server(vsock_config: VsockConfig) {
}
// No matter the result, we need to shut down the worker thread.
vsock_backend.read().unwrap().exit_event.write(1).unwrap();
backend.read().unwrap().exit_event.write(1).unwrap();
}
}
fn main() {
env_logger::init();
let vsock_config = VsockConfig::try_from(VsockArgs::parse()).unwrap();
start_backend_server(vsock_config);
let config = VsockConfig::try_from(VsockArgs::parse()).unwrap();
start_backend_server(config);
}
#[cfg(test)]
@ -114,15 +114,15 @@ mod tests {
#[test]
fn test_vsock_config_setup() {
let vsock_args = VsockArgs::from_args(3, "/tmp/vhost4.socket", "/tmp/vm4.vsock");
let args = VsockArgs::from_args(3, "/tmp/vhost4.socket", "/tmp/vm4.vsock");
let vsock_config = VsockConfig::try_from(vsock_args);
assert!(vsock_config.is_ok());
let config = VsockConfig::try_from(args);
assert!(config.is_ok());
let vsock_config = vsock_config.unwrap();
assert_eq!(vsock_config.get_guest_cid(), 3);
assert_eq!(vsock_config.get_socket_path(), "/tmp/vhost4.socket");
assert_eq!(vsock_config.get_uds_path(), "/tmp/vm4.vsock");
let config = config.unwrap();
assert_eq!(config.get_guest_cid(), 3);
assert_eq!(config.get_socket_path(), "/tmp/vhost4.socket");
assert_eq!(config.get_uds_path(), "/tmp/vm4.vsock");
}
#[test]
@ -131,31 +131,26 @@ mod tests {
const VHOST_SOCKET_PATH: &str = "test_vsock_server.socket";
const VSOCK_SOCKET_PATH: &str = "test_vsock_server.vsock";
let vsock_config = VsockConfig::new(
let config = VsockConfig::new(
CID,
VHOST_SOCKET_PATH.to_string(),
VSOCK_SOCKET_PATH.to_string(),
);
let vsock_backend = Arc::new(RwLock::new(
VhostUserVsockBackend::new(vsock_config).unwrap(),
));
let backend = Arc::new(RwLock::new(VhostUserVsockBackend::new(config).unwrap()));
let vsock_daemon = VhostUserDaemon::new(
let daemon = VhostUserDaemon::new(
String::from("vhost-user-vsock"),
vsock_backend.clone(),
backend.clone(),
GuestMemoryAtomic::new(GuestMemoryMmap::new()),
)
.unwrap();
let vring_workers = vsock_daemon.get_epoll_handlers();
let vring_workers = daemon.get_epoll_handlers();
// VhostUserVsockBackend support a single thread that handles the TX and RX queues
assert_eq!(vsock_backend.read().unwrap().threads.len(), 1);
assert_eq!(backend.read().unwrap().threads.len(), 1);
assert_eq!(
vring_workers.len(),
vsock_backend.read().unwrap().threads.len()
);
assert_eq!(vring_workers.len(), backend.read().unwrap().threads.len());
}
}

4
vsock/src/thread_backend.rs
View File

@ -206,7 +206,7 @@ impl VsockThreadBackend {
self.listener_map
.insert(stream_fd, ConnMapKey::new(pkt.dst_port(), pkt.src_port()));
let vsock_conn = VsockConnection::new_peer_init(
let conn = VsockConnection::new_peer_init(
stream,
pkt.dst_cid(),
pkt.dst_port(),
@ -217,7 +217,7 @@ impl VsockThreadBackend {
);
self.conn_map
.insert(ConnMapKey::new(pkt.dst_port(), pkt.src_port()), vsock_conn);
.insert(ConnMapKey::new(pkt.dst_port(), pkt.src_port()), conn);
self.backend_rxq
.push_back(ConnMapKey::new(pkt.dst_port(), pkt.src_port()));
self.stream_map

60
vsock/src/vhu_vsock.rs
View File

@ -200,16 +200,16 @@ pub(crate) struct VhostUserVsockBackend {
}
impl VhostUserVsockBackend {
pub fn new(vsock_config: VsockConfig) -> Result<Self> {
pub fn new(config: VsockConfig) -> Result<Self> {
let thread = Mutex::new(VhostUserVsockThread::new(
vsock_config.get_uds_path(),
vsock_config.get_guest_cid(),
config.get_uds_path(),
config.get_guest_cid(),
)?);
let queues_per_thread = vec![QUEUE_MASK];
Ok(Self {
config: VirtioVsockConfig {
guest_cid: From::from(vsock_config.get_guest_cid()),
guest_cid: From::from(config.get_guest_cid()),
},
threads: vec![thread],
queues_per_thread,
@ -328,22 +328,22 @@ mod tests {
const VHOST_SOCKET_PATH: &str = "test_vsock_backend.socket";
const VSOCK_SOCKET_PATH: &str = "test_vsock_backend.vsock";
let vsock_config = VsockConfig::new(
let config = VsockConfig::new(
CID,
VHOST_SOCKET_PATH.to_string(),
VSOCK_SOCKET_PATH.to_string(),
);
let vsock_backend = VhostUserVsockBackend::new(vsock_config);
let backend = VhostUserVsockBackend::new(config);
assert!(vsock_backend.is_ok());
let mut vsock_backend = vsock_backend.unwrap();
assert!(backend.is_ok());
let mut backend = backend.unwrap();
assert_eq!(vsock_backend.num_queues(), NUM_QUEUES);
assert_eq!(vsock_backend.max_queue_size(), QUEUE_SIZE);
assert_ne!(vsock_backend.features(), 0);
assert!(!vsock_backend.protocol_features().is_empty());
vsock_backend.set_event_idx(false);
assert_eq!(backend.num_queues(), NUM_QUEUES);
assert_eq!(backend.max_queue_size(), QUEUE_SIZE);
assert_ne!(backend.features(), 0);
assert!(!backend.protocol_features().is_empty());
backend.set_event_idx(false);
let mem = GuestMemoryAtomic::new(
GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x10000)]).unwrap(),
@ -353,34 +353,34 @@ mod tests {
VringRwLock::new(mem.clone(), 0x2000).unwrap(),
];
assert!(vsock_backend.update_memory(mem).is_ok());
assert!(backend.update_memory(mem).is_ok());
let queues_per_thread = vsock_backend.queues_per_thread();
let queues_per_thread = backend.queues_per_thread();
assert_eq!(queues_per_thread.len(), 1);
assert_eq!(queues_per_thread[0], 0b11);
let config = vsock_backend.get_config(0, 8);
let config = backend.get_config(0, 8);
assert_eq!(config.len(), 8);
let cid = u64::from_le_bytes(config.try_into().unwrap());
assert_eq!(cid, CID);
let exit = vsock_backend.exit_event(0);
let exit = backend.exit_event(0);
assert!(exit.is_some());
exit.unwrap().write(1).unwrap();
let ret = vsock_backend.handle_event(RX_QUEUE_EVENT, EventSet::IN, &vrings, 0);
let ret = backend.handle_event(RX_QUEUE_EVENT, EventSet::IN, &vrings, 0);
assert!(ret.is_ok());
assert!(!ret.unwrap());
let ret = vsock_backend.handle_event(TX_QUEUE_EVENT, EventSet::IN, &vrings, 0);
let ret = backend.handle_event(TX_QUEUE_EVENT, EventSet::IN, &vrings, 0);
assert!(ret.is_ok());
assert!(!ret.unwrap());
let ret = vsock_backend.handle_event(EVT_QUEUE_EVENT, EventSet::IN, &vrings, 0);
let ret = backend.handle_event(EVT_QUEUE_EVENT, EventSet::IN, &vrings, 0);
assert!(ret.is_ok());
assert!(!ret.unwrap());
let ret = vsock_backend.handle_event(BACKEND_EVENT, EventSet::IN, &vrings, 0);
let ret = backend.handle_event(BACKEND_EVENT, EventSet::IN, &vrings, 0);
assert!(ret.is_ok());
assert!(!ret.unwrap());
@ -395,22 +395,22 @@ mod tests {
const VHOST_SOCKET_PATH: &str = "test_vsock_backend_failures.socket";
const VSOCK_SOCKET_PATH: &str = "test_vsock_backend_failures.vsock";
let vsock_config = VsockConfig::new(
let config = VsockConfig::new(
CID,
"/sys/not_allowed.socket".to_string(),
"/sys/not_allowed.vsock".to_string(),
);
let vsock_backend = VhostUserVsockBackend::new(vsock_config);
assert!(vsock_backend.is_err());
let backend = VhostUserVsockBackend::new(config);
assert!(backend.is_err());
let vsock_config = VsockConfig::new(
let config = VsockConfig::new(
CID,
VHOST_SOCKET_PATH.to_string(),
VSOCK_SOCKET_PATH.to_string(),
);
let mut vsock_backend = VhostUserVsockBackend::new(vsock_config).unwrap();
let mut backend = VhostUserVsockBackend::new(config).unwrap();
let mem = GuestMemoryAtomic::new(
GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x10000)]).unwrap(),
);
@ -419,21 +419,21 @@ mod tests {
VringRwLock::new(mem.clone(), 0x2000).unwrap(),
];
vsock_backend.update_memory(mem).unwrap();
backend.update_memory(mem).unwrap();
// reading out of the config space, expecting empty config
let config = vsock_backend.get_config(2, 8);
let config = backend.get_config(2, 8);
assert_eq!(config.len(), 0);
assert_eq!(
vsock_backend
backend
.handle_event(RX_QUEUE_EVENT, EventSet::OUT, &vrings, 0)
.unwrap_err()
.to_string(),
Error::HandleEventNotEpollIn.to_string()
);
assert_eq!(
vsock_backend
backend
.handle_event(BACKEND_EVENT + 1, EventSet::IN, &vrings, 0)
.unwrap_err()
.to_string(),

29
vsock/src/vhu_vsock_thread.rs
View File

@ -236,7 +236,7 @@ impl VhostUserVsockThread {
// Create a new connection object an enqueue a connection request
// packet to be sent to the guest
let conn_map_key = ConnMapKey::new(local_port, peer_port);
let mut new_vsock_conn = VsockConnection::new_local_init(
let mut new_conn = VsockConnection::new_local_init(
unix_stream,
VSOCK_HOST_CID,
local_port,
@ -244,13 +244,11 @@ impl VhostUserVsockThread {
peer_port,
self.get_epoll_fd(),
);
new_vsock_conn.rx_queue.enqueue(RxOps::Request);
new_vsock_conn.set_peer_port(peer_port);
new_conn.rx_queue.enqueue(RxOps::Request);
new_conn.set_peer_port(peer_port);
// Add connection object into the backend's maps
self.thread_backend
.conn_map
.insert(conn_map_key, new_vsock_conn);
self.thread_backend.conn_map.insert(conn_map_key, new_conn);
self.thread_backend
.backend_rxq
@ -266,20 +264,19 @@ impl VhostUserVsockThread {
} else {
// Previously connected connection
let key = self.thread_backend.listener_map.get(&fd).unwrap();
let vsock_conn = self.thread_backend.conn_map.get_mut(key).unwrap();
let conn = self.thread_backend.conn_map.get_mut(key).unwrap();
if evset == epoll::Events::EPOLLOUT {
// Flush any remaining data from the tx buffer
match vsock_conn.tx_buf.flush_to(&mut vsock_conn.stream) {
match conn.tx_buf.flush_to(&mut conn.stream) {
Ok(cnt) => {
if cnt > 0 {
vsock_conn.fwd_cnt += Wrapping(cnt as u32);
vsock_conn.rx_queue.enqueue(RxOps::CreditUpdate);
conn.fwd_cnt += Wrapping(cnt as u32);
conn.rx_queue.enqueue(RxOps::CreditUpdate);
}
self.thread_backend.backend_rxq.push_back(ConnMapKey::new(
vsock_conn.local_port,
vsock_conn.peer_port,
));
self.thread_backend
.backend_rxq
.push_back(ConnMapKey::new(conn.local_port, conn.peer_port));
}
Err(e) => {
dbg!("Error: {:?}", e);
@ -293,10 +290,10 @@ impl VhostUserVsockThread {
Self::epoll_unregister(self.epoll_file.as_raw_fd(), fd).unwrap();
// Enqueue a read request
vsock_conn.rx_queue.enqueue(RxOps::Rw);
conn.rx_queue.enqueue(RxOps::Rw);
self.thread_backend
.backend_rxq
.push_back(ConnMapKey::new(vsock_conn.local_port, vsock_conn.peer_port));
.push_back(ConnMapKey::new(conn.local_port, conn.peer_port));
}
}
}

227
vsock/src/vsock_conn.rs
View File

@ -486,56 +486,56 @@ mod tests {
fn test_vsock_conn_init() {
// new locally inititated connection
let dummy_file = VsockDummySocket::new();
let mut vsock_conn_local =
let mut conn_local =
VsockConnection::new_local_init(dummy_file, VSOCK_HOST_CID, 5000, 3, 5001, -1);
assert!(!vsock_conn_local.connect);
assert_eq!(vsock_conn_local.peer_port, 5001);
assert_eq!(vsock_conn_local.rx_queue, RxQueue::new());
assert_eq!(vsock_conn_local.local_cid, VSOCK_HOST_CID);
assert_eq!(vsock_conn_local.local_port, 5000);
assert_eq!(vsock_conn_local.guest_cid, 3);
assert!(!conn_local.connect);
assert_eq!(conn_local.peer_port, 5001);
assert_eq!(conn_local.rx_queue, RxQueue::new());
assert_eq!(conn_local.local_cid, VSOCK_HOST_CID);
assert_eq!(conn_local.local_port, 5000);
assert_eq!(conn_local.guest_cid, 3);
// set peer port
vsock_conn_local.set_peer_port(5002);
assert_eq!(vsock_conn_local.peer_port, 5002);
conn_local.set_peer_port(5002);
assert_eq!(conn_local.peer_port, 5002);
// New connection initiated by the peer/guest
let dummy_file = VsockDummySocket::new();
let mut vsock_conn_peer =
let mut conn_peer =
VsockConnection::new_peer_init(dummy_file, VSOCK_HOST_CID, 5000, 3, 5001, -1, 65536);
assert!(!vsock_conn_peer.connect);
assert_eq!(vsock_conn_peer.peer_port, 5001);
assert_eq!(vsock_conn_peer.rx_queue.dequeue().unwrap(), RxOps::Response);
assert!(!vsock_conn_peer.rx_queue.pending_rx());
assert_eq!(vsock_conn_peer.local_cid, VSOCK_HOST_CID);
assert_eq!(vsock_conn_peer.local_port, 5000);
assert_eq!(vsock_conn_peer.guest_cid, 3);
assert_eq!(vsock_conn_peer.peer_buf_alloc, 65536);
assert!(!conn_peer.connect);
assert_eq!(conn_peer.peer_port, 5001);
assert_eq!(conn_peer.rx_queue.dequeue().unwrap(), RxOps::Response);
assert!(!conn_peer.rx_queue.pending_rx());
assert_eq!(conn_peer.local_cid, VSOCK_HOST_CID);
assert_eq!(conn_peer.local_port, 5000);
assert_eq!(conn_peer.guest_cid, 3);
assert_eq!(conn_peer.peer_buf_alloc, 65536);
}
#[test]
fn test_vsock_conn_credit() {
// new locally inititated connection
let dummy_file = VsockDummySocket::new();
let mut vsock_conn_local =
let mut conn_local =
VsockConnection::new_local_init(dummy_file, VSOCK_HOST_CID, 5000, 3, 5001, -1);
assert_eq!(vsock_conn_local.peer_avail_credit(), 0);
assert!(vsock_conn_local.need_credit_update_from_peer());
assert_eq!(conn_local.peer_avail_credit(), 0);
assert!(conn_local.need_credit_update_from_peer());
vsock_conn_local.peer_buf_alloc = 65536;
assert_eq!(vsock_conn_local.peer_avail_credit(), 65536);
assert!(!vsock_conn_local.need_credit_update_from_peer());
conn_local.peer_buf_alloc = 65536;
assert_eq!(conn_local.peer_avail_credit(), 65536);
assert!(!conn_local.need_credit_update_from_peer());
vsock_conn_local.rx_cnt = Wrapping(32768);
assert_eq!(vsock_conn_local.peer_avail_credit(), 32768);
assert!(!vsock_conn_local.need_credit_update_from_peer());
conn_local.rx_cnt = Wrapping(32768);
assert_eq!(conn_local.peer_avail_credit(), 32768);
assert!(!conn_local.need_credit_update_from_peer());
vsock_conn_local.rx_cnt = Wrapping(65536);
assert_eq!(vsock_conn_local.peer_avail_credit(), 0);
assert!(vsock_conn_local.need_credit_update_from_peer());
conn_local.rx_cnt = Wrapping(65536);
assert_eq!(conn_local.peer_avail_credit(), 0);
assert!(conn_local.need_credit_update_from_peer());
}
#[test]
@ -545,26 +545,26 @@ mod tests {
// new locally inititated connection
let dummy_file = VsockDummySocket::new();
let vsock_conn_local =
let conn_local =
VsockConnection::new_local_init(dummy_file, VSOCK_HOST_CID, 5000, 3, 5001, -1);
// write only descriptor chain
let (mem, mut descr_chain) = prepare_desc_chain_vsock(true, &head_params, 2, 10);
let mem = mem.memory();
let mut vsock_pkt =
let mut pkt =
VsockPacket::from_rx_virtq_chain(mem.deref(), &mut descr_chain, CONN_TX_BUF_SIZE)
.unwrap();
// initialize a vsock packet for the guest
vsock_conn_local.init_pkt(&mut vsock_pkt);
conn_local.init_pkt(&mut pkt);
assert_eq!(vsock_pkt.src_cid(), VSOCK_HOST_CID);
assert_eq!(vsock_pkt.dst_cid(), 3);
assert_eq!(vsock_pkt.src_port(), 5000);
assert_eq!(vsock_pkt.dst_port(), 5001);
assert_eq!(vsock_pkt.type_(), VSOCK_TYPE_STREAM);
assert_eq!(vsock_pkt.buf_alloc(), CONN_TX_BUF_SIZE);
assert_eq!(vsock_pkt.fwd_cnt(), 0);
assert_eq!(pkt.src_cid(), VSOCK_HOST_CID);
assert_eq!(pkt.dst_cid(), 3);
assert_eq!(pkt.src_port(), 5000);
assert_eq!(pkt.dst_port(), 5001);
assert_eq!(pkt.type_(), VSOCK_TYPE_STREAM);
assert_eq!(pkt.buf_alloc(), CONN_TX_BUF_SIZE);
assert_eq!(pkt.fwd_cnt(), 0);
}
#[test]
@ -574,86 +574,86 @@ mod tests {
// new locally inititated connection
let dummy_file = VsockDummySocket::new();
let mut vsock_conn_local =
let mut conn_local =
VsockConnection::new_local_init(dummy_file, VSOCK_HOST_CID, 5000, 3, 5001, -1);
// write only descriptor chain
let (mem, mut descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5);
let mem = mem.memory();
let mut vsock_pkt =
let mut pkt =
VsockPacket::from_rx_virtq_chain(mem.deref(), &mut descr_chain, CONN_TX_BUF_SIZE)
.unwrap();
// VSOCK_OP_REQUEST: new local conn request
vsock_conn_local.rx_queue.enqueue(RxOps::Request);
let vsock_op_req = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(vsock_op_req.is_ok());
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert_eq!(vsock_pkt.op(), VSOCK_OP_REQUEST);
conn_local.rx_queue.enqueue(RxOps::Request);
let op_req = conn_local.recv_pkt(&mut pkt);
assert!(op_req.is_ok());
assert!(!conn_local.rx_queue.pending_rx());
assert_eq!(pkt.op(), VSOCK_OP_REQUEST);
// VSOCK_OP_RST: reset if connection not established
vsock_conn_local.rx_queue.enqueue(RxOps::Rw);
let vsock_op_rst = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(vsock_op_rst.is_ok());
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert_eq!(vsock_pkt.op(), VSOCK_OP_RST);
conn_local.rx_queue.enqueue(RxOps::Rw);
let op_rst = conn_local.recv_pkt(&mut pkt);
assert!(op_rst.is_ok());
assert!(!conn_local.rx_queue.pending_rx());
assert_eq!(pkt.op(), VSOCK_OP_RST);
// VSOCK_OP_CREDIT_UPDATE: need credit update from peer/guest
vsock_conn_local.connect = true;
vsock_conn_local.rx_queue.enqueue(RxOps::Rw);
vsock_conn_local.fwd_cnt = Wrapping(1024);
let vsock_op_credit_update = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(vsock_op_credit_update.is_ok());
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert_eq!(vsock_pkt.op(), VSOCK_OP_CREDIT_REQUEST);
assert_eq!(vsock_conn_local.last_fwd_cnt, Wrapping(1024));
conn_local.connect = true;
conn_local.rx_queue.enqueue(RxOps::Rw);
conn_local.fwd_cnt = Wrapping(1024);
let op_credit_update = conn_local.recv_pkt(&mut pkt);
assert!(op_credit_update.is_ok());
assert!(!conn_local.rx_queue.pending_rx());
assert_eq!(pkt.op(), VSOCK_OP_CREDIT_REQUEST);
assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024));
// VSOCK_OP_SHUTDOWN: zero data read from stream/file
vsock_conn_local.peer_buf_alloc = 65536;
vsock_conn_local.rx_queue.enqueue(RxOps::Rw);
let vsock_op_zero_read_shutdown = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(vsock_op_zero_read_shutdown.is_ok());
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert_eq!(vsock_conn_local.rx_cnt, Wrapping(0));
assert_eq!(vsock_conn_local.last_fwd_cnt, Wrapping(1024));
assert_eq!(vsock_pkt.op(), VSOCK_OP_SHUTDOWN);
conn_local.peer_buf_alloc = 65536;
conn_local.rx_queue.enqueue(RxOps::Rw);
let op_zero_read_shutdown = conn_local.recv_pkt(&mut pkt);
assert!(op_zero_read_shutdown.is_ok());
assert!(!conn_local.rx_queue.pending_rx());
assert_eq!(conn_local.rx_cnt, Wrapping(0));
assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024));
assert_eq!(pkt.op(), VSOCK_OP_SHUTDOWN);
assert_eq!(
vsock_pkt.flags(),
pkt.flags(),
VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND
);
// VSOCK_OP_RW: finite data read from stream/file
vsock_conn_local.stream.write_all(b"hello").unwrap();
vsock_conn_local.rx_queue.enqueue(RxOps::Rw);
let vsock_op_zero_read = vsock_conn_local.recv_pkt(&mut vsock_pkt);
conn_local.stream.write_all(b"hello").unwrap();
conn_local.rx_queue.enqueue(RxOps::Rw);
let op_zero_read = conn_local.recv_pkt(&mut pkt);
// below error due to epoll add
assert!(vsock_op_zero_read.is_err());
assert_eq!(vsock_pkt.op(), VSOCK_OP_RW);
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert_eq!(vsock_pkt.len(), 5);
assert!(op_zero_read.is_err());
assert_eq!(pkt.op(), VSOCK_OP_RW);
assert!(!conn_local.rx_queue.pending_rx());
assert_eq!(pkt.len(), 5);
let buf = &mut [0u8; 5];
assert!(vsock_pkt.data_slice().unwrap().read_slice(buf, 0).is_ok());
assert!(pkt.data_slice().unwrap().read_slice(buf, 0).is_ok());
assert_eq!(buf, b"hello");
// VSOCK_OP_RESPONSE: response from a locally initiated connection
vsock_conn_local.rx_queue.enqueue(RxOps::Response);
let vsock_op_response = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(vsock_op_response.is_ok());
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert_eq!(vsock_pkt.op(), VSOCK_OP_RESPONSE);
assert!(vsock_conn_local.connect);
conn_local.rx_queue.enqueue(RxOps::Response);
let op_response = conn_local.recv_pkt(&mut pkt);
assert!(op_response.is_ok());
assert!(!conn_local.rx_queue.pending_rx());
assert_eq!(pkt.op(), VSOCK_OP_RESPONSE);
assert!(conn_local.connect);
// VSOCK_OP_CREDIT_UPDATE: guest needs credit update
vsock_conn_local.rx_queue.enqueue(RxOps::CreditUpdate);
let vsock_op_credit_update = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(!vsock_conn_local.rx_queue.pending_rx());
assert!(vsock_op_credit_update.is_ok());
assert_eq!(vsock_pkt.op(), VSOCK_OP_CREDIT_UPDATE);
assert_eq!(vsock_conn_local.last_fwd_cnt, Wrapping(1024));
conn_local.rx_queue.enqueue(RxOps::CreditUpdate);
let op_credit_update = conn_local.recv_pkt(&mut pkt);
assert!(!conn_local.rx_queue.pending_rx());
assert!(op_credit_update.is_ok());
assert_eq!(pkt.op(), VSOCK_OP_CREDIT_UPDATE);
assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024));
// non-existent request
let vsock_op_error = vsock_conn_local.recv_pkt(&mut vsock_pkt);
assert!(vsock_op_error.is_err());
let op_error = conn_local.recv_pkt(&mut pkt);
assert!(op_error.is_err());
}
#[test]
@ -663,58 +663,55 @@ mod tests {
// new locally inititated connection
let dummy_file = VsockDummySocket::new();
let mut vsock_conn_local =
let mut conn_local =
VsockConnection::new_local_init(dummy_file, VSOCK_HOST_CID, 5000, 3, 5001, -1);
// write only descriptor chain
let (mem, mut descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5);
let mem = mem.memory();
let mut vsock_pkt =
let mut pkt =
VsockPacket::from_tx_virtq_chain(mem.deref(), &mut descr_chain, CONN_TX_BUF_SIZE)
.unwrap();
// peer credit information
vsock_pkt.set_buf_alloc(65536).set_fwd_cnt(1024);
pkt.set_buf_alloc(65536).set_fwd_cnt(1024);
// check if peer credit information is updated currently
let credit_check = vsock_conn_local.send_pkt(&vsock_pkt);
let credit_check = conn_local.send_pkt(&pkt);
assert!(credit_check.is_ok());
assert_eq!(vsock_conn_local.peer_buf_alloc, 65536);
assert_eq!(vsock_conn_local.peer_fwd_cnt, Wrapping(1024));
assert_eq!(conn_local.peer_buf_alloc, 65536);
assert_eq!(conn_local.peer_fwd_cnt, Wrapping(1024));
// VSOCK_OP_RESPONSE
vsock_pkt.set_op(VSOCK_OP_RESPONSE);
let peer_response = vsock_conn_local.send_pkt(&vsock_pkt);
pkt.set_op(VSOCK_OP_RESPONSE);
let peer_response = conn_local.send_pkt(&pkt);
assert!(peer_response.is_ok());
assert!(vsock_conn_local.connect);
assert!(conn_local.connect);
let mut resp_buf = vec![0; 8];
vsock_conn_local.stream.read_exact(&mut resp_buf).unwrap();
conn_local.stream.read_exact(&mut resp_buf).unwrap();
assert_eq!(resp_buf, b"OK 5001\n");
// VSOCK_OP_RW
vsock_pkt.set_op(VSOCK_OP_RW);
pkt.set_op(VSOCK_OP_RW);
let buf = b"hello";
assert!(vsock_pkt.data_slice().unwrap().write_slice(buf, 0).is_ok());
let rw_response = vsock_conn_local.send_pkt(&vsock_pkt);
assert!(pkt.data_slice().unwrap().write_slice(buf, 0).is_ok());
let rw_response = conn_local.send_pkt(&pkt);
assert!(rw_response.is_ok());
let mut resp_buf = vec![0; 5];
vsock_conn_local.stream.read_exact(&mut resp_buf).unwrap();
conn_local.stream.read_exact(&mut resp_buf).unwrap();
assert_eq!(resp_buf, b"hello");
// VSOCK_OP_CREDIT_REQUEST
vsock_pkt.set_op(VSOCK_OP_CREDIT_REQUEST);
let credit_response = vsock_conn_local.send_pkt(&vsock_pkt);
pkt.set_op(VSOCK_OP_CREDIT_REQUEST);
let credit_response = conn_local.send_pkt(&pkt);
assert!(credit_response.is_ok());
assert_eq!(
vsock_conn_local.rx_queue.peek().unwrap(),
RxOps::CreditUpdate
);
assert_eq!(conn_local.rx_queue.peek().unwrap(), RxOps::CreditUpdate);
// VSOCK_OP_SHUTDOWN
vsock_pkt.set_op(VSOCK_OP_SHUTDOWN);
vsock_pkt.set_flags(VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND);
let shutdown_response = vsock_conn_local.send_pkt(&vsock_pkt);
pkt.set_op(VSOCK_OP_SHUTDOWN);
pkt.set_flags(VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND);
let shutdown_response = conn_local.send_pkt(&pkt);
assert!(shutdown_response.is_ok());
assert!(vsock_conn_local.rx_queue.contains(RxOps::Reset.bitmask()));
assert!(conn_local.rx_queue.contains(RxOps::Reset.bitmask()));
}
}