Merge pull request #1522 from microsoft/enhancement-xdp-refactor

[catpowder] Enhancement: XDP refactor and bug fixes

Author: kyleholohan

scripts/config/azure.yaml

@@ -9,6 +9,10 @@ raw_socket:
   xdp_interface_index: 0
   xdp_cohost_mode: false
   xdp_always_poke_tx: false
+  # If true, will send all packets out on the interface identified by xdp_vf_interface_index;
+  # otherwise, or if xdp_vf_interface_index is not set, will send packets out on the interface
+  # identified by xdp_interface_index.
+  xdp_always_send_on_vf: false
   # Enable the following for XDP cohosting mode, or override in environment:
   # xdp_tcp_ports: [80, 443]
   # xdp_udp_ports: [53]

scripts/config/default.yaml

@@ -9,6 +9,10 @@ raw_socket:
   xdp_interface_index: 0
   xdp_cohost_mode: false
   xdp_always_poke_tx: false
+  # If true, will send all packets out on the interface identified by xdp_vf_interface_index;
+  # otherwise, or if xdp_vf_interface_index is not set, will send packets out on the interface
+  # identified by xdp_interface_index.
+  xdp_always_send_on_vf: false
   # Enable the following for XDP cohosting mode, or override in environment:
   # xdp_tcp_ports: [80, 443]
   # xdp_udp_ports: [53]

src/rust/catpowder/win/cohosting.rs

