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 "L3" protocol: the IP layer.
382#[derive(Debug, Copy, Clone, PartialEq, Eq)]
383pub enum L3Protocol {
384    Unknown,
385    Ipv4,
386    Ipv6,
387}
388
389/// The "L4" protocol: the TCP/UDP layer.
390#[derive(Debug, Copy, Clone, PartialEq, Eq)]
391pub enum L4Protocol {
392    Unknown,
393    Tcp,
394    Udp,
395}
396
397/// The receive checksum state for a packet.
398#[derive(Debug, Copy, Clone, PartialEq, Eq)]
399pub enum RxChecksumState {
400    /// The checksum was not evaluated.
401    Unknown,
402    /// The checksum value is correct.
403    Good,
404    /// The checksum value is incorrect.
405    Bad,
406    /// The checksum has been validated, but the value in the header is wrong.
407    ///
408    /// This occurs when LRO/RSC offload has been performed--multiple packet
409    /// payloads are glommed together without updating the checksum in the first
410    /// packet's header.
411    ValidatedButWrong,
412}
413
414impl RxChecksumState {
415    /// Returns true if the checksum has been validated.
416    pub fn is_valid(self) -> bool {
417        self == Self::Good || self == Self::ValidatedButWrong
418    }
419}
420
421/// A transmit ID. This may be used by multiple segments at the same time.
422#[derive(Debug, Copy, Clone)]
423#[repr(transparent)]
424pub struct TxId(pub u32);
425
426#[derive(Debug, Clone)]
427/// The segment type.
428pub enum TxSegmentType {
429    /// The start of a packet.
430    Head(TxMetadata),
431    /// A packet continuation.
432    Tail,
433}
434
435#[derive(Debug, Clone)]
436/// Transmit packet metadata.
437pub struct TxMetadata {
438    /// The transmit ID.
439    pub id: TxId,
440    /// The number of segments, including this one.
441    pub segment_count: u8,
442    /// Flags.
443    pub flags: TxFlags,
444    /// The total length of the packet in bytes.
445    pub len: u32,
446    /// The length of the Ethernet frame header. Only guaranteed to be set if
447    /// various offload flags are set.
448    pub l2_len: u8,
449    /// The length of the IP header. Only guaranteed to be set if various
450    /// offload flags are set.
451    pub l3_len: u16,
452    /// The length of the TCP header. Only guaranteed to be set if various
453    /// offload flags are set.
454    pub l4_len: u8,
455    /// The offset into the buffer where the L4 header begins (TCP or UDP). Only
456    /// expected to be set if offload (checksum and/or segmentation) flags are set.
457    pub transport_header_offset: u16,
458    /// The maximum segment size, used for segmentation offload (TSO or USO).
459    /// Only guaranteed to be set if [`TxFlags::offload_tcp_segmentation`] or
460    /// [`TxFlags::offload_udp_segmentation`] is set.
461    pub max_segment_size: u16,
462    /// Information about 802.1Q VLAN tagging. When a vlan is in use, this structure
463    /// is populated. Only applies when traffic is being sent over an L2 connection,
464    /// so L3-only or above traffic will not use this option.
465    pub vlan: Option<VlanMetadata>,
466}
467
468/// Flags affecting transmit behavior.
469#[bitfield(u8)]
470pub struct TxFlags {
471    /// Offload IPv4 header checksum calculation.
472    ///
473    /// `l3_protocol`, `l2_len`, and `l3_len` must be set.
474    pub offload_ip_header_checksum: bool,
475    /// Offload the TCP checksum calculation.
476    ///
477    /// `l3_protocol`, `l2_len`, and `l3_len` must be set.
478    pub offload_tcp_checksum: bool,
479    /// Offload the UDP checksum calculation.
480    ///
481    /// `l3_protocol`, `l2_len`, and `l3_len` must be set.
482    pub offload_udp_checksum: bool,
483    /// Offload the TCP segmentation, allowing packets to be larger than the
484    /// MTU.
485    ///
486    /// `l3_protocol`, `l2_len`, `l3_len`, `l4_len`, and `tcp_segment_size` must
487    /// be set.
488    pub offload_tcp_segmentation: bool,
489    /// If true, the packet is IPv4.
490    pub is_ipv4: bool,
491    /// If true, the packet is IPv6. Mutually exclusive with `is_ipv4`.
492    pub is_ipv6: bool,
493    /// Offload UDP segmentation (USO), allowing UDP packets larger than the
494    /// MTU. `l2_len`, `l3_len`, and `max_segment_size` must be set.
495    pub offload_udp_segmentation: bool,
496    #[bits(1)]
497    _reserved: u8,
498}
499
500impl Default for TxMetadata {
501    fn default() -> Self {
502        Self {
503            id: TxId(0),
504            segment_count: 0,
505            len: 0,
506            flags: TxFlags::new(),
507            l2_len: 0,
508            l3_len: 0,
509            l4_len: 0,
510            transport_header_offset: 0,
511            max_segment_size: 0,
512            vlan: None,
513        }
514    }
515}
516
517#[derive(Debug, Clone)]
518/// A transmit packet segment.
519pub struct TxSegment {
520    /// The segment type (head or tail).
521    pub ty: TxSegmentType,
522    /// The guest address of this segment.
523    pub gpa: u64,
524    /// The length of this segment.
525    pub len: u32,
526}
527
528/// Computes the number of packets in `segments`.
529pub fn packet_count(mut segments: &[TxSegment]) -> usize {
530    let mut packet_count = 0;
531    while let Some(head) = segments.first() {
532        let TxSegmentType::Head(metadata) = &head.ty else {
533            unreachable!()
534        };
535        segments = &segments[metadata.segment_count as usize..];
536        packet_count += 1;
537    }
538    packet_count
539}
540
541/// Gets the next packet from a list of segments, returning the packet metadata,
542/// the segments in the packet, and the remaining segments.
543pub fn next_packet(segments: &[TxSegment]) -> (&TxMetadata, &[TxSegment], &[TxSegment]) {
544    let metadata = if let TxSegmentType::Head(metadata) = &segments[0].ty {
545        metadata
546    } else {
547        unreachable!();
548    };
549    let (this, rest) = segments.split_at(metadata.segment_count.into());
550    (metadata, this, rest)
551}
552
553/// Linearizes the next packet in a list of segments, returning the buffer data
554/// and advancing the segment list.
555pub fn linearize(
556    pool: &dyn BufferAccess,
557    segments: &mut &[TxSegment],
558) -> Result<Vec<u8>, GuestMemoryError> {
559    let (head, this, rest) = next_packet(segments);
560    let mut v = vec![0; head.len as usize];
561    let mut offset = 0;
562    let mem = pool.guest_memory();
563    for segment in this {
564        let dest = &mut v[offset..offset + segment.len as usize];
565        mem.read_at(segment.gpa, dest)?;
566        offset += segment.len as usize;
567    }
568    assert_eq!(v.len(), offset);
569    *segments = rest;
570    Ok(v)
571}
572
573#[derive(PartialEq, Debug)]
574pub enum EndpointAction {
575    RestartRequired,
576    LinkStatusNotify(bool),
577}
578
579enum DisconnectableEndpointUpdate {
580    EndpointConnected(Box<dyn Endpoint>),
581    EndpointDisconnected(Rpc<(), Option<Box<dyn Endpoint>>>),
582}
583
584pub struct DisconnectableEndpointControl {
585    send_update: mesh::Sender<DisconnectableEndpointUpdate>,
586    is_ordered: Option<bool>,
587}
588
589impl DisconnectableEndpointControl {
590    pub fn connect(&mut self, endpoint: Box<dyn Endpoint>) -> anyhow::Result<()> {
591        let new_is_ordered = endpoint.is_ordered();
592        if let Some(is_ordered) = self.is_ordered {
593            anyhow::ensure!(
594                !is_ordered || new_is_ordered,
595                "network endpoint cannot be reattached as unordered after being ordered"
596            );
597        } else {
598            self.is_ordered = Some(new_is_ordered);
599        }
600        self.send_update
601            .send(DisconnectableEndpointUpdate::EndpointConnected(endpoint));
602        Ok(())
603    }
604
605    pub async fn disconnect(&mut self) -> anyhow::Result<Option<Box<dyn Endpoint>>> {
606        self.send_update
607            .call(DisconnectableEndpointUpdate::EndpointDisconnected, ())
608            .map_err(anyhow::Error::from)
609            .await
610    }
611}
612
613pub struct DisconnectableEndpointCachedState {
614    is_ordered: bool,
615    tx_offload_support: TxOffloadSupport,
616    multiqueue_support: MultiQueueSupport,
617    tx_fast_completions: bool,
618    link_speed: u64,
619}
620
621pub struct DisconnectableEndpoint {
622    endpoint: Option<Box<dyn Endpoint>>,
623    null_endpoint: Box<dyn Endpoint>,
624    cached_state: Option<DisconnectableEndpointCachedState>,
625    receive_update: Arc<Mutex<mesh::Receiver<DisconnectableEndpointUpdate>>>,
626    notify_disconnect_complete: Option<(
627        Rpc<(), Option<Box<dyn Endpoint>>>,
628        Option<Box<dyn Endpoint>>,
629    )>,
630}
631
632impl InspectMut for DisconnectableEndpoint {
633    fn inspect_mut(&mut self, req: inspect::Request<'_>) {
634        self.current_mut().inspect_mut(req)
635    }
636}
637
638impl DisconnectableEndpoint {
639    pub fn new() -> (Self, DisconnectableEndpointControl) {
640        let (endpoint_tx, endpoint_rx) = mesh::channel();
641        let control = DisconnectableEndpointControl {
642            send_update: endpoint_tx,
643            is_ordered: None,
644        };
645        (
646            Self {
647                endpoint: None,
648                null_endpoint: Box::new(NullEndpoint::new()),
649                cached_state: None,
650                receive_update: Arc::new(Mutex::new(endpoint_rx)),
651                notify_disconnect_complete: None,
652            },
653            control,
654        )
655    }
656
657    fn current(&self) -> &dyn Endpoint {
658        self.endpoint
659            .as_ref()
660            .unwrap_or(&self.null_endpoint)
661            .as_ref()
662    }
663
664    fn current_mut(&mut self) -> &mut dyn Endpoint {
665        self.endpoint
666            .as_mut()
667            .unwrap_or(&mut self.null_endpoint)
668            .as_mut()
669    }
670}
671
672#[async_trait]
673impl Endpoint for DisconnectableEndpoint {
674    fn endpoint_type(&self) -> &'static str {
675        self.current().endpoint_type()
676    }
677
678    async fn get_queues(
679        &mut self,
680        config: Vec<QueueConfig>,
681        rss: Option<&RssConfig<'_>>,
682        queues: &mut Vec<Box<dyn Queue>>,
683    ) -> anyhow::Result<()> {
684        self.current_mut().get_queues(config, rss, queues).await
685    }
686
687    async fn stop(&mut self) {
688        self.current_mut().stop().await
689    }
690
691    fn is_ordered(&self) -> bool {
692        self.cached_state
693            .as_ref()
694            .expect("Endpoint needs connected at least once before use")
695            .is_ordered
696    }
697
698    fn tx_offload_support(&self) -> TxOffloadSupport {
699        self.cached_state
700            .as_ref()
701            .expect("Endpoint needs connected at least once before use")
702            .tx_offload_support
703    }
704
705    fn multiqueue_support(&self) -> MultiQueueSupport {
706        self.cached_state
707            .as_ref()
708            .expect("Endpoint needs connected at least once before use")
709            .multiqueue_support
710    }
711
712    fn tx_fast_completions(&self) -> bool {
713        self.cached_state
714            .as_ref()
715            .expect("Endpoint needs connected at least once before use")
716            .tx_fast_completions
717    }
718
719    async fn set_data_path_to_guest_vf(&self, use_vf: bool) -> anyhow::Result<()> {
720        self.current().set_data_path_to_guest_vf(use_vf).await
721    }
722
723    async fn get_data_path_to_guest_vf(&self) -> anyhow::Result<bool> {
724        self.current().get_data_path_to_guest_vf().await
725    }
726
727    async fn wait_for_endpoint_action(&mut self) -> EndpointAction {
728        // If the previous message disconnected the endpoint, notify the caller
729        // that the operation has completed, returning the old endpoint.
730        if let Some((rpc, old_endpoint)) = self.notify_disconnect_complete.take() {
731            rpc.handle(async |_| old_endpoint).await;
732        }
733
734        enum Message {
735            DisconnectableEndpointUpdate(DisconnectableEndpointUpdate),
736            UpdateFromEndpoint(EndpointAction),
737        }
738        let receiver = self.receive_update.clone();
739        let mut receive_update = receiver.lock().await;
740        let update = async {
741            match receive_update.next().await {
742                Some(m) => Message::DisconnectableEndpointUpdate(m),
743                None => {
744                    pending::<()>().await;
745                    unreachable!()
746                }
747            }
748        };
749        let ep_update = self
750            .current_mut()
751            .wait_for_endpoint_action()
752            .map(Message::UpdateFromEndpoint);
753        let m = (update, ep_update).race().await;
754        match m {
755            Message::DisconnectableEndpointUpdate(
756                DisconnectableEndpointUpdate::EndpointConnected(endpoint),
757            ) => {
758                let old_endpoint = self.endpoint.take();
759                assert!(old_endpoint.is_none());
760                self.endpoint = Some(endpoint);
761                let new_is_ordered = self.current().is_ordered();
762                let is_ordered = if let Some(prev) = &self.cached_state {
763                    assert!(
764                        !prev.is_ordered || new_is_ordered,
765                        "network endpoint reattached as unordered after being ordered"
766                    );
767                    prev.is_ordered
768                } else {
769                    new_is_ordered
770                };
771                self.cached_state = Some(DisconnectableEndpointCachedState {
772                    is_ordered,
773                    tx_offload_support: self.current().tx_offload_support(),
774                    multiqueue_support: self.current().multiqueue_support(),
775                    tx_fast_completions: self.current().tx_fast_completions(),
776                    link_speed: self.current().link_speed(),
777                });
778                EndpointAction::RestartRequired
779            }
780            Message::DisconnectableEndpointUpdate(
781                DisconnectableEndpointUpdate::EndpointDisconnected(rpc),
782            ) => {
783                let old_endpoint = self.endpoint.take();
784                // Wait until the next call into this function to notify the
785                // caller that the operation has completed. This makes it more
786                // likely that the endpoint is no longer referenced (old queues
787                // have been disposed, etc.).
788                self.notify_disconnect_complete = Some((rpc, old_endpoint));
789                EndpointAction::RestartRequired
790            }
791            Message::UpdateFromEndpoint(update) => update,
792        }
793    }
794
795    fn link_speed(&self) -> u64 {
796        self.cached_state
797            .as_ref()
798            .expect("Endpoint needs connected at least once before use")
799            .link_speed
800    }
801}
802
803#[cfg(test)]
804mod disconnectable_endpoint_tests {
805    use super::*;
806    use test_with_tracing::test;
807
808    #[derive(InspectMut)]
809    struct TestEndpoint {
810        is_ordered: bool,
811    }
812
813    #[async_trait]
814    impl Endpoint for TestEndpoint {
815        fn endpoint_type(&self) -> &'static str {
816            "test"
817        }
818
819        async fn get_queues(
820            &mut self,
821            _config: Vec<QueueConfig>,
822            _rss: Option<&RssConfig<'_>>,
823            _queues: &mut Vec<Box<dyn Queue>>,
824        ) -> anyhow::Result<()> {
825            unreachable!()
826        }
827
828        async fn stop(&mut self) {
829            unreachable!()
830        }
831
832        fn is_ordered(&self) -> bool {
833            self.is_ordered
834        }
835    }
836
837    #[test]
838    fn connect_pins_endpoint_ordering() {
839        let (_endpoint, mut control) = DisconnectableEndpoint::new();
840        control
841            .connect(Box::new(TestEndpoint { is_ordered: true }))
842            .unwrap();
843
844        let err = control
845            .connect(Box::new(TestEndpoint { is_ordered: false }))
846            .unwrap_err();
847        assert_eq!(
848            err.to_string(),
849            "network endpoint cannot be reattached as unordered after being ordered"
850        );
851
852        let (_endpoint, mut control) = DisconnectableEndpoint::new();
853        control
854            .connect(Box::new(TestEndpoint { is_ordered: false }))
855            .unwrap();
856        control
857            .connect(Box::new(TestEndpoint { is_ordered: true }))
858            .unwrap();
859    }
860}
net_backend/lib.rs

net_backend/
lib.rs
