[PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

[PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

[Harry van Haaren]

This patch is NOT to be considered for merge, it is a demo
of the Rust APIs for Ethdev. There is no actual implementation
of the APIs against the DPDK C functions, this is Rust API only.

To test/run the code (and uncomment things to see errors)
just apply this patch, cd "rust_api_example" and run
$ cargo run

This will compile the API, and spawn 2x threads to poll on
two Rxq instances. The comments in the code explain how the
"Send" and "Sync" attributes are captured per instances of a
struct (e.g. how RxqHandle -> Rxq restricts thread movement).

Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 rust_api_example/Cargo.toml  |   6 ++
 rust_api_example/src/main.rs | 189 +++++++++++++++++++++++++++++++++++
 2 files changed, 195 insertions(+)
 create mode 100644 rust_api_example/Cargo.toml
 create mode 100644 rust_api_example/src/main.rs

diff --git a/rust_api_example/Cargo.toml b/rust_api_example/Cargo.toml
new file mode 100644
index 0000000000..0137826340
--- /dev/null
+++ b/rust_api_example/Cargo.toml
@@ -0,0 +1,6 @@
+[package]
+name = "rust_api_example"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
diff --git a/rust_api_example/src/main.rs b/rust_api_example/src/main.rs
new file mode 100644
index 0000000000..8d0de50c30
--- /dev/null
+++ b/rust_api_example/src/main.rs
@@ -0,0 +1,189 @@
+// Outline for safe DPDK API bindings
+//  - None of the APIs are actually implemented, this is API design only
+//  - This demo runs 2x threads on 2x Rxqs, and cannot accidentally poll incorrectly
+
+pub mod dpdk {
+    pub mod eth {
+        use super::Mempool;
+
+        #[derive(Debug)]
+        pub struct TxqHandle {/* todo: but same as Rxq */}
+
+        // Handle allows moving between threads, its not polling!
+        #[derive(Debug)]
+        pub struct RxqHandle {
+            port: u16,
+            queue: u16,
+        }
+
+        impl RxqHandle {
+            pub(crate) fn new(port: u16, queue: u16) -> Self {
+                RxqHandle { port, queue }
+            }
+
+            // This function is the key to the API design: it ensures the rx_burst()
+            // function is only available via the Rxq struct, after enable_polling() has been called.
+            // It "consumes" (takes "self" as a parameter, not a '&' reference!) which essentially
+            // destroys/invalidates the handle from the Application level code.
+
+            // It returns an Rxq instance, which has the PhantomData to encode the threading requirements,
+            // and the Rxq has the rx_burst() function: this allows the application to recieve packets.
+            pub fn enable_polling(self) -> Rxq {
+                Rxq {
+                    handle: self,
+                    _phantom: std::marker::PhantomData,
+                }
+            }
+        }
+
+        #[derive(Debug)]
+        pub struct Rxq {
+            handle: RxqHandle,
+            // This "PhantomData" tells the rust compiler to Pretend the Rc<()> is in this struct
+            // but in practice it is a Zero-Sized-Type, so takes up no space. It is a compile-time
+            // language technique to ensure the struct is not moved between threads. This encodes
+            // the API requirement "don't poll from multiple threads without synchronisation (e.g. Mutex)"
+            _phantom: std::marker::PhantomData<std::rc::Rc<()>>,
+        }
+
+        impl Rxq {
+            // TODO: datapath Error types should be lightweight, not String. Here we return ().
+            pub fn rx_burst(&mut self, _mbufs: &mut [u8]) -> Result<usize, ()> {
+                // TODO: Design the Mbuf struct wrapper, and how to best return a batch
+                //  e.g.: investigate "ArrayVec" crate for safe & fixed sized, stack allocated arrays
+                //
+                // There is work to do here, but I want to communicate the general DPDK/EAL/Eth/Rxq concepts
+                // now, this part is not done yet: it is likely the hardest/most performance critical.
+                //
+                // call rte_eth_rx_burst() here
+                println!(
+                    "[thread: {:?}] rx_burst: port {} queue {}",
+                    std::thread::current().id(),
+                    self.handle.port,
+                    self.handle.queue
+                );
+                Ok(0)
+            }
+        }
+
+        #[derive(Debug)]
+        pub struct Port {
+            id: u16,
+            rxqs: Vec<RxqHandle>,
+            txqs: Vec<TxqHandle>,
+        }
+
+        impl Port {
+            // pub(crate) here ensures outside this crate users cannot call this function
+            pub(crate) fn from_u16(id: u16) -> Self {
+                Port {
+                    id,
+                    rxqs: Vec::new(),
+                    txqs: Vec::new(),
+                }
+            }
+
+            pub fn rxqs(&mut self, rxq_count: u16, _mempool: Mempool) -> Result<(), String> {
+                for q in 0..rxq_count {
+                    // call rte_eth_rx_queue_setup() here
+                    self.rxqs.push(RxqHandle::new(self.id, q));
+                }
+                Ok(())
+            }
+
+            pub fn start(&mut self) -> (Vec<RxqHandle>, Vec<TxqHandle>) {
+                // call rte_eth_dev_start() here, then give ownership of Rxq/Txq to app
+                (
+                    std::mem::take(&mut self.rxqs),
+                    std::mem::take(&mut self.txqs),
+                )
+            }
+        }
+    }
+
+    #[derive(Debug, Clone)]
+    // Mempool is a long-life object, which many other DPDK things refer to (e.g. rxq config)
+    // Having a Rust lifetime attached to it (while technically correct) would complicate the
+    // code a LOT, and for little value. This is a tradeoff - happy to discuss more if we want.
+    // The choice here is to derive "Clone", allowing handing over multiple instances of the
+    // same Mempool, similar to how Arc<Mempool> would work, but without the reference counting.
+    pub struct Mempool {}
+
+    impl Mempool {
+        pub fn new(_size: usize) -> Self {
+            Self {}
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct Eal {
+        eth_ports: Option<Vec<eth::Port>>,
+    }
+
+    impl Eal {
+        //  allow init once,
+        pub fn init() -> Result<Self, String> {
+            // EAL init() will do PCI probe and VDev enumeration will find/create eth ports.
+            // This code should loop over the ports, and build up Rust structs representing them
+            let eth_port = vec![eth::Port::from_u16(0)];
+            Ok(Eal {
+                eth_ports: Some(eth_port),
+            })
+        }
+
+        // API to get eth ports, taking ownership. It can be called once.
+        // The return will be None for future calls
+        pub fn take_eth_ports(&mut self) -> Option<Vec<eth::Port>> {
+            self.eth_ports.take()
+        }
+    }
+
+    impl Drop for Eal {
+        fn drop(&mut self) {
+            // todo: rte_eal_cleanup()
+        }
+    }
+} // DPDK mod
+
+fn main() {
+    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
+    let rx_mempool = dpdk::Mempool::new(4096);
+
+    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
+    let mut p = ports.pop().unwrap();
+
+    p.rxqs(2, rx_mempool).expect("rxqs setup ok");
+    println!("{:?}", p);
+
+    let (mut rxqs, _txqs) = p.start();
+    println!("rxqs: {:?}", rxqs);
+
+    let rxq1 = rxqs.pop().unwrap();
+    let rxq2 = rxqs.pop().unwrap();
+
+    // spawn a new thread to use rxq1. This demonstrates that the RxqHandle
+    // type can move between threads - it is not tied to the thread that created it.
+    std::thread::spawn(move || {
+        // Uncomment this: it fails to compile!
+        //   - Rxq2 would be used by this newly-spawned thread
+        //     -- specifically the variable was "moved" into this thread
+        //   - it is also used below (by the main thread)
+        // "value used after move" is the error, on the below code
+        // let mut rxq = rxq2.enable_polling();
+
+        // see docs on enable_polling above to understand how the enable_polling()
+        // function helps to achieve the thread-safety-at-compile-time goal.
+        let mut rxq = rxq1.enable_polling();
+        loop {
+            let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+            std::thread::sleep(std::time::Duration::from_millis(1000));
+        }
+    });
+
+    // main thread polling rxq2
+    let mut rxq = rxq2.enable_polling();
+    loop {
+        let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+        std::thread::sleep(std::time::Duration::from_millis(1000));
+    }
+}
-- 
2.34.1

Re: [PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

[Etelson, Gregory]

Hello Harry,

Thank you for sharing the API.
Please check out my comments below.

Regards,
Gregory

On Thu, 17 Apr 2025, Harry van Haaren wrote:

> External email: Use caution opening links or attachments
>
>
> This patch is NOT to be considered for merge, it is a demo
> of the Rust APIs for Ethdev. There is no actual implementation
> of the APIs against the DPDK C functions, this is Rust API only.
>
> To test/run the code (and uncomment things to see errors)
> just apply this patch, cd "rust_api_example" and run
> $ cargo run
>
> This will compile the API, and spawn 2x threads to poll on
> two Rxq instances. The comments in the code explain how the
> "Send" and "Sync" attributes are captured per instances of a
> struct (e.g. how RxqHandle -> Rxq restricts thread movement).
>
> Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
> ---
> rust_api_example/Cargo.toml  |   6 ++
> rust_api_example/src/main.rs | 189 +++++++++++++++++++++++++++++++++++
> 2 files changed, 195 insertions(+)
> create mode 100644 rust_api_example/Cargo.toml
> create mode 100644 rust_api_example/src/main.rs
>
> diff --git a/rust_api_example/Cargo.toml b/rust_api_example/Cargo.toml
> new file mode 100644
> index 0000000000..0137826340
> --- /dev/null
> +++ b/rust_api_example/Cargo.toml
> @@ -0,0 +1,6 @@
> +[package]
> +name = "rust_api_example"
> +version = "0.1.0"
> +edition = "2021"
> +
> +[dependencies]
> diff --git a/rust_api_example/src/main.rs b/rust_api_example/src/main.rs
> new file mode 100644
> index 0000000000..8d0de50c30
> --- /dev/null
> +++ b/rust_api_example/src/main.rs
> @@ -0,0 +1,189 @@
> +// Outline for safe DPDK API bindings
> +//  - None of the APIs are actually implemented, this is API design only
> +//  - This demo runs 2x threads on 2x Rxqs, and cannot accidentally poll incorrectly
> +
> +pub mod dpdk {
> +    pub mod eth {
> +        use super::Mempool;
> +
> +        #[derive(Debug)]
> +        pub struct TxqHandle {/* todo: but same as Rxq */}
> +
> +        // Handle allows moving between threads, its not polling!
> +        #[derive(Debug)]
> +        pub struct RxqHandle {
> +            port: u16,
> +            queue: u16,
> +        }
> +
> +        impl RxqHandle {
> +            pub(crate) fn new(port: u16, queue: u16) -> Self {
> +                RxqHandle { port, queue }
> +            }
> +
> +            // This function is the key to the API design: it ensures the rx_burst()
> +            // function is only available via the Rxq struct, after enable_polling() has been called.
> +            // It "consumes" (takes "self" as a parameter, not a '&' reference!) which essentially
> +            // destroys/invalidates the handle from the Application level code.
> +
> +            // It returns an Rxq instance, which has the PhantomData to encode the threading requirements,
> +            // and the Rxq has the rx_burst() function: this allows the application to recieve packets.
> +            pub fn enable_polling(self) -> Rxq {
> +                Rxq {
> +                    handle: self,
> +                    _phantom: std::marker::PhantomData,
> +                }
> +            }
> +        }
> +
> +        #[derive(Debug)]
> +        pub struct Rxq {
> +            handle: RxqHandle,
> +            // This "PhantomData" tells the rust compiler to Pretend the Rc<()> is in this struct
> +            // but in practice it is a Zero-Sized-Type, so takes up no space. It is a compile-time
> +            // language technique to ensure the struct is not moved between threads. This encodes
> +            // the API requirement "don't poll from multiple threads without synchronisation (e.g. Mutex)"
> +            _phantom: std::marker::PhantomData<std::rc::Rc<()>>,
> +        }
> +
> +        impl Rxq {
> +            // TODO: datapath Error types should be lightweight, not String. Here we return ().
> +            pub fn rx_burst(&mut self, _mbufs: &mut [u8]) -> Result<usize, ()> {
> +                // TODO: Design the Mbuf struct wrapper, and how to best return a batch
> +                //  e.g.: investigate "ArrayVec" crate for safe & fixed sized, stack allocated arrays
> +                //
> +                // There is work to do here, but I want to communicate the general DPDK/EAL/Eth/Rxq concepts
> +                // now, this part is not done yet: it is likely the hardest/most performance critical.
> +                //
> +                // call rte_eth_rx_burst() here
> +                println!(
> +                    "[thread: {:?}] rx_burst: port {} queue {}",
> +                    std::thread::current().id(),
> +                    self.handle.port,
> +                    self.handle.queue
> +                );
> +                Ok(0)
> +            }
> +        }
> +
> +        #[derive(Debug)]
> +        pub struct Port {
> +            id: u16,
> +            rxqs: Vec<RxqHandle>,
> +            txqs: Vec<TxqHandle>,
> +        }
> +
> +        impl Port {
> +            // pub(crate) here ensures outside this crate users cannot call this function
> +            pub(crate) fn from_u16(id: u16) -> Self {
> +                Port {
> +                    id,
> +                    rxqs: Vec::new(),
> +                    txqs: Vec::new(),
> +                }
> +            }
> +
> +            pub fn rxqs(&mut self, rxq_count: u16, _mempool: Mempool) -> Result<(), String> {
> +                for q in 0..rxq_count {
> +                    // call rte_eth_rx_queue_setup() here
> +                    self.rxqs.push(RxqHandle::new(self.id, q));
> +                }
> +                Ok(())
> +            }
> +
> +            pub fn start(&mut self) -> (Vec<RxqHandle>, Vec<TxqHandle>) {
> +                // call rte_eth_dev_start() here, then give ownership of Rxq/Txq to app

After a call to Port::start, Rx and Tx queues are detached from it's port.
With that model how rte_eth_dev_stop() and subsequent rte_eth_dev_start()
DPDK calls can be implemented ?

> +                (
> +                    std::mem::take(&mut self.rxqs),
> +                    std::mem::take(&mut self.txqs),
> +                )
> +            }
> +        }
> +    }
> +
> +    #[derive(Debug, Clone)]
> +    // Mempool is a long-life object, which many other DPDK things refer to (e.g. rxq config)
> +    // Having a Rust lifetime attached to it (while technically correct) would complicate the
> +    // code a LOT, and for little value. This is a tradeoff - happy to discuss more if we want.
> +    // The choice here is to derive "Clone", allowing handing over multiple instances of the
> +    // same Mempool, similar to how Arc<Mempool> would work, but without the reference counting.
> +    pub struct Mempool {}
> +
> +    impl Mempool {
> +        pub fn new(_size: usize) -> Self {
> +            Self {}
> +        }
> +    }
> +
> +    #[derive(Debug)]
> +    pub struct Eal {
> +        eth_ports: Option<Vec<eth::Port>>,
> +    }
> +
> +    impl Eal {
> +        //  allow init once,
> +        pub fn init() -> Result<Self, String> {
> +            // EAL init() will do PCI probe and VDev enumeration will find/create eth ports.
> +            // This code should loop over the ports, and build up Rust structs representing them
> +            let eth_port = vec![eth::Port::from_u16(0)];
> +            Ok(Eal {
> +                eth_ports: Some(eth_port),
> +            })
> +        }
> +
> +        // API to get eth ports, taking ownership. It can be called once.
> +        // The return will be None for future calls
> +        pub fn take_eth_ports(&mut self) -> Option<Vec<eth::Port>> {
> +            self.eth_ports.take()
> +        }
> +    }
> +
> +    impl Drop for Eal {
> +        fn drop(&mut self) {
> +            // todo: rte_eal_cleanup()
> +        }
> +    }
> +} // DPDK mod
> +
> +fn main() {
> +    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
> +    let rx_mempool = dpdk::Mempool::new(4096);
> +
> +    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");

Eal::take_eth_ports() resets EAL ports.
A call to rte_dev_probe() will ether fail, because Eal::eth_ports is None 
or create another port-0, depending on implementation.

> +    let mut p = ports.pop().unwrap();
> +
> +    p.rxqs(2, rx_mempool).expect("rxqs setup ok");
> +    println!("{:?}", p);
> +
> +    let (mut rxqs, _txqs) = p.start();
> +    println!("rxqs: {:?}", rxqs);
> +
> +    let rxq1 = rxqs.pop().unwrap();
> +    let rxq2 = rxqs.pop().unwrap();
> +
> +    // spawn a new thread to use rxq1. This demonstrates that the RxqHandle
> +    // type can move between threads - it is not tied to the thread that created it.
> +    std::thread::spawn(move || {
> +        // Uncomment this: it fails to compile!
> +        //   - Rxq2 would be used by this newly-spawned thread
> +        //     -- specifically the variable was "moved" into this thread
> +        //   - it is also used below (by the main thread)
> +        // "value used after move" is the error, on the below code
> +        // let mut rxq = rxq2.enable_polling();
> +
> +        // see docs on enable_polling above to understand how the enable_polling()
> +        // function helps to achieve the thread-safety-at-compile-time goal.
> +        let mut rxq = rxq1.enable_polling();
> +        loop {
> +            let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
> +            std::thread::sleep(std::time::Duration::from_millis(1000));
> +        }
> +    });
> +
> +    // main thread polling rxq2
> +    let mut rxq = rxq2.enable_polling();
> +    loop {
> +        let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
> +        std::thread::sleep(std::time::Duration::from_millis(1000));
> +    }
> +}
> --
> 2.34.1
>
>

Re: [PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

[Van Haaren, Harry]

> From: Etelson, Gregory
> Sent: Thursday, April 17, 2025 7:58 PM
> To: Van Haaren, Harry
> Cc: dev@dpdk.org; getelson@nvidia.com; Richardson, Bruce; owen.hilyard@unh.edu
> Subject: Re: [PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq
> 
> Hello Harry,
> 
> Thank you for sharing the API.
> Please check out my comments below.

Thanks for reading & discussion!

<snip>

> > +
> > +            pub fn start(&mut self) -> (Vec<RxqHandle>, Vec<TxqHandle>) {
> > +                // call rte_eth_dev_start() here, then give ownership of Rxq/Txq to app
> 
> After a call to Port::start, Rx and Tx queues are detached from it's port.
> With that model how rte_eth_dev_stop() and subsequent rte_eth_dev_start()
> DPDK calls can be implemented ?

Correct, the RxqHandle and TxqHandle don't have a "back reference" to the port.
There are a number of ways to ensure eth_dev_stop() cannot be called without the
Rxq/Txqs being "returned" to the Port instance first.

Eg: Use an Arc<T>. The port instance "owns" the Arc<T>, which means it is going to keep
   the Arc alive. Now give each Rxq/Txq a clone of this Arc. When the Drop impl of the
   Rxq/Txq runs, it will decrement the Arc. So just letting the Rxq/Txq go out of scope
   will be enough to have the Port understand that handle is now gone.
   
   The port itself can use Arc::into_inner function[1], which returns Option<T>. If the
   Some(T) is returned, then all instances of RxqHandle/TxqHandle have been dropped,
   meaning it is safe to eth_dev_stop(), as it is impossible to poll RXQs if there's no Rxq :)
   [1] https://doc.rust-lang.org/std/sync/struct.Arc.html#method.into_inner

// Pseudo-code here:
Dpdk::Eth::Port::stop(&mut self) -> Result<(), Error> {
    let handles_dropped = self.handle_arc.into_inner(); // returns "T" if its the only reference to the Arc
    if handles_dropped.is_none() {
        return Err("an Rxq or Txq handle remains alive, cannot safely stop this port");
    }
}

There's probably a few others, but that's "idiomatic Rust" solution.
We'd have to pass the Arc from the RxqHandle into the Rxq instance itself too,
but that's fine.

<snip>

> > +fn main() {
> > +    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
> > +    let rx_mempool = dpdk::Mempool::new(4096);
> > +
> > +    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
> 
> Eal::take_eth_ports() resets EAL ports.

I don't think it "resets" here. The "take eth ports" removes the Port instances from
the dpdk::Eal struct, but there's no "reset" behaviour.

> A call to rte_dev_probe() will ether fail, because Eal::eth_ports is None
> or create another port-0, depending on implementation.

I don't see how or why rte_dev_probe() would be called. The idea is not to allow Rust
apps call DPDK C APIs "when they want". The safe Rust API provides the required abstraction.
So its not possible to have another call to rte_dev_probe(), after the 1st time under eal_init().

Similar topic: Hotplug. I have experience with designing C APIs around hotplug
use-cases (Music/DJ software, from before my DPDK/networking days!). I think DPDK has
an interesting "push hotplug" approach (aka, App makes a function call to "request" the device).
Then on successful return, we can call rte_eth_dev_get_port_by_name() to get the u16 port_id,
and build the Port instance from that. Outline API:

enum EalHotplugDev {
    EthDev(Dpdk::Eth::Port), // enums can have contents in Rust :)
    CryptoDev(Dpdk::Crypto),
    // Etc
}

Eal::hotplug_add(bus: String, dev: String, args: String) -> Result<EalHotplugDev, Error> {
    // TODO: call rte_eal_hotplug_add()
    // TODO: identify how to know if its an Eth, Crypto, Dma, or other dev type?
    match (dev_type) {
        "eth" => {
	    let port_id = rte_eth_dev_get_port_by_name(dev);
	    EalHotplugDev::EthDev( Dpdk::Eth::Port::new(port_id) )
        }
    }
}

Applications could then do:
  let Ok(dev) = eal.hotplug_add("pci", "02:00.0", "dev_option=true") else {
      // failed to hotplug, log error?
      return;
  }
  match (dev) {
      EthDev => {
          // handle the dev here, e.g. configure & spawn thread to poll Rxq like before.
      }
  }

I like having an outline of difficult to "bolt on" features (hotplug is typically hard to add later..)
but I recommend we focus on getting core APIs and such running before more detail/time/implementation here.


Regards, -Harry

Re: [PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

[Gregory Etelson]

[-- Attachment #1: Type: text/plain, Size: 6018 bytes --]

Hello Harry,

I implemented a working echo server with your API.
The code is here: https://github.com/getelson-at-mellanox/rdpdk/tree/safe-q

Several changes:

  *
DPDK configuration is split to 3 mandatory steps:
     *
port configuration in
Port::configure(&mut self, rxq_num: u16, txq_num: u16) -> Result<(), String>
     *
Rx queues configuration in
Port::config_rxqs(&mut self, desc_num: u16, mempool: DpdkMempool) -> Result<(), String>
     *
Tx queues configuration in
Port::config_txqs(&mut self, desc_num: u16) -> Result<(), String>
  *
In the IO thread, I renamed the `enable_polling()` to `activate()` for Rx/Tx symmetry.
  *
I renamed `port` and `q` struct members to `port_id`, `queue_id`

Build steps:


  1.
Apply https://github.com/getelson-at-mellanox/rdpdk/blob/safe-q/dpdk-patches/0001-rust-export-missing-port-objects.patch to DPDK source.
  2.
Install DPDK
  3.
Set PKG_CONFIG_PATH to DPDK installation

Activation:

# cargo run --example echo -- -a <port PCI address>

Regards,
Gregory




________________________________
From: Van Haaren, Harry <harry.van.haaren@intel.com>
Sent: Friday, April 18, 2025 14:40
To: Gregory Etelson <getelson@nvidia.com>
Cc: dev@dpdk.org <dev@dpdk.org>; Richardson, Bruce <bruce.richardson@intel.com>; owen.hilyard@unh.edu <owen.hilyard@unh.edu>
Subject: Re: [PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

External email: Use caution opening links or attachments


> From: Etelson, Gregory
> Sent: Thursday, April 17, 2025 7:58 PM
> To: Van Haaren, Harry
> Cc: dev@dpdk.org; getelson@nvidia.com; Richardson, Bruce; owen.hilyard@unh.edu
> Subject: Re: [PATCH] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq
>
> Hello Harry,
>
> Thank you for sharing the API.
> Please check out my comments below.

Thanks for reading & discussion!

<snip>

> > +
> > +            pub fn start(&mut self) -> (Vec<RxqHandle>, Vec<TxqHandle>) {
> > +                // call rte_eth_dev_start() here, then give ownership of Rxq/Txq to app
>
> After a call to Port::start, Rx and Tx queues are detached from it's port.
> With that model how rte_eth_dev_stop() and subsequent rte_eth_dev_start()
> DPDK calls can be implemented ?

Correct, the RxqHandle and TxqHandle don't have a "back reference" to the port.
There are a number of ways to ensure eth_dev_stop() cannot be called without the
Rxq/Txqs being "returned" to the Port instance first.

Eg: Use an Arc<T>. The port instance "owns" the Arc<T>, which means it is going to keep
   the Arc alive. Now give each Rxq/Txq a clone of this Arc. When the Drop impl of the
   Rxq/Txq runs, it will decrement the Arc. So just letting the Rxq/Txq go out of scope
   will be enough to have the Port understand that handle is now gone.

   The port itself can use Arc::into_inner function[1], which returns Option<T>. If the
   Some(T) is returned, then all instances of RxqHandle/TxqHandle have been dropped,
   meaning it is safe to eth_dev_stop(), as it is impossible to poll RXQs if there's no Rxq :)
   [1] https://doc.rust-lang.org/std/sync/struct.Arc.html#method.into_inner

// Pseudo-code here:
Dpdk::Eth::Port::stop(&mut self) -> Result<(), Error> {
    let handles_dropped = self.handle_arc.into_inner(); // returns "T" if its the only reference to the Arc
    if handles_dropped.is_none() {
        return Err("an Rxq or Txq handle remains alive, cannot safely stop this port");
    }
}

There's probably a few others, but that's "idiomatic Rust" solution.
We'd have to pass the Arc from the RxqHandle into the Rxq instance itself too,
but that's fine.

<snip>

> > +fn main() {
> > +    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
> > +    let rx_mempool = dpdk::Mempool::new(4096);
> > +
> > +    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
>
> Eal::take_eth_ports() resets EAL ports.

I don't think it "resets" here. The "take eth ports" removes the Port instances from
the dpdk::Eal struct, but there's no "reset" behaviour.

> A call to rte_dev_probe() will ether fail, because Eal::eth_ports is None
> or create another port-0, depending on implementation.

I don't see how or why rte_dev_probe() would be called. The idea is not to allow Rust
apps call DPDK C APIs "when they want". The safe Rust API provides the required abstraction.
So its not possible to have another call to rte_dev_probe(), after the 1st time under eal_init().

Similar topic: Hotplug. I have experience with designing C APIs around hotplug
use-cases (Music/DJ software, from before my DPDK/networking days!). I think DPDK has
an interesting "push hotplug" approach (aka, App makes a function call to "request" the device).
Then on successful return, we can call rte_eth_dev_get_port_by_name() to get the u16 port_id,
and build the Port instance from that. Outline API:

enum EalHotplugDev {
    EthDev(Dpdk::Eth::Port), // enums can have contents in Rust :)
    CryptoDev(Dpdk::Crypto),
    // Etc
}

Eal::hotplug_add(bus: String, dev: String, args: String) -> Result<EalHotplugDev, Error> {
    // TODO: call rte_eal_hotplug_add()
    // TODO: identify how to know if its an Eth, Crypto, Dma, or other dev type?
    match (dev_type) {
        "eth" => {
            let port_id = rte_eth_dev_get_port_by_name(dev);
            EalHotplugDev::EthDev( Dpdk::Eth::Port::new(port_id) )
        }
    }
}

Applications could then do:
  let Ok(dev) = eal.hotplug_add("pci", "02:00.0", "dev_option=true") else {
      // failed to hotplug, log error?
      return;
  }
  match (dev) {
      EthDev => {
          // handle the dev here, e.g. configure & spawn thread to poll Rxq like before.
      }
  }

I like having an outline of difficult to "bolt on" features (hotplug is typically hard to add later..)
but I recommend we focus on getting core APIs and such running before more detail/time/implementation here.


Regards, -Harry

[-- Attachment #2: Type: text/html, Size: 14136 bytes --]

[PATCH 1/3] rust: RFC/demo of safe API for Dpdk Eal, Eth and Rxq

[Harry van Haaren]

This patch is NOT to be considered for merge, it is a demo
of the Rust APIs for Ethdev. There is no actual implementation
of the APIs against the DPDK C functions, this is Rust API only.

To test/run the code (and uncomment things to see errors)
just apply this patch, cd "rust_api_example" and run
$ cargo run

This will compile the API, and spawn 2x threads to poll on
two Rxq instances. The comments in the code explain how the
"Send" and "Sync" attributes are captured per instances of a
struct (e.g. how RxqHandle -> Rxq restricts thread movement).

Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 rust_api_example/Cargo.toml  |   6 ++
 rust_api_example/src/main.rs | 189 +++++++++++++++++++++++++++++++++++
 2 files changed, 195 insertions(+)
 create mode 100644 rust_api_example/Cargo.toml
 create mode 100644 rust_api_example/src/main.rs

diff --git a/rust_api_example/Cargo.toml b/rust_api_example/Cargo.toml
new file mode 100644
index 0000000000..0137826340
--- /dev/null
+++ b/rust_api_example/Cargo.toml
@@ -0,0 +1,6 @@
+[package]
+name = "rust_api_example"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
diff --git a/rust_api_example/src/main.rs b/rust_api_example/src/main.rs
new file mode 100644
index 0000000000..8d0de50c30
--- /dev/null
+++ b/rust_api_example/src/main.rs
@@ -0,0 +1,189 @@
+// Outline for safe DPDK API bindings
+//  - None of the APIs are actually implemented, this is API design only
+//  - This demo runs 2x threads on 2x Rxqs, and cannot accidentally poll incorrectly
+
+pub mod dpdk {
+    pub mod eth {
+        use super::Mempool;
+
+        #[derive(Debug)]
+        pub struct TxqHandle {/* todo: but same as Rxq */}
+
+        // Handle allows moving between threads, its not polling!
+        #[derive(Debug)]
+        pub struct RxqHandle {
+            port: u16,
+            queue: u16,
+        }
+
+        impl RxqHandle {
+            pub(crate) fn new(port: u16, queue: u16) -> Self {
+                RxqHandle { port, queue }
+            }
+
+            // This function is the key to the API design: it ensures the rx_burst()
+            // function is only available via the Rxq struct, after enable_polling() has been called.
+            // It "consumes" (takes "self" as a parameter, not a '&' reference!) which essentially
+            // destroys/invalidates the handle from the Application level code.
+
+            // It returns an Rxq instance, which has the PhantomData to encode the threading requirements,
+            // and the Rxq has the rx_burst() function: this allows the application to recieve packets.
+            pub fn enable_polling(self) -> Rxq {
+                Rxq {
+                    handle: self,
+                    _phantom: std::marker::PhantomData,
+                }
+            }
+        }
+
+        #[derive(Debug)]
+        pub struct Rxq {
+            handle: RxqHandle,
+            // This "PhantomData" tells the rust compiler to Pretend the Rc<()> is in this struct
+            // but in practice it is a Zero-Sized-Type, so takes up no space. It is a compile-time
+            // language technique to ensure the struct is not moved between threads. This encodes
+            // the API requirement "don't poll from multiple threads without synchronisation (e.g. Mutex)"
+            _phantom: std::marker::PhantomData<std::rc::Rc<()>>,
+        }
+
+        impl Rxq {
+            // TODO: datapath Error types should be lightweight, not String. Here we return ().
+            pub fn rx_burst(&mut self, _mbufs: &mut [u8]) -> Result<usize, ()> {
+                // TODO: Design the Mbuf struct wrapper, and how to best return a batch
+                //  e.g.: investigate "ArrayVec" crate for safe & fixed sized, stack allocated arrays
+                //
+                // There is work to do here, but I want to communicate the general DPDK/EAL/Eth/Rxq concepts
+                // now, this part is not done yet: it is likely the hardest/most performance critical.
+                //
+                // call rte_eth_rx_burst() here
+                println!(
+                    "[thread: {:?}] rx_burst: port {} queue {}",
+                    std::thread::current().id(),
+                    self.handle.port,
+                    self.handle.queue
+                );
+                Ok(0)
+            }
+        }
+
+        #[derive(Debug)]
+        pub struct Port {
+            id: u16,
+            rxqs: Vec<RxqHandle>,
+            txqs: Vec<TxqHandle>,
+        }
+
+        impl Port {
+            // pub(crate) here ensures outside this crate users cannot call this function
+            pub(crate) fn from_u16(id: u16) -> Self {
+                Port {
+                    id,
+                    rxqs: Vec::new(),
+                    txqs: Vec::new(),
+                }
+            }
+
+            pub fn rxqs(&mut self, rxq_count: u16, _mempool: Mempool) -> Result<(), String> {
+                for q in 0..rxq_count {
+                    // call rte_eth_rx_queue_setup() here
+                    self.rxqs.push(RxqHandle::new(self.id, q));
+                }
+                Ok(())
+            }
+
+            pub fn start(&mut self) -> (Vec<RxqHandle>, Vec<TxqHandle>) {
+                // call rte_eth_dev_start() here, then give ownership of Rxq/Txq to app
+                (
+                    std::mem::take(&mut self.rxqs),
+                    std::mem::take(&mut self.txqs),
+                )
+            }
+        }
+    }
+
+    #[derive(Debug, Clone)]
+    // Mempool is a long-life object, which many other DPDK things refer to (e.g. rxq config)
+    // Having a Rust lifetime attached to it (while technically correct) would complicate the
+    // code a LOT, and for little value. This is a tradeoff - happy to discuss more if we want.
+    // The choice here is to derive "Clone", allowing handing over multiple instances of the
+    // same Mempool, similar to how Arc<Mempool> would work, but without the reference counting.
+    pub struct Mempool {}
+
+    impl Mempool {
+        pub fn new(_size: usize) -> Self {
+            Self {}
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct Eal {
+        eth_ports: Option<Vec<eth::Port>>,
+    }
+
+    impl Eal {
+        //  allow init once,
+        pub fn init() -> Result<Self, String> {
+            // EAL init() will do PCI probe and VDev enumeration will find/create eth ports.
+            // This code should loop over the ports, and build up Rust structs representing them
+            let eth_port = vec![eth::Port::from_u16(0)];
+            Ok(Eal {
+                eth_ports: Some(eth_port),
+            })
+        }
+
+        // API to get eth ports, taking ownership. It can be called once.
+        // The return will be None for future calls
+        pub fn take_eth_ports(&mut self) -> Option<Vec<eth::Port>> {
+            self.eth_ports.take()
+        }
+    }
+
+    impl Drop for Eal {
+        fn drop(&mut self) {
+            // todo: rte_eal_cleanup()
+        }
+    }
+} // DPDK mod
+
+fn main() {
+    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
+    let rx_mempool = dpdk::Mempool::new(4096);
+
+    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
+    let mut p = ports.pop().unwrap();
+
+    p.rxqs(2, rx_mempool).expect("rxqs setup ok");
+    println!("{:?}", p);
+
+    let (mut rxqs, _txqs) = p.start();
+    println!("rxqs: {:?}", rxqs);
+
+    let rxq1 = rxqs.pop().unwrap();
+    let rxq2 = rxqs.pop().unwrap();
+
+    // spawn a new thread to use rxq1. This demonstrates that the RxqHandle
+    // type can move between threads - it is not tied to the thread that created it.
+    std::thread::spawn(move || {
+        // Uncomment this: it fails to compile!
+        //   - Rxq2 would be used by this newly-spawned thread
+        //     -- specifically the variable was "moved" into this thread
+        //   - it is also used below (by the main thread)
+        // "value used after move" is the error, on the below code
+        // let mut rxq = rxq2.enable_polling();
+
+        // see docs on enable_polling above to understand how the enable_polling()
+        // function helps to achieve the thread-safety-at-compile-time goal.
+        let mut rxq = rxq1.enable_polling();
+        loop {
+            let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+            std::thread::sleep(std::time::Duration::from_millis(1000));
+        }
+    });
+
+    // main thread polling rxq2
+    let mut rxq = rxq2.enable_polling();
+    loop {
+        let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+        std::thread::sleep(std::time::Duration::from_millis(1000));
+    }
+}
-- 
2.34.1

[PATCH 2/3] rust: split main into example, refactor to lib.rs

[Harry van Haaren]

Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 rust_api_example/examples/eth_poll.rs    | 35 +++++++++++++++++++
 rust_api_example/src/{main.rs => lib.rs} | 43 ------------------------
 2 files changed, 35 insertions(+), 43 deletions(-)
 create mode 100644 rust_api_example/examples/eth_poll.rs
 rename rust_api_example/src/{main.rs => lib.rs} (77%)

diff --git a/rust_api_example/examples/eth_poll.rs b/rust_api_example/examples/eth_poll.rs
new file mode 100644
index 0000000000..cde28df68d
--- /dev/null
+++ b/rust_api_example/examples/eth_poll.rs
@@ -0,0 +1,35 @@
+// Examples should not require any "unsafe" code.
+#![deny(unsafe_code)]
+
+use rust_api_example::dpdk::{self};
+
+fn main() {
+    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
+    let rx_mempool = dpdk::Mempool::new(4096);
+
+    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
+    let mut p = ports.pop().unwrap();
+
+    p.rxqs(2, rx_mempool).expect("rxqs setup ok");
+    println!("{:?}", p);
+
+    let (mut rxqs, _txqs) = p.start();
+    println!("rxqs: {:?}", rxqs);
+
+    let rxq1 = rxqs.pop().unwrap();
+    let rxq2 = rxqs.pop().unwrap();
+
+    std::thread::spawn(move || {
+        let mut rxq = rxq1.enable_polling();
+        loop {
+            let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+            std::thread::sleep(std::time::Duration::from_millis(1000));
+        }
+    });
+
+    let mut rxq = rxq2.enable_polling();
+    loop {
+        let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+        std::thread::sleep(std::time::Duration::from_millis(1000));
+    }
+}
\ No newline at end of file
diff --git a/rust_api_example/src/main.rs b/rust_api_example/src/lib.rs
similarity index 77%
rename from rust_api_example/src/main.rs
rename to rust_api_example/src/lib.rs
index 8d0de50c30..0d13b06d85 100644
--- a/rust_api_example/src/main.rs
+++ b/rust_api_example/src/lib.rs
@@ -144,46 +144,3 @@ pub mod dpdk {
         }
     }
 } // DPDK mod
-
-fn main() {
-    let mut dpdk = dpdk::Eal::init().expect("dpdk must init ok");
-    let rx_mempool = dpdk::Mempool::new(4096);
-
-    let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
-    let mut p = ports.pop().unwrap();
-
-    p.rxqs(2, rx_mempool).expect("rxqs setup ok");
-    println!("{:?}", p);
-
-    let (mut rxqs, _txqs) = p.start();
-    println!("rxqs: {:?}", rxqs);
-
-    let rxq1 = rxqs.pop().unwrap();
-    let rxq2 = rxqs.pop().unwrap();
-
-    // spawn a new thread to use rxq1. This demonstrates that the RxqHandle
-    // type can move between threads - it is not tied to the thread that created it.
-    std::thread::spawn(move || {
-        // Uncomment this: it fails to compile!
-        //   - Rxq2 would be used by this newly-spawned thread
-        //     -- specifically the variable was "moved" into this thread
-        //   - it is also used below (by the main thread)
-        // "value used after move" is the error, on the below code
-        // let mut rxq = rxq2.enable_polling();
-
-        // see docs on enable_polling above to understand how the enable_polling()
-        // function helps to achieve the thread-safety-at-compile-time goal.
-        let mut rxq = rxq1.enable_polling();
-        loop {
-            let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
-            std::thread::sleep(std::time::Duration::from_millis(1000));
-        }
-    });
-
-    // main thread polling rxq2
-    let mut rxq = rxq2.enable_polling();
-    loop {
-        let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
-        std::thread::sleep(std::time::Duration::from_millis(1000));
-    }
-}
-- 
2.34.1

[PATCH 3/3] rust: showcase port Rxq return for stop() and reconfigure

[Harry van Haaren]

Since the refactor, use this command to run/test:
  cargo r --example eth_poll

Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 rust_api_example/examples/eth_poll.rs | 45 ++++++++++++++++++++---
 rust_api_example/src/lib.rs           | 52 ++++++++++++++++++++++++---
 2 files changed, 88 insertions(+), 9 deletions(-)

diff --git a/rust_api_example/examples/eth_poll.rs b/rust_api_example/examples/eth_poll.rs
index cde28df68d..0ef0a28ab9 100644
--- a/rust_api_example/examples/eth_poll.rs
+++ b/rust_api_example/examples/eth_poll.rs
@@ -10,7 +10,7 @@ fn main() {
     let mut ports = dpdk.take_eth_ports().expect("take eth ports ok");
     let mut p = ports.pop().unwrap();
 
-    p.rxqs(2, rx_mempool).expect("rxqs setup ok");
+    p.rxqs(2, rx_mempool.clone()).expect("rxqs setup ok");
     println!("{:?}", p);
 
     let (mut rxqs, _txqs) = p.start();
@@ -21,15 +21,50 @@ fn main() {
 
     std::thread::spawn(move || {
         let mut rxq = rxq1.enable_polling();
-        loop {
+        for _ in 0..3 {
             let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
             std::thread::sleep(std::time::Duration::from_millis(1000));
         }
     });
 
-    let mut rxq = rxq2.enable_polling();
-    loop {
-        let _nb_mbufs = rxq.rx_burst(&mut [0; 32]);
+    // "shadowing" variables is a common pattern in Rust, and is used here to
+    // allow us to use the same variable name but for Rxq instead of RxqHandle.
+    let mut rxq2 = rxq2.enable_polling();
+    for _ in 0..2 {
+        let _nb_mbufs = rxq2.rx_burst(&mut [0; 32]);
         std::thread::sleep(std::time::Duration::from_millis(1000));
     }
+
+    // Important! As Port::stop() relies on RxqHandle's being dropped to
+    // reduce the refcount, if the rxq is NOT dropped, it will NOT allow
+    // the port to be stopped. This is actually a win for Safety (no polling stopped NIC ports)
+    // but also a potential bug/hiccup at application code level.
+    // Uncomment this line to see the loop below stall forever (waiting for Arc ref count to drop from 2 to 1)
+    drop(rxq2);
+
+    loop {
+        let r = p.stop();
+        match r {
+            Ok(_v) => {
+                println!("stopping port");
+                break;
+            }
+            Err(e) => {
+                println!("stop() returns error: {}", e);
+            }
+        };
+        std::thread::sleep(std::time::Duration::from_millis(300));
+    }
+
+    // Reconfigure after stop()
+    p.rxqs(4, rx_mempool.clone()).expect("rxqs setup ok");
+    println!("{:?}", p);
+
+    // queues is a tuple of (rxqs, txqs) here
+    let queues = p.start();
+    println!("queues: {:?}", queues);
+    drop(queues);
+
+    p.stop().expect("stop() ok");
+    println!("stopped port");
 }
\ No newline at end of file
diff --git a/rust_api_example/src/lib.rs b/rust_api_example/src/lib.rs
index 0d13b06d85..6b795fc227 100644
--- a/rust_api_example/src/lib.rs
+++ b/rust_api_example/src/lib.rs
@@ -5,20 +5,47 @@
 pub mod dpdk {
     pub mod eth {
         use super::Mempool;
-
+        use std::sync::Arc;
+
+        // PortHandle here is used as a refcount of "Outstanding Rx/Tx queues".
+        // This is useful, but the "runstate" of the port is also useful. They are
+        // similar, but not identical. A more elegant solution is likely possible.
+        #[derive(Debug, Clone)]
+        #[allow(unused)]
+        pub(crate) struct PortHandle(Arc<()>);
+
+        impl PortHandle {
+            fn new() -> Self {
+                PortHandle(Arc::new(()))
+            }
+            fn stop(&mut self) -> Result<(), usize> {
+                // if the count is 1, only the Port itself has a handle left.
+                // In that case, the count cannot go up, so we can stop.
+                // The strange "Arc::<()>::function()" syntax here is "Fully qualified syntax":
+                //  - https://doc.rust-lang.org/std/sync/struct.Arc.html#deref-behavior
+                let sc = Arc::<()>::strong_count(&self.0);
+                if  sc == 1 {
+                    Ok(())
+                } else {
+                    Err(sc)
+                }
+            }
+        }
+        
         #[derive(Debug)]
         pub struct TxqHandle {/* todo: but same as Rxq */}
 
         // Handle allows moving between threads, its not polling!
         #[derive(Debug)]
         pub struct RxqHandle {
+            _handle: PortHandle,
             port: u16,
             queue: u16,
         }
 
         impl RxqHandle {
-            pub(crate) fn new(port: u16, queue: u16) -> Self {
-                RxqHandle { port, queue }
+            pub(crate) fn new(handle: PortHandle, port: u16, queue: u16) -> Self {
+                RxqHandle { _handle: handle, port, queue }
             }
 
             // This function is the key to the API design: it ensures the rx_burst()
@@ -68,6 +95,7 @@ pub mod dpdk {
 
         #[derive(Debug)]
         pub struct Port {
+            handle: PortHandle,
             id: u16,
             rxqs: Vec<RxqHandle>,
             txqs: Vec<TxqHandle>,
@@ -77,6 +105,7 @@ pub mod dpdk {
             // pub(crate) here ensures outside this crate users cannot call this function
             pub(crate) fn from_u16(id: u16) -> Self {
                 Port {
+                    handle: PortHandle::new(),
                     id,
                     rxqs: Vec::new(),
                     txqs: Vec::new(),
@@ -84,10 +113,14 @@ pub mod dpdk {
             }
 
             pub fn rxqs(&mut self, rxq_count: u16, _mempool: Mempool) -> Result<(), String> {
+                // ensure no old ports remain
+                self.rxqs.clear();
+
                 for q in 0..rxq_count {
                     // call rte_eth_rx_queue_setup() here
-                    self.rxqs.push(RxqHandle::new(self.id, q));
+                    self.rxqs.push(RxqHandle::new(self.handle.clone(), self.id, q));
                 }
+                println!("{:?}", self.handle);
                 Ok(())
             }
 
@@ -98,6 +131,17 @@ pub mod dpdk {
                     std::mem::take(&mut self.txqs),
                 )
             }
+
+            pub fn stop(&mut self) -> Result<(), String> {
+                match self.handle.stop() {
+                    Ok(_v) => {
+                        // call rte_eth_dev_stop() here
+                        println!("stopping port {}", self.id);
+                        Ok(())
+                    }
+                    Err(e) => Err(format!("Port has {} Rxq/Txq handles outstanding", e)),
+                }
+            }
         }
     }
 
-- 
2.34.1