@@ -0,0 +1,223 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+//======================================================================================================================
+// Imports
+//======================================================================================================================
+
+use std::rc::Rc;
+
+use windows::Win32::Networking::WinSock::{
+    closesocket, socket, WSACleanup, WSAIoctl, WSAStartup, AF_INET, INET_PORT_RANGE, INET_PORT_RESERVATION_INSTANCE,
+    INVALID_SOCKET, IN_ADDR, IPPROTO_TCP, IPPROTO_UDP, SIO_ACQUIRE_PORT_RESERVATION, SOCKET, SOCK_DGRAM, SOCK_STREAM,
+    WSADATA,
+};
+
+use crate::{
+    catnap::transport::error::expect_last_wsa_error,
+    catpowder::win::ring::RuleSet,
+    demikernel::config::Config,
+    inetstack::protocols::{layer4::ephemeral::EphemeralPorts, Protocol},
+    runtime::fail::Fail,
+};
+
+//======================================================================================================================
+// Structures
+//======================================================================================================================
+
+/// State to track port sharing state with the kernel for cohosting.
+pub struct PortSharingState {
+    /// The local IP address of the host, from the config.
+    local_ip: IN_ADDR,
+    /// All TCP ports to be redirected.
+    tcp_ports: Vec<u16>,
+    /// All UDP ports to be redirected.
+    udp_ports: Vec<u16>,
+    /// The Winsock socket used to reserve ephemeral ports with the kernel.
+    reserved_socket: SOCKET,
+    /// The list of reserved ephemeral ports.
+    reserved_ports: Vec<u16>,
+}
+
+/// The state of cohosting.
+pub enum CohostingMode {
+    /// No cohosting mode is enabled.
+    None,
+    /// Port sharing mode is enabled.
+    PortSharing(PortSharingState),
+}
+
+//======================================================================================================================
+// Implementations
+//======================================================================================================================
+
+impl CohostingMode {
+    /// Creates a new instance of `CohostingMode` based on the provided configuration.
+    pub fn new(config: &Config) -> Result<Self, Fail> {
+        if config.xdp_cohost_mode()? == false {
+            return Ok(CohostingMode::None);
+        }
+
+        let local_ip: IN_ADDR = IN_ADDR::from(config.local_ipv4_addr()?);
+
+        let (mut tcp_ports, mut udp_ports) = config.xdp_cohost_ports()?;
+
+        let reserved_protocol: Option<Protocol> = config.xdp_reserved_port_protocol()?;
+        let reserved_port_count: Option<u16> = config.xdp_reserved_port_count()?;
+
+        let (reserved_socket, reserved_ports): (SOCKET, Vec<u16>) =
+            if reserved_protocol.is_some() && reserved_port_count.is_some() {
+                let protocol: Protocol = reserved_protocol.unwrap();
+                let port_count: u16 = reserved_port_count.unwrap();
+
+                let mut data: WSADATA = WSADATA::default();
+                if unsafe { WSAStartup(0x202u16, &mut data as *mut WSADATA) } != 0 {
+                    return Err(expect_last_wsa_error());
+                }
+
+                trace!("reserving {} ports with protocol {:?}", port_count, protocol);
+
+                let (socket, ports) = reserve_port_blocks(port_count, protocol).or_else(|f: Fail| {
+                    let _ = unsafe { WSACleanup() };
+                    Err(f)
+                })?;
+
+                match protocol {
+                    Protocol::Tcp => tcp_ports.extend(ports.iter().cloned()),
+                    Protocol::Udp => udp_ports.extend(ports.iter().cloned()),
+                }
+
+                (socket, ports)
+            } else {
+                trace!("reserved port options not set; no ports reserved");
+                (INVALID_SOCKET, vec![])
+            };
+
+        trace!(
+            "XDP cohost mode enabled. TCP ports: {:?}, UDP ports: {:?}",
+            tcp_ports,
+            udp_ports
+        );
+
+        Ok(CohostingMode::PortSharing(PortSharingState {
+            local_ip,
+            tcp_ports,
+            udp_ports,
+            reserved_socket,
+            reserved_ports,
+        }))
+    }
+
+    pub fn create_ruleset(&self) -> Rc<RuleSet> {
+        match self {
+            CohostingMode::None => return RuleSet::new_redirect_all(),
+            CohostingMode::PortSharing(state) => {
+                RuleSet::new_cohost(state.local_ip, state.tcp_ports.as_slice(), state.udp_ports.as_slice())
+            },
+        }
+    }
+
+    pub fn ephemeral_ports(&self) -> EphemeralPorts {
+        match self {
+            CohostingMode::None => return EphemeralPorts::default(),
+            CohostingMode::PortSharing(state) => {
+                if state.reserved_ports.is_empty() {
+                    EphemeralPorts::default()
+                } else {
+                    EphemeralPorts::new(state.reserved_ports.as_slice()).unwrap()
+                }
+            },
+        }
+    }
+}
+
+fn reserve_port_blocks(port_count: u16, protocol: Protocol) -> Result<(SOCKET, Vec<u16>), Fail> {
+    const MAX_HALVINGS: usize = 5;
+    let mut ports: Vec<u16> = Vec::with_capacity(port_count as usize);
+
+    let mut reservation_len: u16 = port_count;
+    let mut halvings: usize = 0;
+
+    let (sock_type, protocol) = match protocol {
+        Protocol::Tcp => (SOCK_STREAM, IPPROTO_TCP.0),
+        Protocol::Udp => (SOCK_DGRAM, IPPROTO_UDP.0),
+    };
+
+    let s: SOCKET = unsafe { socket(AF_INET.0.into(), sock_type, protocol) };
+    if s == INVALID_SOCKET {
+        return Err(expect_last_wsa_error());
+    }
+
+    while ports.len() < port_count as usize {
+        trace!("reserve_port_blocks(): trying reservation length: {}", reservation_len);
+        match reserve_ports(reservation_len, s) {
+            Ok((start, count, _)) if count > 0 => {
+                let end: u16 = start + (count - 1);
+                trace!("reserve_port_blocks(): reserved ports: {}-{}", start, end);
+                ports.extend(start..=end);
+            },
+            Ok(_) => {
+                panic!("reserve_port_blocks(): reserved zero ports");
+            },
+            Err(e) => {
+                halvings += 1;
+                if halvings >= MAX_HALVINGS || reservation_len == 1 {
+                    error!("reserve_port_blocks(): failed to reserve ports; giving up: {:?}", e);
+                    let _ = unsafe { closesocket(s) };
+                    return Err(e);
+                } else {
+                    trace!(
+                        "reserve_port_blocks(): failed to reserve ports; halving reservation size: {:?}",
+                        e
+                    );
+                    reservation_len /= 2;
+                }
+            },
+        }
+    }
+
+    Ok((s, ports))
+}
+
+fn reserve_ports(port_count: u16, s: SOCKET) -> Result<(u16, u16, u64), Fail> {
+    let port_range: INET_PORT_RANGE = INET_PORT_RANGE {
+        StartPort: 0,
+        NumberOfPorts: port_count,
+    };
+
+    let mut reservation: INET_PORT_RESERVATION_INSTANCE = INET_PORT_RESERVATION_INSTANCE::default();
+    let mut bytes_out: u32 = 0;
+
+    let result: i32 = unsafe {
+        WSAIoctl(
+            s,
+            SIO_ACQUIRE_PORT_RESERVATION,
+            Some(&port_range as *const INET_PORT_RANGE as *mut libc::c_void),
+            std::mem::size_of::<INET_PORT_RANGE>() as u32,
+            Some(&mut reservation as *mut INET_PORT_RESERVATION_INSTANCE as *mut libc::c_void),
+            std::mem::size_of::<INET_PORT_RESERVATION_INSTANCE>() as u32,
+            &mut bytes_out,
+            None,
+            None,
+        )
+    };
+
+    if result != 0 {
+        return Err(expect_last_wsa_error());
+    }
+
+    Ok((
+        u16::from_be(reservation.Reservation.StartPort),
+        reservation.Reservation.NumberOfPorts,
+        reservation.Token.Token,
+    ))
+}
+
+impl Drop for PortSharingState {
+    fn drop(&mut self) {
+        if self.reserved_socket != INVALID_SOCKET {
+            let _ = unsafe { closesocket(self.reserved_socket) };
+            let _ = unsafe { WSACleanup() };
+        }
+    }
+}

src/rust/catpowder/win/interface.rs

