new port rx and tx loops

Author: drunkirishcoder

core/src/ffi/dpdk.rs

@@ -22,7 +22,8 @@ use crate::{debug, error};
 use anyhow::Result;
 use capsule_ffi as cffi;
 use std::fmt;
-use std::ops::DerefMut;
+use std::mem;
+use std::ops::{Deref, DerefMut};
 use std::os::raw;
 use std::panic::{self, AssertUnwindSafe};
 use std::ptr;
@@ -90,6 +91,15 @@ impl From<raw::c_int> for SocketId {
 /// A `rte_mempool` pointer.
 pub(crate) type MempoolPtr = EasyPtr<cffi::rte_mempool>;
 
+impl Clone for MempoolPtr {
+    fn clone(&self) -> Self {
+        self.0.clone().into()
+    }
+}
+
+// Allows the pointer to go across thread/lcore boundaries.
+unsafe impl Send for MempoolPtr {}
+
 /// Creates a mbuf pool.
 pub(crate) fn pktmbuf_pool_create<S: Into<String>>(
     name: S,
@@ -125,9 +135,14 @@ pub(crate) fn mempool_lookup<S: Into<String>>(name: S) -> Result<MempoolPtr> {
     Ok(EasyPtr(ptr))
 }
 
+/// Returns the number of elements which have been allocated from the mempool.
+pub(crate) fn mempool_in_use_count(mp: &MempoolPtr) -> usize {
+    unsafe { cffi::rte_mempool_in_use_count(mp.deref()) as usize }
+}
+
 /// Frees a mempool.
-pub(crate) fn mempool_free(ptr: &mut MempoolPtr) {
-    unsafe { cffi::rte_mempool_free(ptr.deref_mut()) };
+pub(crate) fn mempool_free(mp: &mut MempoolPtr) {
+    unsafe { cffi::rte_mempool_free(mp.deref_mut()) };
 }
 
 /// An opaque identifier for a logical execution unit of the processor.
@@ -353,10 +368,26 @@ pub(crate) fn eth_dev_configure(
     }
 }
 
+/// An opaque identifier for a port's receive queue.
+#[derive(Copy, Clone)]
+pub(crate) struct PortRxQueueId(u16);
+
+impl fmt::Debug for PortRxQueueId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "rxq{}", self.0)
+    }
+}
+
+impl From<usize> for PortRxQueueId {
+    fn from(id: usize) -> Self {
+        PortRxQueueId(id as u16)
+    }
+}
+
 /// Allocates and sets up a receive queue for a device.
 pub(crate) fn eth_rx_queue_setup(
     port_id: PortId,
-    rx_queue_id: usize,
+    rx_queue_id: PortRxQueueId,
     nb_rx_desc: usize,
     socket_id: SocketId,
     rx_conf: Option<&cffi::rte_eth_rxconf>,
@@ -365,7 +396,7 @@ pub(crate) fn eth_rx_queue_setup(
     unsafe {
         cffi::rte_eth_rx_queue_setup(
             port_id.0,
-            rx_queue_id as u16,
+            rx_queue_id.0,
             nb_rx_desc as u16,
             socket_id.0 as raw::c_uint,
             rx_conf.map_or(ptr::null(), |conf| conf),
@@ -376,18 +407,50 @@ pub(crate) fn eth_rx_queue_setup(
     }
 }
 
+/// Retrieves a burst of input packets from a receive queue of a device.
+pub(crate) fn eth_rx_burst(port_id: PortId, queue_id: PortRxQueueId, rx_pkts: &mut Vec<MbufPtr>) {
+    let nb_pkts = rx_pkts.capacity();
+
+    unsafe {
+        let len = cffi::_rte_eth_rx_burst(
+            port_id.0,
+            queue_id.0,
+            rx_pkts.as_mut_ptr() as *mut *mut cffi::rte_mbuf,
+            nb_pkts as u16,
+        );
+
+        rx_pkts.set_len(len as usize);
+    }
+}
+
+/// An opaque identifier for a port's transmit queue.
+#[derive(Copy, Clone)]
+pub(crate) struct PortTxQueueId(u16);
+
+impl fmt::Debug for PortTxQueueId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "txq{}", self.0)
+    }
+}
+
+impl From<usize> for PortTxQueueId {
+    fn from(id: usize) -> Self {
+        PortTxQueueId(id as u16)
+    }
+}
+
 /// Allocates and sets up a transmit queue for a device.
 pub(crate) fn eth_tx_queue_setup(
     port_id: PortId,
-    tx_queue_id: usize,
+    tx_queue_id: PortTxQueueId,
     nb_tx_desc: usize,
     socket_id: SocketId,
     tx_conf: Option<&cffi::rte_eth_txconf>,
 ) -> Result<()> {
     unsafe {
         cffi::rte_eth_tx_queue_setup(
             port_id.0,
-            tx_queue_id as u16,
+            tx_queue_id.0,
             nb_tx_desc as u16,
             socket_id.0 as raw::c_uint,
             tx_conf.map_or(ptr::null(), |conf| conf),
@@ -397,9 +460,57 @@ pub(crate) fn eth_tx_queue_setup(
     }
 }
 
+/// Sends a burst of output packets on a transmit queue of a device.
+pub(crate) fn eth_tx_burst(
+    port_id: PortId,
+    queue_id: PortTxQueueId,
+    tx_pkts: &mut Vec<MbufPtr>,
+) -> usize {
+    let nb_pkts = tx_pkts.len();
+
+    let sent = unsafe {
+        cffi::_rte_eth_tx_burst(
+            port_id.0,
+            queue_id.0,
+            tx_pkts.as_mut_ptr() as *mut *mut cffi::rte_mbuf,
+            nb_pkts as u16,
+        )
+    } as usize;
+
+    if nb_pkts > sent {
+        // wasn't able to send everything.
+        mem::forget(tx_pkts.drain(..sent));
+    } else {
+        unsafe {
+            tx_pkts.set_len(0);
+        }
+    }
+
+    sent
+}
+
+/// Starts a device.
+pub(crate) fn eth_dev_start(port_id: PortId) -> Result<()> {
+    unsafe {
+        cffi::rte_eth_dev_start(port_id.0)
+            .into_result(DpdkError::from_errno)
+            .map(|_| ())
+    }
+}
+
+/// Stops a device.
+pub(crate) fn eth_dev_stop(port_id: PortId) {
+    unsafe {
+        cffi::rte_eth_dev_stop(port_id.0);
+    }
+}
+
 /// A `rte_mbuf` pointer.
 pub(crate) type MbufPtr = EasyPtr<cffi::rte_mbuf>;
 
+// Allows the pointer to go across thread/lcore boundaries.
+unsafe impl Send for MbufPtr {}
+
 /// Allocates a new mbuf from a mempool.
 pub(crate) fn pktmbuf_alloc(mp: &mut MempoolPtr) -> Result<MbufPtr> {
     let ptr =

core/src/rt2/lcore.rs

@@ -16,51 +16,22 @@
 * SPDX-License-Identifier: Apache-2.0
 */
 
+use super::ShutdownTrigger;
 use crate::ffi::dpdk::{self, LcoreId};
 use crate::{debug, info};
-use anyhow::{anyhow, Result};
-use async_channel::{self, Receiver, Recv, Sender};
+use anyhow::Result;
 use async_executor::Executor;
 use futures_lite::future;
 use std::collections::HashMap;
 use std::future::Future;
 use std::sync::Arc;
-
-/// Trigger for the shutdown.
-pub(crate) struct Trigger(Sender<()>);
-
-impl Trigger {
-    /// Triggers the shutdown.
-    pub(crate) fn fire(self) {
-        drop(self.0)
-    }
-}
-
-/// Shutdown wait handle.
-pub(crate) struct Shutdown(Receiver<()>);
-
-impl Shutdown {
-    /// A future that waits till the trigger is fired.
-    pub(crate) fn wait(&self) -> Recv<'_, ()> {
-        self.0.recv()
-    }
-}
-
-/// Creates a shutdown and trigger pair.
-///
-/// Leverages the behavior of an async channel. When the sender is dropped
-/// from scope, it closes the channel and causes the receiver side future
-/// in the executor queue to resolve.
-pub(crate) fn shutdown_trigger() -> (Trigger, Shutdown) {
-    let (s, r) = async_channel::unbounded();
-    (Trigger(s), Shutdown(r))
-}
+use thiserror::Error;
 
 /// An async executor abstraction on top of a DPDK logical core.
 pub(crate) struct Lcore {
     id: LcoreId,
     executor: Arc<Executor<'static>>,
-    trigger: Option<Trigger>,
+    shutdown: Option<ShutdownTrigger>,
 }
 
 impl Lcore {
@@ -71,23 +42,29 @@ impl Lcore {
     /// Returns `DpdkError` if the executor fails to run on the given lcore.
     fn new(id: LcoreId) -> Result<Self> {
         debug!(?id, "starting lcore.");
+        let trigger = ShutdownTrigger::new();
         let executor = Arc::new(Executor::new());
-        let (trigger, shutdown) = shutdown_trigger();
 
+        let handle = trigger.get_wait();
         let executor2 = Arc::clone(&executor);
         dpdk::eal_remote_launch(id, move || {
             info!(?id, "lcore started.");
-            let _ = future::block_on(executor2.run(shutdown.wait()));
+            let _ = future::block_on(executor2.run(handle.wait()));
             info!(?id, "lcore stopped.");
         })?;
 
         Ok(Lcore {
             id,
             executor,
-            trigger: Some(trigger),
+            shutdown: Some(trigger),
         })
     }
 
+    /// Returns the lcore id.
+    pub(crate) fn id(&self) -> LcoreId {
+        self.id
+    }
+
     /// Spawns an async task and waits for it to complete.
     pub(crate) fn block_on<T: Send + 'static>(
         &self,
@@ -105,22 +82,30 @@ impl Lcore {
 
 impl Drop for Lcore {
     fn drop(&mut self) {
-        if let Some(trigger) = self.trigger.take() {
+        if let Some(trigger) = self.shutdown.take() {
             debug!(id = ?self.id, "stopping lcore.");
             trigger.fire();
         }
     }
 }
 
+/// Lcore not found error.
+#[derive(Debug, Error)]
+#[error("lcore not found.")]
+pub(crate) struct LcoreNotFound;
+
 /// Map to lookup the lcore by the assigned id.
 pub(crate) struct LcoreMap(HashMap<usize, Lcore>);
 
 impl LcoreMap {
     /// Returns the lcore with the assigned id.
-    fn get(&self, id: usize) -> Result<&Lcore> {
-        self.0
-            .get(&id)
-            .ok_or_else(|| anyhow!("lcore with id '{}' not found.", id))
+    pub(crate) fn get(&self, id: usize) -> Result<&Lcore> {
+        self.0.get(&id).ok_or_else(|| LcoreNotFound.into())
+    }
+
+    /// Returns a lcore iterator.
+    pub(crate) fn iter(&self) -> impl Iterator<Item = &Lcore> {
+        self.0.values()
     }
 }