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

net_backend/
lib.rs