@@ -0,0 +1,137 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+//======================================================================================================================
+// Imports
+//======================================================================================================================
+
+use std::{num::NonZeroU32, rc::Rc};
+
+use crate::{
+    catpowder::win::{
+        api::XdpApi,
+        ring::{RuleSet, RxRing, TxRing},
+        socket::XdpSocket,
+    },
+    demikernel::config::Config,
+    runtime::fail::Fail,
+};
+
+//======================================================================================================================
+// Structures
+//======================================================================================================================
+
+/// State for the XDP interface.
+pub struct Interface {
+    /// Currently only one TX is created for all sends on the interface.
+    pub tx_ring: TxRing,
+    /// RX rings for the interface, one per RSS queue.
+    pub rx_rings: Vec<RxRing>,
+    /// Sockets for the interface, one for each *Ring member above, with a description of the socket.
+    pub sockets: Vec<(String, XdpSocket)>,
+}
+
+//======================================================================================================================
+// Implementations
+//======================================================================================================================
+
+impl Interface {
+    /// Creates a new interface for the given configuration. The interface creates [queue_count] RX
+    /// rings.
+    pub fn new(
+        api: &mut XdpApi,
+        ifindex: u32,
+        queue_count: NonZeroU32,
+        ruleset: Rc<RuleSet>,
+        config: &Config,
+    ) -> Result<Self, Fail> {
+        let (tx_buffer_count, tx_ring_size) = config.tx_buffer_config()?;
+        let (rx_buffer_count, rx_ring_size) = config.rx_buffer_config()?;
+        let mtu: u16 = config.mtu()?;
+
+        let (tx_ring_size, tx_buffer_count): (NonZeroU32, NonZeroU32) =
+            validate_ring_config(tx_ring_size, tx_buffer_count, "tx")?;
+        let (rx_ring_size, rx_buffer_count): (NonZeroU32, NonZeroU32) =
+            validate_ring_config(rx_ring_size, rx_buffer_count, "rx")?;
+
+        let always_poke: bool = config.xdp_always_poke_tx()?;
+
+        let mut rx_rings: Vec<RxRing> = Vec::with_capacity(queue_count.get() as usize);
+        let mut sockets: Vec<(String, XdpSocket)> = Vec::new();
+
+        let tx_ring: TxRing = TxRing::new(
+            api,
+            tx_ring_size.get(),
+            tx_buffer_count.get(),
+            mtu,
+            ifindex,
+            0,
+            always_poke,
+        )?;
+        sockets.push((format!("if {} tx 0", ifindex), tx_ring.socket().clone()));
+
+        for queueid in 0..queue_count.get() {
+            let mut ring: RxRing = RxRing::new(
+                api,
+                rx_ring_size.get(),
+                rx_buffer_count.get(),
+                mtu,
+                ifindex,
+                queueid,
+                ruleset.clone(),
+            )?;
+            ring.provide_buffers();
+            sockets.push((format!("if {} rx {}", ifindex, queueid), ring.socket().clone()));
+            rx_rings.push(ring);
+        }
+
+        trace!("Created {} RX rings on interface {}", rx_rings.len(), ifindex);
+
+        Ok(Self {
+            tx_ring,
+            rx_rings,
+            sockets,
+        })
+    }
+
+    pub fn return_tx_buffers(&mut self) {
+        self.tx_ring.return_buffers();
+    }
+
+    pub fn provide_rx_buffers(&mut self) {
+        for ring in self.rx_rings.iter_mut() {
+            ring.provide_buffers();
+        }
+    }
+}
+
+//======================================================================================================================
+// Functions
+//======================================================================================================================
+
+/// Validates the ring size and buffer count for the given configuration.
+fn validate_ring_config(ring_size: u32, buf_count: u32, config: &str) -> Result<(NonZeroU32, NonZeroU32), Fail> {
+    let ring_size: NonZeroU32 = NonZeroU32::try_from(ring_size)
+        .map_err(Fail::from)
+        .and_then(|v: NonZeroU32| {
+            if !v.is_power_of_two() {
+                let cause: String = format!("{}_ring_size must be a power of two: {}", config, v.get());
+                Err(Fail::new(libc::EINVAL, &cause))
+            } else {
+                Ok(v)
+            }
+        })?;
+
+    let buf_count: NonZeroU32 = if buf_count < ring_size.get() {
+        let cause: String = format!(
+            "{}_buffer_count must be greater than or equal to {}_ring_size",
+            config, config
+        );
+        return Err(Fail::new(libc::EINVAL, &cause));
+    } else {
+        // Safety: since buffer_count >= ring_size, we can safely create a NonZeroU32.
+        unsafe { NonZeroU32::new_unchecked(buf_count) }
+    };
+
+    Ok((ring_size, buf_count))
+}

src/rust/catpowder/win/mod.rs

@@ -6,7 +6,11 @@
 //======================================================================================================================
 
 mod api;
+mod cohosting;
+mod interface;
+mod observability;
 mod ring;
+mod rss;
 mod socket;
 
 //======================================================================================================================