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 futures::FutureExt;
45use futures::StreamExt;
46use futures::TryFutureExt;
47use futures::lock::Mutex;
48use futures_concurrency::future::Race;
49use guestmem::GuestMemory;
50use guestmem::GuestMemoryError;
51use inspect::InspectMut;
52use inspect_counters::Counter;
53use mesh::rpc::Rpc;
54use mesh::rpc::RpcSend;
55use null::NullEndpoint;
56use pal_async::driver::Driver;
57use std::future::pending;
58use std::sync::Arc;
59use std::task::Context;
60use std::task::Poll;
61use thiserror::Error;
62
63pub use net_backend_core::*;
64
65/// Per-queue configuration passed to [`Endpoint::get_queues`].
66///
67/// Contains only an async driver handle. Receive buffers are posted
68/// separately via [`Queue::rx_avail`] after queue creation.
69pub struct QueueConfig {
70    pub driver: Box<dyn Driver>,
71}
72
73/// A network endpoint — the backend side of a NIC.
74///
75/// An endpoint is a factory for [`Queue`] objects. It represents a
76/// connection to some packet transport (TAP device, hardware NIC,
77/// user-space network stack, etc.) and can create one or more queues
78/// for parallel TX/RX processing.
79///
80/// Frontends (e.g. `virtio_net`, `netvsp`, `gdma`) own the endpoint
81/// and call [`get_queues`](Endpoint::get_queues) when the guest
82/// activates the NIC.
83#[async_trait]
84pub trait Endpoint: Send + Sync + InspectMut {
85    /// Returns an informational endpoint type.
86    fn endpoint_type(&self) -> &'static str;
87
88    /// Initializes the queues associated with the endpoint.
89    async fn get_queues(
90        &mut self,
91        config: Vec<QueueConfig>,
92        rss: Option<&RssConfig<'_>>,
93        queues: &mut Vec<Box<dyn Queue>>,
94    ) -> anyhow::Result<()>;
95
96    /// Stops the endpoint.
97    ///
98    /// All queues returned via `get_queues` must have been dropped.
99    async fn stop(&mut self);
100
101    /// Whether the endpoint completes buffers in the order they were made
102    /// available (RX buffers returned from `rx_poll`, and TX packets completed,
103    /// in available-ring order).
104    fn is_ordered(&self) -> bool;
105
106    /// Specifies the supported set of transmit offloads.
107    fn tx_offload_support(&self) -> TxOffloadSupport {
108        TxOffloadSupport::default()
109    }
110
111    /// Specifies parameters related to supporting multiple queues.
112    fn multiqueue_support(&self) -> MultiQueueSupport {
113        MultiQueueSupport {
114            max_queues: 1,
115            indirection_table_size: 0,
116        }
117    }
118
119    /// If true, transmits are guaranteed to complete quickly. This is used to
120    /// allow eliding tx notifications from the guest when there are already
121    /// some tx packets in flight.
122    fn tx_fast_completions(&self) -> bool {
123        false
124    }
125
126    /// Sets the current data path for packet flow (e.g. via vmbus synthnic or through virtual function).
127    /// This is only supported for endpoints that pair with an accelerated device.
128    async fn set_data_path_to_guest_vf(&self, _use_vf: bool) -> anyhow::Result<()> {
129        Err(anyhow::Error::msg("Unsupported in current endpoint"))
130    }
131
132    async fn get_data_path_to_guest_vf(&self) -> anyhow::Result<bool> {
133        Err(anyhow::Error::msg("Unsupported in current endpoint"))
134    }
135
136    /// On completion, the return value indicates the specific endpoint action to take.
137    async fn wait_for_endpoint_action(&mut self) -> EndpointAction {
138        pending().await
139    }
140
141    /// Link speed in bps.
142    fn link_speed(&self) -> u64 {
143        // Reporting a reasonable default value (10Gbps) here that the individual endpoints
144        // can overwrite.
145        10 * 1000 * 1000 * 1000
146    }
147}
148
149#[derive(Error, Debug)]
150pub enum TxError {
151    #[error("error requiring queue restart. {0}")]
152    TryRestart(#[source] anyhow::Error),
153    #[error("unrecoverable error. {0}")]
154    Fatal(#[source] anyhow::Error),
155}
156pub trait BackendQueueStats {
157    fn rx_errors(&self) -> Counter;
158    fn tx_errors(&self) -> Counter;
159    fn rx_packets(&self) -> Counter;
160    fn tx_packets(&self) -> Counter;
161    fn tx_vlan_packets(&self) -> Counter {
162        Counter::new()
163    }
164    fn rx_vlan_packets(&self) -> Counter {
165        Counter::new()
166    }
167}
168
169/// A single TX/RX data path for sending and receiving network packets.
170///
171/// Created by [`Endpoint::get_queues`] and driven by the frontend in
172/// a poll loop. Every method that touches receive buffers takes
173/// `pool: &mut dyn BufferAccess` so the frontend retains ownership
174/// of guest memory state.
175///
176/// Typical poll loop:
177/// ```text
178/// loop {
179///     poll_ready(cx, pool)  // wait for backend events
180///     rx_poll(pool, ..)     // drain completed receives
181///     tx_avail(pool, ..)    // post guest TX packets
182///     tx_poll(pool, ..)     // drain TX completions
183/// }
184/// ```
185#[async_trait]
186pub trait Queue: Send + InspectMut {
187    /// Updates the queue's target VP.
188    async fn update_target_vp(&mut self, target_vp: u32) {
189        let _ = target_vp;
190    }
191
192    /// Polls the queue for readiness.
193    fn poll_ready(&mut self, cx: &mut Context<'_>, pool: &mut dyn BufferAccess) -> Poll<()>;
194
195    /// Makes receive buffers available for use by the device.
196    fn rx_avail(&mut self, pool: &mut dyn BufferAccess, done: &[RxId]);
197
198    /// Polls the device for receives.
199    fn rx_poll(
200        &mut self,
201        pool: &mut dyn BufferAccess,
202        packets: &mut [RxId],
203    ) -> anyhow::Result<usize>;
204
205    /// Posts transmits to the device.
206    ///
207    /// Returns `Ok(false)` if the segments will complete asynchronously.
208    fn tx_avail(
209        &mut self,
210        pool: &mut dyn BufferAccess,
211        segments: &[TxSegment],
212    ) -> anyhow::Result<(bool, usize)>;
213
214    /// Polls the device for transmit completions.
215    fn tx_poll(&mut self, pool: &mut dyn BufferAccess, done: &mut [TxId])
216    -> Result<usize, TxError>;
217
218    /// Get queue statistics
219    fn queue_stats(&self) -> Option<&dyn BackendQueueStats> {
220        None // Default implementation - not all queues implement stats
221    }
222}
223
224/// Frontend-owned access to guest receive buffers.
225///
226/// Each frontend implements this trait to map [`RxId`] values to
227/// guest memory regions. The backend writes received packet data
228/// and metadata through these methods.
229///
230/// The frontend owns the `BufferAccess` and passes `&mut` references
231/// to [`Queue`] methods. This means no `Arc`/`Mutex` is needed
232/// between the frontend and backend for buffer access—the borrow
233/// checker enforces exclusive access statically.
234pub trait BufferAccess {
235    /// The associated guest memory accessor.
236    fn guest_memory(&self) -> &GuestMemory;
237
238    /// Writes data to the specified buffer.
239    fn write_data(&mut self, id: RxId, data: &[u8]);
240
241    /// Appends the guest address segments for the specified buffer to `buf`.
242    ///
243    /// Callers must clear `buf` before calling if they do not want segments
244    /// from a previous call to be retained.
245    fn push_guest_addresses(&self, id: RxId, buf: &mut Vec<RxBufferSegment>);
246
247    /// The capacity of the specified buffer in bytes.
248    fn capacity(&self, id: RxId) -> u32;
249
250    /// Sets the packet metadata for the receive.
251    fn write_header(&mut self, id: RxId, metadata: &RxMetadata);
252
253    /// Writes the packet header and data in a single call.
254    fn write_packet(&mut self, id: RxId, metadata: &RxMetadata, data: &[u8]) {
255        self.write_data(id, data);
256        self.write_header(id, metadata);
257    }
258
259    /// Writes the packet header and a payload composed of multiple
260    /// discontiguous segments, in order, as a single logical packet.
261    ///
262    /// This allows callers to hand off a frame whose bytes are not contiguous
263    /// in memory (for example, an Ethernet/IP/TCP header followed by payload
264    /// that wraps a ring buffer) without first linearizing it into a scratch
265    /// buffer.
266    ///
267    /// The default implementation copies the segments into a temporary
268    /// contiguous buffer and forwards to [`BufferAccess::write_packet`].
269    /// Backends that write directly into guest memory should override this to
270    /// write each segment at its running offset and avoid the copy.
271    fn write_packet_segments(&mut self, id: RxId, metadata: &RxMetadata, segments: &[&[u8]]) {
272        if let [segment] = segments {
273            self.write_packet(id, metadata, segment);
274            return;
275        }
276        let total = segments.iter().map(|s| s.len()).sum();
277        let mut data = Vec::with_capacity(total);
278        for segment in segments {
279            data.extend_from_slice(segment);
280        }
281        self.write_packet(id, metadata, &data);
282    }
283}
284
285/// Linearizes the next packet in a list of segments, returning the buffer data
286/// and advancing the segment list.
287pub fn linearize(
288    pool: &dyn BufferAccess,
289    segments: &mut &[TxSegment],
290) -> Result<Vec<u8>, GuestMemoryError> {
291    let (head, this, rest) = next_packet(segments);
292    let mut v = vec![0; head.len as usize];
293    let mut offset = 0;
294    let mem = pool.guest_memory();
295    for segment in this {
296        let dest = &mut v[offset..offset + segment.len as usize];
297        mem.read_at(segment.gpa, dest)?;
298        offset += segment.len as usize;
299    }
300    assert_eq!(v.len(), offset);
301    *segments = rest;
302    Ok(v)
303}
304
305enum DisconnectableEndpointUpdate {
306    EndpointConnected(Box<dyn Endpoint>),
307    EndpointDisconnected(Rpc<(), Option<Box<dyn Endpoint>>>),
308}
309
310pub struct DisconnectableEndpointControl {
311    send_update: mesh::Sender<DisconnectableEndpointUpdate>,
312    is_ordered: Option<bool>,
313}
314
315impl DisconnectableEndpointControl {
316    pub fn connect(&mut self, endpoint: Box<dyn Endpoint>) -> anyhow::Result<()> {
317        let new_is_ordered = endpoint.is_ordered();
318        if let Some(is_ordered) = self.is_ordered {
319            anyhow::ensure!(
320                !is_ordered || new_is_ordered,
321                "network endpoint cannot be reattached as unordered after being ordered"
322            );
323        } else {
324            self.is_ordered = Some(new_is_ordered);
325        }
326        self.send_update
327            .send(DisconnectableEndpointUpdate::EndpointConnected(endpoint));
328        Ok(())
329    }
330
331    pub async fn disconnect(&mut self) -> anyhow::Result<Option<Box<dyn Endpoint>>> {
332        self.send_update
333            .call(DisconnectableEndpointUpdate::EndpointDisconnected, ())
334            .map_err(anyhow::Error::from)
335            .await
336    }
337}
338
339pub struct DisconnectableEndpointCachedState {
340    is_ordered: bool,
341    tx_offload_support: TxOffloadSupport,
342    multiqueue_support: MultiQueueSupport,
343    tx_fast_completions: bool,
344    link_speed: u64,
345}
346
347pub struct DisconnectableEndpoint {
348    endpoint: Option<Box<dyn Endpoint>>,
349    null_endpoint: Box<dyn Endpoint>,
350    cached_state: Option<DisconnectableEndpointCachedState>,
351    receive_update: Arc<Mutex<mesh::Receiver<DisconnectableEndpointUpdate>>>,
352    notify_disconnect_complete: Option<(
353        Rpc<(), Option<Box<dyn Endpoint>>>,
354        Option<Box<dyn Endpoint>>,
355    )>,
356}
357
358impl InspectMut for DisconnectableEndpoint {
359    fn inspect_mut(&mut self, req: inspect::Request<'_>) {
360        self.current_mut().inspect_mut(req)
361    }
362}
363
364impl DisconnectableEndpoint {
365    pub fn new() -> (Self, DisconnectableEndpointControl) {
366        let (endpoint_tx, endpoint_rx) = mesh::channel();
367        let control = DisconnectableEndpointControl {
368            send_update: endpoint_tx,
369            is_ordered: None,
370        };
371        (
372            Self {
373                endpoint: None,
374                null_endpoint: Box::new(NullEndpoint::new()),
375                cached_state: None,
376                receive_update: Arc::new(Mutex::new(endpoint_rx)),
377                notify_disconnect_complete: None,
378            },
379            control,
380        )
381    }
382
383    fn current(&self) -> &dyn Endpoint {
384        self.endpoint
385            .as_ref()
386            .unwrap_or(&self.null_endpoint)
387            .as_ref()
388    }
389
390    fn current_mut(&mut self) -> &mut dyn Endpoint {
391        self.endpoint
392            .as_mut()
393            .unwrap_or(&mut self.null_endpoint)
394            .as_mut()
395    }
396}
397
398#[async_trait]
399impl Endpoint for DisconnectableEndpoint {
400    fn endpoint_type(&self) -> &'static str {
401        self.current().endpoint_type()
402    }
403
404    async fn get_queues(
405        &mut self,
406        config: Vec<QueueConfig>,
407        rss: Option<&RssConfig<'_>>,
408        queues: &mut Vec<Box<dyn Queue>>,
409    ) -> anyhow::Result<()> {
410        self.current_mut().get_queues(config, rss, queues).await
411    }
412
413    async fn stop(&mut self) {
414        self.current_mut().stop().await
415    }
416
417    fn is_ordered(&self) -> bool {
418        self.cached_state
419            .as_ref()
420            .expect("Endpoint needs connected at least once before use")
421            .is_ordered
422    }
423
424    fn tx_offload_support(&self) -> TxOffloadSupport {
425        self.cached_state
426            .as_ref()
427            .expect("Endpoint needs connected at least once before use")
428            .tx_offload_support
429    }
430
431    fn multiqueue_support(&self) -> MultiQueueSupport {
432        self.cached_state
433            .as_ref()
434            .expect("Endpoint needs connected at least once before use")
435            .multiqueue_support
436    }
437
438    fn tx_fast_completions(&self) -> bool {
439        self.cached_state
440            .as_ref()
441            .expect("Endpoint needs connected at least once before use")
442            .tx_fast_completions
443    }
444
445    async fn set_data_path_to_guest_vf(&self, use_vf: bool) -> anyhow::Result<()> {
446        self.current().set_data_path_to_guest_vf(use_vf).await
447    }
448
449    async fn get_data_path_to_guest_vf(&self) -> anyhow::Result<bool> {
450        self.current().get_data_path_to_guest_vf().await
451    }
452
453    async fn wait_for_endpoint_action(&mut self) -> EndpointAction {
454        // If the previous message disconnected the endpoint, notify the caller
455        // that the operation has completed, returning the old endpoint.
456        if let Some((rpc, old_endpoint)) = self.notify_disconnect_complete.take() {
457            rpc.handle(async |_| old_endpoint).await;
458        }
459
460        enum Message {
461            DisconnectableEndpointUpdate(DisconnectableEndpointUpdate),
462            UpdateFromEndpoint(EndpointAction),
463        }
464        let receiver = self.receive_update.clone();
465        let mut receive_update = receiver.lock().await;
466        let update = async {
467            match receive_update.next().await {
468                Some(m) => Message::DisconnectableEndpointUpdate(m),
469                None => {
470                    pending::<()>().await;
471                    unreachable!()
472                }
473            }
474        };
475        let ep_update = self
476            .current_mut()
477            .wait_for_endpoint_action()
478            .map(Message::UpdateFromEndpoint);
479        let m = (update, ep_update).race().await;
480        match m {
481            Message::DisconnectableEndpointUpdate(
482                DisconnectableEndpointUpdate::EndpointConnected(endpoint),
483            ) => {
484                let old_endpoint = self.endpoint.take();
485                assert!(old_endpoint.is_none());
486                self.endpoint = Some(endpoint);
487                let new_is_ordered = self.current().is_ordered();
488                let is_ordered = if let Some(prev) = &self.cached_state {
489                    assert!(
490                        !prev.is_ordered || new_is_ordered,
491                        "network endpoint reattached as unordered after being ordered"
492                    );
493                    prev.is_ordered
494                } else {
495                    new_is_ordered
496                };
497                self.cached_state = Some(DisconnectableEndpointCachedState {
498                    is_ordered,
499                    tx_offload_support: self.current().tx_offload_support(),
500                    multiqueue_support: self.current().multiqueue_support(),
501                    tx_fast_completions: self.current().tx_fast_completions(),
502                    link_speed: self.current().link_speed(),
503                });
504                EndpointAction::RestartRequired
505            }
506            Message::DisconnectableEndpointUpdate(
507                DisconnectableEndpointUpdate::EndpointDisconnected(rpc),
508            ) => {
509                let old_endpoint = self.endpoint.take();
510                // Wait until the next call into this function to notify the
511                // caller that the operation has completed. This makes it more
512                // likely that the endpoint is no longer referenced (old queues
513                // have been disposed, etc.).
514                self.notify_disconnect_complete = Some((rpc, old_endpoint));
515                EndpointAction::RestartRequired
516            }
517            Message::UpdateFromEndpoint(update) => update,
518        }
519    }
520
521    fn link_speed(&self) -> u64 {
522        self.cached_state
523            .as_ref()
524            .expect("Endpoint needs connected at least once before use")
525            .link_speed
526    }
527}
528
529#[cfg(test)]
530mod disconnectable_endpoint_tests {
531    use super::*;
532    use test_with_tracing::test;
533
534    #[derive(InspectMut)]
535    struct TestEndpoint {
536        is_ordered: bool,
537    }
538
539    #[async_trait]
540    impl Endpoint for TestEndpoint {
541        fn endpoint_type(&self) -> &'static str {
542            "test"
543        }
544
545        async fn get_queues(
546            &mut self,
547            _config: Vec<QueueConfig>,
548            _rss: Option<&RssConfig<'_>>,
549            _queues: &mut Vec<Box<dyn Queue>>,
550        ) -> anyhow::Result<()> {
551            unreachable!()
552        }
553
554        async fn stop(&mut self) {
555            unreachable!()
556        }
557
558        fn is_ordered(&self) -> bool {
559            self.is_ordered
560        }
561    }
562
563    #[test]
564    fn connect_pins_endpoint_ordering() {
565        let (_endpoint, mut control) = DisconnectableEndpoint::new();
566        control
567            .connect(Box::new(TestEndpoint { is_ordered: true }))
568            .unwrap();
569
570        let err = control
571            .connect(Box::new(TestEndpoint { is_ordered: false }))
572            .unwrap_err();
573        assert_eq!(
574            err.to_string(),
575            "network endpoint cannot be reattached as unordered after being ordered"
576        );
577
578        let (_endpoint, mut control) = DisconnectableEndpoint::new();
579        control
580            .connect(Box::new(TestEndpoint { is_ordered: false }))
581            .unwrap();
582        control
583            .connect(Box::new(TestEndpoint { is_ordered: true }))
584            .unwrap();
585    }
586}
net_backend/lib.rs

net_backend/
lib.rs
