Since the refactor, use this command to run/test:
cargo r --example eth_poll
Signed-off-by: Harry van Haaren <harry.van.haa...@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
