diff --git a/Cargo.lock b/Cargo.lock index 2bb9687..adf4016 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1834,7 +1834,7 @@ dependencies = [ [[package]] name = "p4rs" version = "0.1.0" -source = "git+https://github.com/oxidecomputer/p4?branch=main#c8b190f46901d4c1b8eef1ca35be243b18653471" +source = "git+https://github.com/oxidecomputer/p4?branch=main#89b725ca3d534089675ef59f1606882218b6a3c8" dependencies = [ "bitvec", "num", @@ -2774,6 +2774,7 @@ dependencies = [ "p4rs", "serde", "serde_json", + "tempfile", "tokio", ] diff --git a/README.md b/README.md index 763ce5a..49d531e 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # SoftNpu SoftNpu is a software network processing unit. The main thing SoftNpu -implementations do is provide an I/O harness and management interface for +implementations do is provide an I/O harness and management interface for [x4c](https://github.com/oxidecomputer/p4) compiled P4 programs. I/O harnesses provide a way to get packets in and out of P4 pipeline programs and management interfaces provide a means to manage P4 @@ -12,7 +12,7 @@ This repository contains the following. - Management message definitions for SoftNpu implementations. - A standalone SoftNpu implementation. -Two SoftNpu implementations currently exist. +Two SoftNpu implementations currently exist. 1. The standalone implementation in this repository 2. A propolis ASIC emulator. diff --git a/client/src/bin/softnpu.rs b/client/src/bin/softnpu.rs index 7e2a7a9..4ce7ce2 100644 --- a/client/src/bin/softnpu.rs +++ b/client/src/bin/softnpu.rs @@ -391,13 +391,16 @@ async fn run(log: Logger) -> Result<()> { Err(_) => continue, }; - softnpu::handle_management_message( + if let Err(e) = softnpu::handle_management_message( msg, &pipe, &uds, client.as_ref(), config.ports.len(), ) - .await; + .await + { + error!(log, "management request failed"; "error" => %e); + } } } diff --git a/lib/Cargo.toml b/lib/Cargo.toml index 88ab764..87eee01 100644 --- a/lib/Cargo.toml +++ b/lib/Cargo.toml @@ -8,3 +8,7 @@ p4rs.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true + +[dev-dependencies] +tempfile.workspace = true +tokio = { workspace = true, features = ["rt", "macros"] } diff --git a/lib/src/lib.rs b/lib/src/lib.rs index ffb68ac..e869ff5 100644 --- a/lib/src/lib.rs +++ b/lib/src/lib.rs @@ -5,6 +5,7 @@ use p4rs::{Pipeline, TableEntry}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; +use std::io; use std::path::Path; use std::sync::Mutex; use tokio::net::UnixDatagram; @@ -19,6 +20,7 @@ pub enum ManagementRequest { TableAdd(TableAdd), TableRemove(TableRemove), DumpRequest, + TableClear(TableClear), } #[derive(Debug, Serialize, Deserialize)] @@ -41,13 +43,18 @@ pub struct TableRemove { pub keyset_data: Vec, } +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct TableClear { + pub table: String, +} + pub async fn handle_management_message( msg: ManagementRequest, pipeline: &Mutex>, uds: &UnixDatagram, uds_dst: &Path, radix: usize, -) { +) -> io::Result<()> { match msg { ManagementRequest::TableAdd(tm) => { let mut pl = pipeline.lock().unwrap(); @@ -63,10 +70,18 @@ pub async fn handle_management_message( let mut pl = pipeline.lock().unwrap(); pl.remove_table_entry(&tm.table, &tm.keyset_data); } + ManagementRequest::TableClear(tc) => { + let mut pl = pipeline.lock().unwrap(); + if !pl.clear_table_entries(&tc.table) { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!("unknown table: {}", tc.table), + )); + } + } ManagementRequest::RadixRequest => { let response = ManagementResponse::RadixResponse(radix as u16); - let buf = serde_json::to_vec(&response).unwrap(); - uds.send_to(&buf, uds_dst).await.unwrap(); + send_response(response, uds, uds_dst).await?; } ManagementRequest::DumpRequest => { let result = { @@ -82,8 +97,148 @@ pub async fn handle_management_message( }; let response = ManagementResponse::DumpResponse(result); - let buf = serde_json::to_vec(&response).unwrap(); - uds.send_to(&buf, uds_dst).await.unwrap(); + send_response(response, uds, uds_dst).await?; + } + } + Ok(()) +} + +async fn send_response( + response: ManagementResponse, + uds: &UnixDatagram, + uds_dst: &Path, +) -> io::Result<()> { + let buf = serde_json::to_vec(&response).map_err(io::Error::other)?; + uds.send_to(&buf, uds_dst).await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + struct TestPipeline { + tables: BTreeMap>, + } + + impl Default for TestPipeline { + fn default() -> Self { + let allow_source = |keyset_data: Vec| TableEntry { + action_id: "allow_source".to_string(), + keyset_data, + parameter_data: Vec::new(), + }; + // Missing ids in this map are considered unknown tables. + let tables = BTreeMap::from([ + // IPv4 source filter key: 192.0.2.1/32, dst 224.1.2.3 (LE). + ( + "populated".to_string(), + vec![allow_source(vec![192, 0, 2, 1, 32, 3, 2, 1, 224])], + ), + ("empty".to_string(), Vec::new()), + ( + "other".to_string(), + vec![allow_source(vec![192, 0, 2, 2, 32, 4, 2, 1, 224])], + ), + ]); + Self { tables } + } + } + + impl Pipeline for TestPipeline { + fn process_packet<'a>( + &mut self, + _port: u16, + _packet: &mut p4rs::packet_in<'a>, + ) -> Vec<(p4rs::packet_out<'a>, u16)> { + unreachable!() + } + + fn add_table_entry( + &mut self, + _table: &str, + _action: &str, + _keys: &[u8], + _parameters: &[u8], + _priority: u32, + ) { + unreachable!() + } + + fn remove_table_entry(&mut self, _table: &str, _keys: &[u8]) { + unreachable!() + } + + fn clear_table_entries(&mut self, table: &str) -> bool { + match self.tables.get_mut(table) { + Some(entries) => { + entries.clear(); + true + } + None => false, + } + } + + fn get_table_entries(&self, table: &str) -> Option> { + self.tables.get(table).map(|entries| { + entries + .iter() + .map(|entry| TableEntry { + action_id: entry.action_id.clone(), + keyset_data: entry.keyset_data.clone(), + parameter_data: entry.parameter_data.clone(), + }) + .collect() + }) + } + + fn get_table_ids(&self) -> Vec<&str> { + unreachable!() + } + } + + #[tokio::test] + async fn table_clear_empties_only_requested_table() { + let dir = tempfile::tempdir().unwrap(); + let server = UnixDatagram::bind(dir.path().join("server")).unwrap(); + let client_path = dir.path().join("client"); + let pipeline: Mutex> = + Mutex::new(Box::new(TestPipeline::default())); + let table_lens = |pipeline: &Mutex>| { + let pl = pipeline.lock().unwrap(); + ["populated", "empty", "other", "missing"] + .map(|table| pl.get_table_entries(table).as_ref().map(Vec::len)) + }; + assert_eq!(table_lens(&pipeline), [Some(1), Some(0), Some(1), None]); + + for (table, expected) in [ + ("populated", [Some(0), Some(0), Some(1), None]), + ("populated", [Some(0), Some(0), Some(1), None]), + ("empty", [Some(0), Some(0), Some(1), None]), + ("missing", [Some(0), Some(0), Some(1), None]), + ("other", [Some(0), Some(0), Some(0), None]), + ] { + let request = ManagementRequest::TableClear(TableClear { + table: table.to_string(), + }); + let result = handle_management_message( + request, + &pipeline, + &server, + &client_path, + 4, + ) + .await; + if table == "missing" { + assert_eq!(result.unwrap_err().kind(), io::ErrorKind::NotFound); + } else { + result.unwrap(); + } + assert_eq!( + table_lens(&pipeline), + expected, + "after clearing {table}" + ); } } }