multiconnectionmanager.rs 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763
  1. use super::{
  2. protocol::VsockAddr, vsock::ConnectionInfo, DisconnectReason, SocketError, VirtIOSocket,
  3. VsockEvent, VsockEventType,
  4. };
  5. use crate::{transport::Transport, Hal, Result};
  6. use alloc::{boxed::Box, vec::Vec};
  7. use core::cmp::min;
  8. use core::convert::TryInto;
  9. use core::hint::spin_loop;
  10. use log::debug;
  11. use zerocopy::FromBytes;
  12. const PER_CONNECTION_BUFFER_CAPACITY: usize = 1024;
  13. /// A higher level interface for VirtIO socket (vsock) devices.
  14. ///
  15. /// This keeps track of multiple vsock connections.
  16. ///
  17. /// # Example
  18. ///
  19. /// ```
  20. /// # use virtio_drivers::{Error, Hal};
  21. /// # use virtio_drivers::transport::Transport;
  22. /// use virtio_drivers::device::socket::{VirtIOSocket, VsockAddr, VsockConnectionManager};
  23. ///
  24. /// # fn example<HalImpl: Hal, T: Transport>(transport: T) -> Result<(), Error> {
  25. /// let mut socket = VsockConnectionManager::new(VirtIOSocket::<HalImpl, _>::new(transport)?);
  26. ///
  27. /// // Start a thread to call `socket.poll()` and handle events.
  28. ///
  29. /// let remote_address = VsockAddr { cid: 2, port: 42 };
  30. /// let local_port = 1234;
  31. /// socket.connect(remote_address, local_port)?;
  32. ///
  33. /// // Wait until `socket.poll()` returns an event indicating that the socket is connected.
  34. ///
  35. /// socket.send(remote_address, local_port, "Hello world".as_bytes())?;
  36. ///
  37. /// socket.shutdown(remote_address, local_port)?;
  38. /// # Ok(())
  39. /// # }
  40. /// ```
  41. pub struct VsockConnectionManager<H: Hal, T: Transport> {
  42. driver: VirtIOSocket<H, T>,
  43. connections: Vec<Connection>,
  44. listening_ports: Vec<u32>,
  45. }
  46. #[derive(Debug)]
  47. struct Connection {
  48. info: ConnectionInfo,
  49. buffer: RingBuffer,
  50. /// The peer sent a SHUTDOWN request, but we haven't yet responded with a RST because there is
  51. /// still data in the buffer.
  52. peer_requested_shutdown: bool,
  53. }
  54. impl Connection {
  55. fn new(peer: VsockAddr, local_port: u32) -> Self {
  56. let mut info = ConnectionInfo::new(peer, local_port);
  57. info.buf_alloc = PER_CONNECTION_BUFFER_CAPACITY.try_into().unwrap();
  58. Self {
  59. info,
  60. buffer: RingBuffer::new(PER_CONNECTION_BUFFER_CAPACITY),
  61. peer_requested_shutdown: false,
  62. }
  63. }
  64. }
  65. impl<H: Hal, T: Transport> VsockConnectionManager<H, T> {
  66. /// Construct a new connection manager wrapping the given low-level VirtIO socket driver.
  67. pub fn new(driver: VirtIOSocket<H, T>) -> Self {
  68. Self {
  69. driver,
  70. connections: Vec::new(),
  71. listening_ports: Vec::new(),
  72. }
  73. }
  74. /// Returns the CID which has been assigned to this guest.
  75. pub fn guest_cid(&self) -> u64 {
  76. self.driver.guest_cid()
  77. }
  78. /// Allows incoming connections on the given port number.
  79. pub fn listen(&mut self, port: u32) {
  80. if !self.listening_ports.contains(&port) {
  81. self.listening_ports.push(port);
  82. }
  83. }
  84. /// Stops allowing incoming connections on the given port number.
  85. pub fn unlisten(&mut self, port: u32) {
  86. self.listening_ports.retain(|p| *p != port);
  87. }
  88. /// Sends a request to connect to the given destination.
  89. ///
  90. /// This returns as soon as the request is sent; you should wait until `poll` returns a
  91. /// `VsockEventType::Connected` event indicating that the peer has accepted the connection
  92. /// before sending data.
  93. pub fn connect(&mut self, destination: VsockAddr, src_port: u32) -> Result {
  94. if self.connections.iter().any(|connection| {
  95. connection.info.dst == destination && connection.info.src_port == src_port
  96. }) {
  97. return Err(SocketError::ConnectionExists.into());
  98. }
  99. let new_connection = Connection::new(destination, src_port);
  100. self.driver.connect(&new_connection.info)?;
  101. debug!("Connection requested: {:?}", new_connection.info);
  102. self.connections.push(new_connection);
  103. Ok(())
  104. }
  105. /// Sends the buffer to the destination.
  106. pub fn send(&mut self, destination: VsockAddr, src_port: u32, buffer: &[u8]) -> Result {
  107. let connection = self
  108. .connections
  109. .iter_mut()
  110. .find(|connection| {
  111. connection.info.dst == destination && connection.info.src_port == src_port
  112. })
  113. .ok_or(SocketError::NotConnected)?;
  114. self.driver.send(buffer, &mut connection.info)
  115. }
  116. /// Polls the vsock device to receive data or other updates.
  117. pub fn poll(&mut self) -> Result<Option<VsockEvent>> {
  118. let guest_cid = self.driver.guest_cid();
  119. let connections = &mut self.connections;
  120. let result = self.driver.poll(|event, body| {
  121. let connection = connections
  122. .iter_mut()
  123. .find(|connection| event.matches_connection(&connection.info, guest_cid));
  124. // Skip events which don't match any connection we know about, unless they are a
  125. // connection request.
  126. let connection = if let Some(connection) = connection {
  127. connection
  128. } else if let VsockEventType::ConnectionRequest = event.event_type {
  129. // If the requested connection already exists or the CID isn't ours, ignore it.
  130. if connection.is_some() || event.destination.cid != guest_cid {
  131. return Ok(None);
  132. }
  133. // Add the new connection to our list, at least for now. It will be removed again
  134. // below if we weren't listening on the port.
  135. connections.push(Connection::new(event.source, event.destination.port));
  136. connections.last_mut().unwrap()
  137. } else {
  138. return Ok(None);
  139. };
  140. // Update stored connection info.
  141. connection.info.update_for_event(&event);
  142. if let VsockEventType::Received { length } = event.event_type {
  143. // Copy to buffer
  144. if !connection.buffer.write(body) {
  145. return Err(SocketError::OutputBufferTooShort(length).into());
  146. }
  147. }
  148. Ok(Some(event))
  149. })?;
  150. let Some(event) = result else {
  151. return Ok(None);
  152. };
  153. // The connection must exist because we found it above in the callback.
  154. let (connection_index, connection) = connections
  155. .iter_mut()
  156. .enumerate()
  157. .find(|(_, connection)| event.matches_connection(&connection.info, guest_cid))
  158. .unwrap();
  159. match event.event_type {
  160. VsockEventType::ConnectionRequest => {
  161. if self.listening_ports.contains(&event.destination.port) {
  162. self.driver.accept(&connection.info)?;
  163. } else {
  164. // Reject the connection request and remove it from our list.
  165. self.driver.force_close(&connection.info)?;
  166. self.connections.swap_remove(connection_index);
  167. // No need to pass the request on to the client, as we've already rejected it.
  168. return Ok(None);
  169. }
  170. }
  171. VsockEventType::Connected => {}
  172. VsockEventType::Disconnected { reason } => {
  173. // Wait until client reads all data before removing connection.
  174. if connection.buffer.is_empty() {
  175. if reason == DisconnectReason::Shutdown {
  176. self.driver.force_close(&connection.info)?;
  177. }
  178. self.connections.swap_remove(connection_index);
  179. } else {
  180. connection.peer_requested_shutdown = true;
  181. }
  182. }
  183. VsockEventType::Received { .. } => {
  184. // Already copied the buffer in the callback above.
  185. }
  186. VsockEventType::CreditRequest => {
  187. // If the peer requested credit, send an update.
  188. self.driver.credit_update(&connection.info)?;
  189. // No need to pass the request on to the client, we've already handled it.
  190. return Ok(None);
  191. }
  192. VsockEventType::CreditUpdate => {}
  193. }
  194. Ok(Some(event))
  195. }
  196. /// Reads data received from the given connection.
  197. pub fn recv(&mut self, peer: VsockAddr, src_port: u32, buffer: &mut [u8]) -> Result<usize> {
  198. let (connection_index, connection) = self
  199. .connections
  200. .iter_mut()
  201. .enumerate()
  202. .find(|(_, connection)| {
  203. connection.info.dst == peer && connection.info.src_port == src_port
  204. })
  205. .ok_or(SocketError::NotConnected)?;
  206. // Copy from ring buffer
  207. let bytes_read = connection.buffer.read(buffer);
  208. connection.info.done_forwarding(bytes_read);
  209. // If buffer is now empty and the peer requested shutdown, finish shutting down the
  210. // connection.
  211. if connection.peer_requested_shutdown && connection.buffer.is_empty() {
  212. self.driver.force_close(&connection.info)?;
  213. self.connections.swap_remove(connection_index);
  214. }
  215. Ok(bytes_read)
  216. }
  217. /// Blocks until we get some event from the vsock device.
  218. pub fn wait_for_event(&mut self) -> Result<VsockEvent> {
  219. loop {
  220. if let Some(event) = self.poll()? {
  221. return Ok(event);
  222. } else {
  223. spin_loop();
  224. }
  225. }
  226. }
  227. /// Requests to shut down the connection cleanly.
  228. ///
  229. /// This returns as soon as the request is sent; you should wait until `poll` returns a
  230. /// `VsockEventType::Disconnected` event if you want to know that the peer has acknowledged the
  231. /// shutdown.
  232. pub fn shutdown(&mut self, destination: VsockAddr, src_port: u32) -> Result {
  233. let connection = self
  234. .connections
  235. .iter()
  236. .find(|connection| {
  237. connection.info.dst == destination && connection.info.src_port == src_port
  238. })
  239. .ok_or(SocketError::NotConnected)?;
  240. self.driver.shutdown(&connection.info)
  241. }
  242. /// Forcibly closes the connection without waiting for the peer.
  243. pub fn force_close(&mut self, destination: VsockAddr, src_port: u32) -> Result {
  244. let (index, connection) = self
  245. .connections
  246. .iter()
  247. .enumerate()
  248. .find(|(_, connection)| {
  249. connection.info.dst == destination && connection.info.src_port == src_port
  250. })
  251. .ok_or(SocketError::NotConnected)?;
  252. self.driver.force_close(&connection.info)?;
  253. self.connections.swap_remove(index);
  254. Ok(())
  255. }
  256. }
  257. #[derive(Debug)]
  258. struct RingBuffer {
  259. buffer: Box<[u8]>,
  260. /// The number of bytes currently in the buffer.
  261. used: usize,
  262. /// The index of the first used byte in the buffer.
  263. start: usize,
  264. }
  265. impl RingBuffer {
  266. pub fn new(capacity: usize) -> Self {
  267. Self {
  268. buffer: FromBytes::new_box_slice_zeroed(capacity),
  269. used: 0,
  270. start: 0,
  271. }
  272. }
  273. /// Returns the number of bytes currently used in the buffer.
  274. pub fn used(&self) -> usize {
  275. self.used
  276. }
  277. /// Returns true iff there are currently no bytes in the buffer.
  278. pub fn is_empty(&self) -> bool {
  279. self.used == 0
  280. }
  281. /// Returns the number of bytes currently free in the buffer.
  282. pub fn available(&self) -> usize {
  283. self.buffer.len() - self.used
  284. }
  285. /// Adds the given bytes to the buffer if there is enough capacity for them all.
  286. ///
  287. /// Returns true if they were added, or false if they were not.
  288. pub fn write(&mut self, bytes: &[u8]) -> bool {
  289. if bytes.len() > self.available() {
  290. return false;
  291. }
  292. let end = (self.start + self.used) % self.buffer.len();
  293. let write_before_wraparound = min(bytes.len(), self.buffer.len() - end);
  294. let write_after_wraparound = bytes
  295. .len()
  296. .checked_sub(write_before_wraparound)
  297. .unwrap_or_default();
  298. self.buffer[end..end + write_before_wraparound]
  299. .copy_from_slice(&bytes[0..write_before_wraparound]);
  300. self.buffer[0..write_after_wraparound].copy_from_slice(&bytes[write_before_wraparound..]);
  301. self.used += bytes.len();
  302. true
  303. }
  304. /// Reads and removes as many bytes as possible from the buffer, up to the length of the given
  305. /// buffer.
  306. pub fn read(&mut self, out: &mut [u8]) -> usize {
  307. let bytes_read = min(self.used, out.len());
  308. // The number of bytes to copy out between `start` and the end of the buffer.
  309. let read_before_wraparound = min(bytes_read, self.buffer.len() - self.start);
  310. // The number of bytes to copy out from the beginning of the buffer after wrapping around.
  311. let read_after_wraparound = bytes_read
  312. .checked_sub(read_before_wraparound)
  313. .unwrap_or_default();
  314. out[0..read_before_wraparound]
  315. .copy_from_slice(&self.buffer[self.start..self.start + read_before_wraparound]);
  316. out[read_before_wraparound..bytes_read]
  317. .copy_from_slice(&self.buffer[0..read_after_wraparound]);
  318. self.used -= bytes_read;
  319. self.start = (self.start + bytes_read) % self.buffer.len();
  320. bytes_read
  321. }
  322. }
  323. #[cfg(test)]
  324. mod tests {
  325. use super::*;
  326. use crate::{
  327. device::socket::{
  328. protocol::{SocketType, VirtioVsockConfig, VirtioVsockHdr, VirtioVsockOp},
  329. vsock::{VsockBufferStatus, QUEUE_SIZE, RX_QUEUE_IDX, TX_QUEUE_IDX},
  330. },
  331. hal::fake::FakeHal,
  332. transport::{
  333. fake::{FakeTransport, QueueStatus, State},
  334. DeviceStatus, DeviceType,
  335. },
  336. volatile::ReadOnly,
  337. };
  338. use alloc::{sync::Arc, vec};
  339. use core::{mem::size_of, ptr::NonNull};
  340. use std::{sync::Mutex, thread};
  341. use zerocopy::{AsBytes, FromBytes};
  342. #[test]
  343. fn send_recv() {
  344. let host_cid = 2;
  345. let guest_cid = 66;
  346. let host_port = 1234;
  347. let guest_port = 4321;
  348. let host_address = VsockAddr {
  349. cid: host_cid,
  350. port: host_port,
  351. };
  352. let hello_from_guest = "Hello from guest";
  353. let hello_from_host = "Hello from host";
  354. let mut config_space = VirtioVsockConfig {
  355. guest_cid_low: ReadOnly::new(66),
  356. guest_cid_high: ReadOnly::new(0),
  357. };
  358. let state = Arc::new(Mutex::new(State {
  359. status: DeviceStatus::empty(),
  360. driver_features: 0,
  361. guest_page_size: 0,
  362. interrupt_pending: false,
  363. queues: vec![
  364. QueueStatus::default(),
  365. QueueStatus::default(),
  366. QueueStatus::default(),
  367. ],
  368. }));
  369. let transport = FakeTransport {
  370. device_type: DeviceType::Socket,
  371. max_queue_size: 32,
  372. device_features: 0,
  373. config_space: NonNull::from(&mut config_space),
  374. state: state.clone(),
  375. };
  376. let mut socket = VsockConnectionManager::new(
  377. VirtIOSocket::<FakeHal, FakeTransport<VirtioVsockConfig>>::new(transport).unwrap(),
  378. );
  379. // Start a thread to simulate the device.
  380. let handle = thread::spawn(move || {
  381. // Wait for connection request.
  382. State::wait_until_queue_notified(&state, TX_QUEUE_IDX);
  383. assert_eq!(
  384. VirtioVsockHdr::read_from(
  385. state
  386. .lock()
  387. .unwrap()
  388. .read_from_queue::<QUEUE_SIZE>(TX_QUEUE_IDX)
  389. .as_slice()
  390. )
  391. .unwrap(),
  392. VirtioVsockHdr {
  393. op: VirtioVsockOp::Request.into(),
  394. src_cid: guest_cid.into(),
  395. dst_cid: host_cid.into(),
  396. src_port: guest_port.into(),
  397. dst_port: host_port.into(),
  398. len: 0.into(),
  399. socket_type: SocketType::Stream.into(),
  400. flags: 0.into(),
  401. buf_alloc: 1024.into(),
  402. fwd_cnt: 0.into(),
  403. }
  404. );
  405. // Accept connection and give the peer enough credit to send the message.
  406. state.lock().unwrap().write_to_queue::<QUEUE_SIZE>(
  407. RX_QUEUE_IDX,
  408. VirtioVsockHdr {
  409. op: VirtioVsockOp::Response.into(),
  410. src_cid: host_cid.into(),
  411. dst_cid: guest_cid.into(),
  412. src_port: host_port.into(),
  413. dst_port: guest_port.into(),
  414. len: 0.into(),
  415. socket_type: SocketType::Stream.into(),
  416. flags: 0.into(),
  417. buf_alloc: 50.into(),
  418. fwd_cnt: 0.into(),
  419. }
  420. .as_bytes(),
  421. );
  422. // Expect the guest to send some data.
  423. State::wait_until_queue_notified(&state, TX_QUEUE_IDX);
  424. let request = state
  425. .lock()
  426. .unwrap()
  427. .read_from_queue::<QUEUE_SIZE>(TX_QUEUE_IDX);
  428. assert_eq!(
  429. request.len(),
  430. size_of::<VirtioVsockHdr>() + hello_from_guest.len()
  431. );
  432. assert_eq!(
  433. VirtioVsockHdr::read_from_prefix(request.as_slice()).unwrap(),
  434. VirtioVsockHdr {
  435. op: VirtioVsockOp::Rw.into(),
  436. src_cid: guest_cid.into(),
  437. dst_cid: host_cid.into(),
  438. src_port: guest_port.into(),
  439. dst_port: host_port.into(),
  440. len: (hello_from_guest.len() as u32).into(),
  441. socket_type: SocketType::Stream.into(),
  442. flags: 0.into(),
  443. buf_alloc: 1024.into(),
  444. fwd_cnt: 0.into(),
  445. }
  446. );
  447. assert_eq!(
  448. &request[size_of::<VirtioVsockHdr>()..],
  449. hello_from_guest.as_bytes()
  450. );
  451. println!("Host sending");
  452. // Send a response.
  453. let mut response = vec![0; size_of::<VirtioVsockHdr>() + hello_from_host.len()];
  454. VirtioVsockHdr {
  455. op: VirtioVsockOp::Rw.into(),
  456. src_cid: host_cid.into(),
  457. dst_cid: guest_cid.into(),
  458. src_port: host_port.into(),
  459. dst_port: guest_port.into(),
  460. len: (hello_from_host.len() as u32).into(),
  461. socket_type: SocketType::Stream.into(),
  462. flags: 0.into(),
  463. buf_alloc: 50.into(),
  464. fwd_cnt: (hello_from_guest.len() as u32).into(),
  465. }
  466. .write_to_prefix(response.as_mut_slice());
  467. response[size_of::<VirtioVsockHdr>()..].copy_from_slice(hello_from_host.as_bytes());
  468. state
  469. .lock()
  470. .unwrap()
  471. .write_to_queue::<QUEUE_SIZE>(RX_QUEUE_IDX, &response);
  472. // Expect a shutdown.
  473. State::wait_until_queue_notified(&state, TX_QUEUE_IDX);
  474. assert_eq!(
  475. VirtioVsockHdr::read_from(
  476. state
  477. .lock()
  478. .unwrap()
  479. .read_from_queue::<QUEUE_SIZE>(TX_QUEUE_IDX)
  480. .as_slice()
  481. )
  482. .unwrap(),
  483. VirtioVsockHdr {
  484. op: VirtioVsockOp::Shutdown.into(),
  485. src_cid: guest_cid.into(),
  486. dst_cid: host_cid.into(),
  487. src_port: guest_port.into(),
  488. dst_port: host_port.into(),
  489. len: 0.into(),
  490. socket_type: SocketType::Stream.into(),
  491. flags: 0.into(),
  492. buf_alloc: 1024.into(),
  493. fwd_cnt: (hello_from_host.len() as u32).into(),
  494. }
  495. );
  496. });
  497. socket.connect(host_address, guest_port).unwrap();
  498. assert_eq!(
  499. socket.wait_for_event().unwrap(),
  500. VsockEvent {
  501. source: host_address,
  502. destination: VsockAddr {
  503. cid: guest_cid,
  504. port: guest_port,
  505. },
  506. event_type: VsockEventType::Connected,
  507. buffer_status: VsockBufferStatus {
  508. buffer_allocation: 50,
  509. forward_count: 0,
  510. },
  511. }
  512. );
  513. println!("Guest sending");
  514. socket
  515. .send(host_address, guest_port, "Hello from guest".as_bytes())
  516. .unwrap();
  517. println!("Guest waiting to receive.");
  518. assert_eq!(
  519. socket.wait_for_event().unwrap(),
  520. VsockEvent {
  521. source: host_address,
  522. destination: VsockAddr {
  523. cid: guest_cid,
  524. port: guest_port,
  525. },
  526. event_type: VsockEventType::Received {
  527. length: hello_from_host.len()
  528. },
  529. buffer_status: VsockBufferStatus {
  530. buffer_allocation: 50,
  531. forward_count: hello_from_guest.len() as u32,
  532. },
  533. }
  534. );
  535. println!("Guest getting received data.");
  536. let mut buffer = [0u8; 64];
  537. assert_eq!(
  538. socket.recv(host_address, guest_port, &mut buffer).unwrap(),
  539. hello_from_host.len()
  540. );
  541. assert_eq!(
  542. &buffer[0..hello_from_host.len()],
  543. hello_from_host.as_bytes()
  544. );
  545. socket.shutdown(host_address, guest_port).unwrap();
  546. handle.join().unwrap();
  547. }
  548. #[test]
  549. fn incoming_connection() {
  550. let host_cid = 2;
  551. let guest_cid = 66;
  552. let host_port = 1234;
  553. let guest_port = 4321;
  554. let wrong_guest_port = 4444;
  555. let host_address = VsockAddr {
  556. cid: host_cid,
  557. port: host_port,
  558. };
  559. let mut config_space = VirtioVsockConfig {
  560. guest_cid_low: ReadOnly::new(66),
  561. guest_cid_high: ReadOnly::new(0),
  562. };
  563. let state = Arc::new(Mutex::new(State {
  564. status: DeviceStatus::empty(),
  565. driver_features: 0,
  566. guest_page_size: 0,
  567. interrupt_pending: false,
  568. queues: vec![
  569. QueueStatus::default(),
  570. QueueStatus::default(),
  571. QueueStatus::default(),
  572. ],
  573. }));
  574. let transport = FakeTransport {
  575. device_type: DeviceType::Socket,
  576. max_queue_size: 32,
  577. device_features: 0,
  578. config_space: NonNull::from(&mut config_space),
  579. state: state.clone(),
  580. };
  581. let mut socket = VsockConnectionManager::new(
  582. VirtIOSocket::<FakeHal, FakeTransport<VirtioVsockConfig>>::new(transport).unwrap(),
  583. );
  584. socket.listen(guest_port);
  585. // Start a thread to simulate the device.
  586. let handle = thread::spawn(move || {
  587. // Send a connection request for a port the guest isn't listening on.
  588. println!("Host sending connection request to wrong port");
  589. state.lock().unwrap().write_to_queue::<QUEUE_SIZE>(
  590. RX_QUEUE_IDX,
  591. VirtioVsockHdr {
  592. op: VirtioVsockOp::Request.into(),
  593. src_cid: host_cid.into(),
  594. dst_cid: guest_cid.into(),
  595. src_port: host_port.into(),
  596. dst_port: wrong_guest_port.into(),
  597. len: 0.into(),
  598. socket_type: SocketType::Stream.into(),
  599. flags: 0.into(),
  600. buf_alloc: 50.into(),
  601. fwd_cnt: 0.into(),
  602. }
  603. .as_bytes(),
  604. );
  605. // Expect a rejection.
  606. println!("Host waiting for rejection");
  607. State::wait_until_queue_notified(&state, TX_QUEUE_IDX);
  608. assert_eq!(
  609. VirtioVsockHdr::read_from(
  610. state
  611. .lock()
  612. .unwrap()
  613. .read_from_queue::<QUEUE_SIZE>(TX_QUEUE_IDX)
  614. .as_slice()
  615. )
  616. .unwrap(),
  617. VirtioVsockHdr {
  618. op: VirtioVsockOp::Rst.into(),
  619. src_cid: guest_cid.into(),
  620. dst_cid: host_cid.into(),
  621. src_port: wrong_guest_port.into(),
  622. dst_port: host_port.into(),
  623. len: 0.into(),
  624. socket_type: SocketType::Stream.into(),
  625. flags: 0.into(),
  626. buf_alloc: 1024.into(),
  627. fwd_cnt: 0.into(),
  628. }
  629. );
  630. // Send a connection request for a port the guest is listening on.
  631. println!("Host sending connection request to right port");
  632. state.lock().unwrap().write_to_queue::<QUEUE_SIZE>(
  633. RX_QUEUE_IDX,
  634. VirtioVsockHdr {
  635. op: VirtioVsockOp::Request.into(),
  636. src_cid: host_cid.into(),
  637. dst_cid: guest_cid.into(),
  638. src_port: host_port.into(),
  639. dst_port: guest_port.into(),
  640. len: 0.into(),
  641. socket_type: SocketType::Stream.into(),
  642. flags: 0.into(),
  643. buf_alloc: 50.into(),
  644. fwd_cnt: 0.into(),
  645. }
  646. .as_bytes(),
  647. );
  648. // Expect a response.
  649. println!("Host waiting for response");
  650. State::wait_until_queue_notified(&state, TX_QUEUE_IDX);
  651. assert_eq!(
  652. VirtioVsockHdr::read_from(
  653. state
  654. .lock()
  655. .unwrap()
  656. .read_from_queue::<QUEUE_SIZE>(TX_QUEUE_IDX)
  657. .as_slice()
  658. )
  659. .unwrap(),
  660. VirtioVsockHdr {
  661. op: VirtioVsockOp::Response.into(),
  662. src_cid: guest_cid.into(),
  663. dst_cid: host_cid.into(),
  664. src_port: guest_port.into(),
  665. dst_port: host_port.into(),
  666. len: 0.into(),
  667. socket_type: SocketType::Stream.into(),
  668. flags: 0.into(),
  669. buf_alloc: 1024.into(),
  670. fwd_cnt: 0.into(),
  671. }
  672. );
  673. println!("Host finished");
  674. });
  675. // Expect an incoming connection.
  676. println!("Guest expecting incoming connection.");
  677. assert_eq!(
  678. socket.wait_for_event().unwrap(),
  679. VsockEvent {
  680. source: host_address,
  681. destination: VsockAddr {
  682. cid: guest_cid,
  683. port: guest_port,
  684. },
  685. event_type: VsockEventType::ConnectionRequest,
  686. buffer_status: VsockBufferStatus {
  687. buffer_allocation: 50,
  688. forward_count: 0,
  689. },
  690. }
  691. );
  692. handle.join().unwrap();
  693. }
  694. }