net_backend/
lib.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! Network backend traits and infrastructure.
5//!
6//! This crate defines the abstraction boundary between network
7//! **frontends** (guest-facing devices) and network **backends**
8//! (host-side packet I/O). The key types are:
9//!
10//! * [`Endpoint`] — a backend factory. One per NIC, responsible for
11//!   creating [`Queue`] objects when the frontend activates the device.
12//!
13//! * [`Queue`] — a single TX/RX data path. Backends implement this to
14//!   send and receive packets. A device may have multiple queues (RSS).
15//!
16//! * [`BufferAccess`] — owned by the frontend, provides access to
17//!   guest memory receive buffers. Passed by `&mut` reference to every
18//!   [`Queue`] method that needs it, so the frontend retains exclusive
19//!   ownership and no internal locking is required.
20//!
21//! ## Lifecycle
22//!
23//! 1. The frontend creates a [`BufferAccess`] implementation and one
24//!    [`QueueConfig`] per desired queue (containing just a driver).
25//! 2. It calls [`Endpoint::get_queues`], which returns boxed [`Queue`]
26//!    objects.
27//! 3. The frontend posts initial receive buffers by calling
28//!    [`Queue::rx_avail`] with its [`BufferAccess`].
29//! 4. The main loop polls [`Queue::poll_ready`] for backend events,
30//!    then calls [`Queue::rx_poll`] / [`Queue::tx_avail`] /
31//!    [`Queue::tx_poll`] to exchange packets—always passing
32//!    `&mut dyn BufferAccess`.
33//! 5. On shutdown, queues are dropped and [`Endpoint::stop`] is called.
34
35#![expect(missing_docs)]
36#![forbid(unsafe_code)]
37
38pub mod loopback;
39pub mod null;
40pub mod resolve;
41pub mod tests;
42
43use async_trait::async_trait;
44use bitfield_struct::bitfield;
45use futures::FutureExt;
46use futures::StreamExt;
47use futures::TryFutureExt;
48use futures::lock::Mutex;
49use futures_concurrency::future::Race;
50use guestmem::GuestMemory;
51use guestmem::GuestMemoryError;
52use inspect::InspectMut;
53use inspect_counters::Counter;
54use mesh::rpc::Rpc;
55use mesh::rpc::RpcSend;
56use null::NullEndpoint;
57use pal_async::driver::Driver;
58use std::future::pending;
59use std::sync::Arc;
60use std::task::Context;
61use std::task::Poll;
62use thiserror::Error;
63
64/// Per-queue configuration passed to [`Endpoint::get_queues`].
65///
66/// Contains only an async driver handle. Receive buffers are posted
67/// separately via [`Queue::rx_avail`] after queue creation.
68pub struct QueueConfig {
69    pub driver: Box<dyn Driver>,
70}
71
72/// A network endpoint — the backend side of a NIC.
73///
74/// An endpoint is a factory for [`Queue`] objects. It represents a
75/// connection to some packet transport (TAP device, hardware NIC,
76/// user-space network stack, etc.) and can create one or more queues
77/// for parallel TX/RX processing.
78///
79/// Frontends (e.g. `virtio_net`, `netvsp`, `gdma`) own the endpoint
80/// and call [`get_queues`](Endpoint::get_queues) when the guest
81/// activates the NIC.
82#[async_trait]
83pub trait Endpoint: Send + Sync + InspectMut {
84    /// Returns an informational endpoint type.
85    fn endpoint_type(&self) -> &'static str;
86
87    /// Initializes the queues associated with the endpoint.
88    async fn get_queues(
89        &mut self,
90        config: Vec<QueueConfig>,
91        rss: Option<&RssConfig<'_>>,
92        queues: &mut Vec<Box<dyn Queue>>,
93    ) -> anyhow::Result<()>;
94
95    /// Stops the endpoint.
96    ///
97    /// All queues returned via `get_queues` must have been dropped.
98    async fn stop(&mut self);
99
100    /// Whether the endpoint completes buffers in the order they were made
101    /// available (RX buffers returned from `rx_poll`, and TX packets completed,
102    /// in available-ring order).
103    fn is_ordered(&self) -> bool;
104
105    /// Specifies the supported set of transmit offloads.
106    fn tx_offload_support(&self) -> TxOffloadSupport {
107        TxOffloadSupport::default()
108    }
109
110    /// Specifies parameters related to supporting multiple queues.
111    fn multiqueue_support(&self) -> MultiQueueSupport {
112        MultiQueueSupport {
113            max_queues: 1,
114            indirection_table_size: 0,
115        }
116    }
117
118    /// If true, transmits are guaranteed to complete quickly. This is used to
119    /// allow eliding tx notifications from the guest when there are already
120    /// some tx packets in flight.
121    fn tx_fast_completions(&self) -> bool {
122        false
123    }
124
125    /// Sets the current data path for packet flow (e.g. via vmbus synthnic or through virtual function).
126    /// This is only supported for endpoints that pair with an accelerated device.
127    async fn set_data_path_to_guest_vf(&self, _use_vf: bool) -> anyhow::Result<()> {
128        Err(anyhow::Error::msg("Unsupported in current endpoint"))
129    }
130
131    async fn get_data_path_to_guest_vf(&self) -> anyhow::Result<bool> {
132        Err(anyhow::Error::msg("Unsupported in current endpoint"))
133    }
134
135    /// On completion, the return value indicates the specific endpoint action to take.
136    async fn wait_for_endpoint_action(&mut self) -> EndpointAction {
137        pending().await
138    }
139
140    /// Link speed in bps.
141    fn link_speed(&self) -> u64 {
142        // Reporting a reasonable default value (10Gbps) here that the individual endpoints
143        // can overwrite.
144        10 * 1000 * 1000 * 1000
145    }
146}
147
148/// Multi-queue related support.
149#[derive(Debug, Copy, Clone)]
150pub struct MultiQueueSupport {
151    /// The number of supported queues.
152    pub max_queues: u16,
153    /// The size of the RSS indirection table.
154    pub indirection_table_size: u16,
155}
156
157/// The set of supported transmit offloads.
158#[derive(Debug, Copy, Clone, Default)]
159pub struct TxOffloadSupport {
160    /// IPv4 header checksum offload.
161    pub ipv4_header: bool,
162    /// TCP checksum offload.
163    pub tcp: bool,
164    /// UDP checksum offload.
165    pub udp: bool,
166    /// TCP segmentation offload.
167    pub tso: bool,
168    /// UDP segmentation offload (USO).
169    pub uso: bool,
170}
171
172#[derive(Debug, Clone)]
173pub struct RssConfig<'a> {
174    pub key: &'a [u8],
175    pub indirection_table: &'a [u16],
176    pub flags: u32, // TODO
177}
178
179#[derive(Error, Debug)]
180pub enum TxError {
181    #[error("error requiring queue restart. {0}")]
182    TryRestart(#[source] anyhow::Error),
183    #[error("unrecoverable error. {0}")]
184    Fatal(#[source] anyhow::Error),
185}
186pub trait BackendQueueStats {
187    fn rx_errors(&self) -> Counter;
188    fn tx_errors(&self) -> Counter;
189    fn rx_packets(&self) -> Counter;
190    fn tx_packets(&self) -> Counter;
191    fn tx_vlan_packets(&self) -> Counter {
192        Counter::new()
193    }
194    fn rx_vlan_packets(&self) -> Counter {
195        Counter::new()
196    }
197}
198
199/// A single TX/RX data path for sending and receiving network packets.
200///
201/// Created by [`Endpoint::get_queues`] and driven by the frontend in
202/// a poll loop. Every method that touches receive buffers takes
203/// `pool: &mut dyn BufferAccess` so the frontend retains ownership
204/// of guest memory state.
205///
206/// Typical poll loop:
207/// ```text
208/// loop {
209///     poll_ready(cx, pool)  // wait for backend events
210///     rx_poll(pool, ..)     // drain completed receives
211///     tx_avail(pool, ..)    // post guest TX packets
212///     tx_poll(pool, ..)     // drain TX completions
213/// }
214/// ```
215#[async_trait]
216pub trait Queue: Send + InspectMut {
217    /// Updates the queue's target VP.
218    async fn update_target_vp(&mut self, target_vp: u32) {
219        let _ = target_vp;
220    }
221
222    /// Polls the queue for readiness.
223    fn poll_ready(&mut self, cx: &mut Context<'_>, pool: &mut dyn BufferAccess) -> Poll<()>;
224
225    /// Makes receive buffers available for use by the device.
226    fn rx_avail(&mut self, pool: &mut dyn BufferAccess, done: &[RxId]);
227
228    /// Polls the device for receives.
229    fn rx_poll(
230        &mut self,
231        pool: &mut dyn BufferAccess,
232        packets: &mut [RxId],
233    ) -> anyhow::Result<usize>;
234
235    /// Posts transmits to the device.
236    ///
237    /// Returns `Ok(false)` if the segments will complete asynchronously.
238    fn tx_avail(
239        &mut self,
240        pool: &mut dyn BufferAccess,
241        segments: &[TxSegment],
242    ) -> anyhow::Result<(bool, usize)>;
243
244    /// Polls the device for transmit completions.
245    fn tx_poll(&mut self, pool: &mut dyn BufferAccess, done: &mut [TxId])
246    -> Result<usize, TxError>;
247
248    /// Get queue statistics
249    fn queue_stats(&self) -> Option<&dyn BackendQueueStats> {
250        None // Default implementation - not all queues implement stats
251    }
252}
253
254/// Frontend-owned access to guest receive buffers.
255///
256/// Each frontend implements this trait to map [`RxId`] values to
257/// guest memory regions. The backend writes received packet data
258/// and metadata through these methods.
259///
260/// The frontend owns the `BufferAccess` and passes `&mut` references
261/// to [`Queue`] methods. This means no `Arc`/`Mutex` is needed
262/// between the frontend and backend for buffer access—the borrow
263/// checker enforces exclusive access statically.
264pub trait BufferAccess {
265    /// The associated guest memory accessor.
266    fn guest_memory(&self) -> &GuestMemory;
267
268    /// Writes data to the specified buffer.
269    fn write_data(&mut self, id: RxId, data: &[u8]);
270
271    /// Appends the guest address segments for the specified buffer to `buf`.
272    ///
273    /// Callers must clear `buf` before calling if they do not want segments
274    /// from a previous call to be retained.
275    fn push_guest_addresses(&self, id: RxId, buf: &mut Vec<RxBufferSegment>);
276
277    /// The capacity of the specified buffer in bytes.
278    fn capacity(&self, id: RxId) -> u32;
279
280    /// Sets the packet metadata for the receive.
281    fn write_header(&mut self, id: RxId, metadata: &RxMetadata);
282
283    /// Writes the packet header and data in a single call.
284    fn write_packet(&mut self, id: RxId, metadata: &RxMetadata, data: &[u8]) {
285        self.write_data(id, data);
286        self.write_header(id, metadata);
287    }
288
289    /// Writes the packet header and a payload composed of multiple
290    /// discontiguous segments, in order, as a single logical packet.
291    ///
292    /// This allows callers to hand off a frame whose bytes are not contiguous
293    /// in memory (for example, an Ethernet/IP/TCP header followed by payload
294    /// that wraps a ring buffer) without first linearizing it into a scratch
295    /// buffer.
296    ///
297    /// The default implementation copies the segments into a temporary
298    /// contiguous buffer and forwards to [`BufferAccess::write_packet`].
299    /// Backends that write directly into guest memory should override this to
300    /// write each segment at its running offset and avoid the copy.
301    fn write_packet_segments(&mut self, id: RxId, metadata: &RxMetadata, segments: &[&[u8]]) {
302        if let [segment] = segments {
303            self.write_packet(id, metadata, segment);
304            return;
305        }
306        let total = segments.iter().map(|s| s.len()).sum();
307        let mut data = Vec::with_capacity(total);
308        for segment in segments {
309            data.extend_from_slice(segment);
310        }
311        self.write_packet(id, metadata, &data);
312    }
313}
314
315pub const ETHERNET_HEADER_LEN: u32 = 14;
316pub const ETHERNET_VLAN_HEADER_LEN: u32 = 18;
317
318pub const IPV4_MIN_HEADER_LEN: u16 = 20;
319pub const IPV6_MIN_HEADER_LEN: u16 = 40;
320
321#[bitfield(u16)]
322pub struct VlanMetadata {
323    /// Priority for 802.1Q.
324    #[bits(3)]
325    pub priority: u8,
326    /// In pretty much every circumstance this is false. When
327    /// it is used, setting DEI will inform switches/routing infra
328    /// that this can be dropped before higher priority traffic.
329    pub drop_eligible_indicator: bool,
330    /// The 802.1Q ID for this transmission.
331    #[bits(12)]
332    pub vlan_id: u16,
333}
334
335/// A receive buffer ID.
336#[derive(Debug, Copy, Clone)]
337#[repr(transparent)]
338pub struct RxId(pub u32);
339
340/// An individual segment in guest memory of a receive buffer.
341#[derive(Debug, Copy, Clone)]
342pub struct RxBufferSegment {
343    /// Guest physical address.
344    pub gpa: u64,
345    /// The number of bytes in this range.
346    pub len: u32,
347}
348
349/// Receive packet metadata.
350#[derive(Debug, Copy, Clone)]
351pub struct RxMetadata {
352    /// The offset of the packet data from the beginning of the receive buffer.
353    pub offset: usize,
354    /// The length of the packet in bytes.
355    pub len: usize,
356    /// The IP checksum validation state.
357    pub ip_checksum: RxChecksumState,
358    /// The L4 checksum validation state.
359    pub l4_checksum: RxChecksumState,
360    /// The L4 protocol.
361    pub l4_protocol: L4Protocol,
362    /// Information about 802.1Q VLAN tagging. When a vlan is in use, this structure
363    /// is populated. Only applies when traffic is being received over an L2 connection,
364    /// so L3-only or above traffic will not use this option.
365    pub vlan: Option<VlanMetadata>,
366}
367
368impl Default for RxMetadata {
369    fn default() -> Self {
370        Self {
371            offset: 0,
372            len: 0,
373            ip_checksum: RxChecksumState::Unknown,
374            l4_checksum: RxChecksumState::Unknown,
375            l4_protocol: L4Protocol::Unknown,
376            vlan: None,
377        }
378    }
379}
380
381/// The "L4" protocol: the TCP/UDP layer.
382#[derive(Debug, Copy, Clone, PartialEq, Eq)]
383pub enum L4Protocol {
384    Unknown,
385    Tcp,
386    Udp,
387}
388
389/// The receive checksum state for a packet.
390#[derive(Debug, Copy, Clone, PartialEq, Eq)]
391pub enum RxChecksumState {
392    /// The checksum was not evaluated.
393    Unknown,
394    /// The checksum value is correct.
395    Good,
396    /// The checksum value is incorrect.
397    Bad,
398    /// The checksum has been validated, but the value in the header is wrong.
399    ///
400    /// This occurs when LRO/RSC offload has been performed--multiple packet
401    /// payloads are glommed together without updating the checksum in the first
402    /// packet's header.
403    ValidatedButWrong,
404}
405
406impl RxChecksumState {
407    /// Returns true if the checksum has been validated.
408    pub fn is_valid(self) -> bool {
409        self == Self::Good || self == Self::ValidatedButWrong
410    }
411}
412
413/// A transmit ID. This may be used by multiple segments at the same time.
414#[derive(Debug, Copy, Clone)]
415#[repr(transparent)]
416pub struct TxId(pub u32);
417
418#[derive(Debug, Clone)]
419/// The segment type.
420pub enum TxSegmentType {
421    /// The start of a packet.
422    Head(TxMetadata),
423    /// A packet continuation.
424    Tail,
425}
426
427#[derive(Debug, Clone)]
428/// Transmit packet metadata.
429pub struct TxMetadata {
430    /// The transmit ID.
431    pub id: TxId,
432    /// The number of segments, including this one.
433    pub segment_count: u8,
434    /// Flags.
435    pub flags: TxFlags,
436    /// The total length of the packet in bytes.
437    pub len: u32,
438    /// The length of the Ethernet frame header. Only guaranteed to be set if
439    /// various offload flags are set.
440    pub l2_len: u8,
441    /// The length of the IP header. Only guaranteed to be set if various
442    /// offload flags are set.
443    pub l3_len: u16,
444    /// The length of the TCP header. Only guaranteed to be set if various
445    /// offload flags are set.
446    pub l4_len: u8,
447    /// The offset into the buffer where the L4 header begins (TCP or UDP). Only
448    /// expected to be set if offload (checksum and/or segmentation) flags are set.
449    pub transport_header_offset: u16,
450    /// The maximum segment size, used for segmentation offload (TSO or USO).
451    /// Only guaranteed to be set if [`TxFlags::offload_tcp_segmentation`] or
452    /// [`TxFlags::offload_udp_segmentation`] is set.
453    pub max_segment_size: u16,
454    /// Information about 802.1Q VLAN tagging. When a vlan is in use, this structure
455    /// is populated. Only applies when traffic is being sent over an L2 connection,
456    /// so L3-only or above traffic will not use this option.
457    pub vlan: Option<VlanMetadata>,
458}
459
460/// Flags affecting transmit behavior.
461#[bitfield(u8)]
462pub struct TxFlags {
463    /// Offload IPv4 header checksum calculation.
464    ///
465    /// `l3_protocol`, `l2_len`, and `l3_len` must be set.
466    pub offload_ip_header_checksum: bool,
467    /// Offload the TCP checksum calculation.
468    ///
469    /// `l3_protocol`, `l2_len`, and `l3_len` must be set.
470    pub offload_tcp_checksum: bool,
471    /// Offload the UDP checksum calculation.
472    ///
473    /// `l3_protocol`, `l2_len`, and `l3_len` must be set.
474    pub offload_udp_checksum: bool,
475    /// Offload the TCP segmentation, allowing packets to be larger than the
476    /// MTU.
477    ///
478    /// `l3_protocol`, `l2_len`, `l3_len`, `l4_len`, and `tcp_segment_size` must
479    /// be set.
480    pub offload_tcp_segmentation: bool,
481    /// If true, the packet is IPv4.
482    pub is_ipv4: bool,
483    /// If true, the packet is IPv6. Mutually exclusive with `is_ipv4`.
484    pub is_ipv6: bool,
485    /// Offload UDP segmentation (USO), allowing UDP packets larger than the
486    /// MTU. `l2_len`, `l3_len`, and `max_segment_size` must be set.
487    pub offload_udp_segmentation: bool,
488    #[bits(1)]
489    _reserved: u8,
490}
491
492impl Default for TxMetadata {
493    fn default() -> Self {
494        Self {
495            id: TxId(0),
496            segment_count: 0,
497            len: 0,
498            flags: TxFlags::new(),
499            l2_len: 0,
500            l3_len: 0,
501            l4_len: 0,
502            transport_header_offset: 0,
503            max_segment_size: 0,
504            vlan: None,
505        }
506    }
507}
508
509#[derive(Debug, Clone)]
510/// A transmit packet segment.
511pub struct TxSegment {
512    /// The segment type (head or tail).
513    pub ty: TxSegmentType,
514    /// The guest address of this segment.
515    pub gpa: u64,
516    /// The length of this segment.
517    pub len: u32,
518}
519
520/// Computes the number of packets in `segments`.
521pub fn packet_count(mut segments: &[TxSegment]) -> usize {
522    let mut packet_count = 0;
523    while let Some(head) = segments.first() {
524        let TxSegmentType::Head(metadata) = &head.ty else {
525            unreachable!()
526        };
527        segments = &segments[metadata.segment_count as usize..];
528        packet_count += 1;
529    }
530    packet_count
531}
532
533/// Gets the next packet from a list of segments, returning the packet metadata,
534/// the segments in the packet, and the remaining segments.
535pub fn next_packet(segments: &[TxSegment]) -> (&TxMetadata, &[TxSegment], &[TxSegment]) {
536    let metadata = if let TxSegmentType::Head(metadata) = &segments[0].ty {
537        metadata
538    } else {
539        unreachable!();
540    };
541    let (this, rest) = segments.split_at(metadata.segment_count.into());
542    (metadata, this, rest)
543}
544
545/// Linearizes the next packet in a list of segments, returning the buffer data
546/// and advancing the segment list.
547pub fn linearize(
548    pool: &dyn BufferAccess,
549    segments: &mut &[TxSegment],
550) -> Result<Vec<u8>, GuestMemoryError> {
551    let (head, this, rest) = next_packet(segments);
552    let mut v = vec![0; head.len as usize];
553    let mut offset = 0;
554    let mem = pool.guest_memory();
555    for segment in this {
556        let dest = &mut v[offset..offset + segment.len as usize];
557        mem.read_at(segment.gpa, dest)?;
558        offset += segment.len as usize;
559    }
560    assert_eq!(v.len(), offset);
561    *segments = rest;
562    Ok(v)
563}
564
565#[derive(PartialEq, Debug)]
566pub enum EndpointAction {
567    RestartRequired,
568    LinkStatusNotify(bool),
569}
570
571enum DisconnectableEndpointUpdate {
572    EndpointConnected(Box<dyn Endpoint>),
573    EndpointDisconnected(Rpc<(), Option<Box<dyn Endpoint>>>),
574}
575
576pub struct DisconnectableEndpointControl {
577    send_update: mesh::Sender<DisconnectableEndpointUpdate>,
578    is_ordered: Option<bool>,
579}
580
581impl DisconnectableEndpointControl {
582    pub fn connect(&mut self, endpoint: Box<dyn Endpoint>) -> anyhow::Result<()> {
583        let new_is_ordered = endpoint.is_ordered();
584        if let Some(is_ordered) = self.is_ordered {
585            anyhow::ensure!(
586                !is_ordered || new_is_ordered,
587                "network endpoint cannot be reattached as unordered after being ordered"
588            );
589        } else {
590            self.is_ordered = Some(new_is_ordered);
591        }
592        self.send_update
593            .send(DisconnectableEndpointUpdate::EndpointConnected(endpoint));
594        Ok(())
595    }
596
597    pub async fn disconnect(&mut self) -> anyhow::Result<Option<Box<dyn Endpoint>>> {
598        self.send_update
599            .call(DisconnectableEndpointUpdate::EndpointDisconnected, ())
600            .map_err(anyhow::Error::from)
601            .await
602    }
603}
604
605pub struct DisconnectableEndpointCachedState {
606    is_ordered: bool,
607    tx_offload_support: TxOffloadSupport,
608    multiqueue_support: MultiQueueSupport,
609    tx_fast_completions: bool,
610    link_speed: u64,
611}
612
613pub struct DisconnectableEndpoint {
614    endpoint: Option<Box<dyn Endpoint>>,
615    null_endpoint: Box<dyn Endpoint>,
616    cached_state: Option<DisconnectableEndpointCachedState>,
617    receive_update: Arc<Mutex<mesh::Receiver<DisconnectableEndpointUpdate>>>,
618    notify_disconnect_complete: Option<(
619        Rpc<(), Option<Box<dyn Endpoint>>>,
620        Option<Box<dyn Endpoint>>,
621    )>,
622}
623
624impl InspectMut for DisconnectableEndpoint {
625    fn inspect_mut(&mut self, req: inspect::Request<'_>) {
626        self.current_mut().inspect_mut(req)
627    }
628}
629
630impl DisconnectableEndpoint {
631    pub fn new() -> (Self, DisconnectableEndpointControl) {
632        let (endpoint_tx, endpoint_rx) = mesh::channel();
633        let control = DisconnectableEndpointControl {
634            send_update: endpoint_tx,
635            is_ordered: None,
636        };
637        (
638            Self {
639                endpoint: None,
640                null_endpoint: Box::new(NullEndpoint::new()),
641                cached_state: None,
642                receive_update: Arc::new(Mutex::new(endpoint_rx)),
643                notify_disconnect_complete: None,
644            },
645            control,
646        )
647    }
648
649    fn current(&self) -> &dyn Endpoint {
650        self.endpoint
651            .as_ref()
652            .unwrap_or(&self.null_endpoint)
653            .as_ref()
654    }
655
656    fn current_mut(&mut self) -> &mut dyn Endpoint {
657        self.endpoint
658            .as_mut()
659            .unwrap_or(&mut self.null_endpoint)
660            .as_mut()
661    }
662}
663
664#[async_trait]
665impl Endpoint for DisconnectableEndpoint {
666    fn endpoint_type(&self) -> &'static str {
667        self.current().endpoint_type()
668    }
669
670    async fn get_queues(
671        &mut self,
672        config: Vec<QueueConfig>,
673        rss: Option<&RssConfig<'_>>,
674        queues: &mut Vec<Box<dyn Queue>>,
675    ) -> anyhow::Result<()> {
676        self.current_mut().get_queues(config, rss, queues).await
677    }
678
679    async fn stop(&mut self) {
680        self.current_mut().stop().await
681    }
682
683    fn is_ordered(&self) -> bool {
684        self.cached_state
685            .as_ref()
686            .expect("Endpoint needs connected at least once before use")
687            .is_ordered
688    }
689
690    fn tx_offload_support(&self) -> TxOffloadSupport {
691        self.cached_state
692            .as_ref()
693            .expect("Endpoint needs connected at least once before use")
694            .tx_offload_support
695    }
696
697    fn multiqueue_support(&self) -> MultiQueueSupport {
698        self.cached_state
699            .as_ref()
700            .expect("Endpoint needs connected at least once before use")
701            .multiqueue_support
702    }
703
704    fn tx_fast_completions(&self) -> bool {
705        self.cached_state
706            .as_ref()
707            .expect("Endpoint needs connected at least once before use")
708            .tx_fast_completions
709    }
710
711    async fn set_data_path_to_guest_vf(&self, use_vf: bool) -> anyhow::Result<()> {
712        self.current().set_data_path_to_guest_vf(use_vf).await
713    }
714
715    async fn get_data_path_to_guest_vf(&self) -> anyhow::Result<bool> {
716        self.current().get_data_path_to_guest_vf().await
717    }
718
719    async fn wait_for_endpoint_action(&mut self) -> EndpointAction {
720        // If the previous message disconnected the endpoint, notify the caller
721        // that the operation has completed, returning the old endpoint.
722        if let Some((rpc, old_endpoint)) = self.notify_disconnect_complete.take() {
723            rpc.handle(async |_| old_endpoint).await;
724        }
725
726        enum Message {
727            DisconnectableEndpointUpdate(DisconnectableEndpointUpdate),
728            UpdateFromEndpoint(EndpointAction),
729        }
730        let receiver = self.receive_update.clone();
731        let mut receive_update = receiver.lock().await;
732        let update = async {
733            match receive_update.next().await {
734                Some(m) => Message::DisconnectableEndpointUpdate(m),
735                None => {
736                    pending::<()>().await;
737                    unreachable!()
738                }
739            }
740        };
741        let ep_update = self
742            .current_mut()
743            .wait_for_endpoint_action()
744            .map(Message::UpdateFromEndpoint);
745        let m = (update, ep_update).race().await;
746        match m {
747            Message::DisconnectableEndpointUpdate(
748                DisconnectableEndpointUpdate::EndpointConnected(endpoint),
749            ) => {
750                let old_endpoint = self.endpoint.take();
751                assert!(old_endpoint.is_none());
752                self.endpoint = Some(endpoint);
753                let new_is_ordered = self.current().is_ordered();
754                let is_ordered = if let Some(prev) = &self.cached_state {
755                    assert!(
756                        !prev.is_ordered || new_is_ordered,
757                        "network endpoint reattached as unordered after being ordered"
758                    );
759                    prev.is_ordered
760                } else {
761                    new_is_ordered
762                };
763                self.cached_state = Some(DisconnectableEndpointCachedState {
764                    is_ordered,
765                    tx_offload_support: self.current().tx_offload_support(),
766                    multiqueue_support: self.current().multiqueue_support(),
767                    tx_fast_completions: self.current().tx_fast_completions(),
768                    link_speed: self.current().link_speed(),
769                });
770                EndpointAction::RestartRequired
771            }
772            Message::DisconnectableEndpointUpdate(
773                DisconnectableEndpointUpdate::EndpointDisconnected(rpc),
774            ) => {
775                let old_endpoint = self.endpoint.take();
776                // Wait until the next call into this function to notify the
777                // caller that the operation has completed. This makes it more
778                // likely that the endpoint is no longer referenced (old queues
779                // have been disposed, etc.).
780                self.notify_disconnect_complete = Some((rpc, old_endpoint));
781                EndpointAction::RestartRequired
782            }
783            Message::UpdateFromEndpoint(update) => update,
784        }
785    }
786
787    fn link_speed(&self) -> u64 {
788        self.cached_state
789            .as_ref()
790            .expect("Endpoint needs connected at least once before use")
791            .link_speed
792    }
793}
794
795#[cfg(test)]
796mod disconnectable_endpoint_tests {
797    use super::*;
798    use test_with_tracing::test;
799
800    #[derive(InspectMut)]
801    struct TestEndpoint {
802        is_ordered: bool,
803    }
804
805    #[async_trait]
806    impl Endpoint for TestEndpoint {
807        fn endpoint_type(&self) -> &'static str {
808            "test"
809        }
810
811        async fn get_queues(
812            &mut self,
813            _config: Vec<QueueConfig>,
814            _rss: Option<&RssConfig<'_>>,
815            _queues: &mut Vec<Box<dyn Queue>>,
816        ) -> anyhow::Result<()> {
817            unreachable!()
818        }
819
820        async fn stop(&mut self) {
821            unreachable!()
822        }
823
824        fn is_ordered(&self) -> bool {
825            self.is_ordered
826        }
827    }
828
829    #[test]
830    fn connect_pins_endpoint_ordering() {
831        let (_endpoint, mut control) = DisconnectableEndpoint::new();
832        control
833            .connect(Box::new(TestEndpoint { is_ordered: true }))
834            .unwrap();
835
836        let err = control
837            .connect(Box::new(TestEndpoint { is_ordered: false }))
838            .unwrap_err();
839        assert_eq!(
840            err.to_string(),
841            "network endpoint cannot be reattached as unordered after being ordered"
842        );
843
844        let (_endpoint, mut control) = DisconnectableEndpoint::new();
845        control
846            .connect(Box::new(TestEndpoint { is_ordered: false }))
847            .unwrap();
848        control
849            .connect(Box::new(TestEndpoint { is_ordered: true }))
850            .unwrap();
851    }
852}
net_backend/lib.rs

net_backend/
lib.rs
