new mbuf allocation

Author: drunkirishcoder

core/src/dpdk/mbuf.rs

@@ -16,15 +16,14 @@
 * SPDX-License-Identifier: Apache-2.0
 */
 
-use super::MEMPOOL;
-use crate::dpdk::{DpdkError, MempoolError};
-use crate::ffi::{self, ToResult};
+use crate::ffi::dpdk::{self, MbufPtr};
 use crate::packets::{Internal, Packet};
+use crate::rt2::Mempool;
 use crate::{ensure, trace};
 use anyhow::Result;
+use capsule_ffi as cffi;
 use std::fmt;
 use std::mem;
-use std::os::raw;
 use std::ptr::{self, NonNull};
 use std::slice;
 use thiserror::Error;
@@ -113,21 +112,21 @@ enum MbufInner {
     /// Original version of the message buffer that should be freed when it goes
     /// out of scope, unless the ownership of the pointer is given back to
     /// DPDK on transmit.
-    Original(NonNull<ffi::rte_mbuf>),
+    Original(NonNull<cffi::rte_mbuf>),
     /// A clone version of the message buffer that should not be freed when
     /// it goes out of scope.
-    Clone(NonNull<ffi::rte_mbuf>),
+    Clone(NonNull<cffi::rte_mbuf>),
 }
 
 impl MbufInner {
-    fn ptr(&self) -> &NonNull<ffi::rte_mbuf> {
+    fn ptr(&self) -> &NonNull<cffi::rte_mbuf> {
         match self {
             MbufInner::Original(raw) => raw,
             MbufInner::Clone(raw) => raw,
         }
     }
 
-    fn ptr_mut(&mut self) -> &mut NonNull<ffi::rte_mbuf> {
+    fn ptr_mut(&mut self) -> &mut NonNull<cffi::rte_mbuf> {
         match self {
             MbufInner::Original(ref mut raw) => raw,
             MbufInner::Clone(ref mut raw) => raw,
@@ -144,23 +143,23 @@ impl Mbuf {
     ///
     /// # Errors
     ///
-    /// Returns `MempoolError::Exhausted` if the allocation of mbuf fails.
+    /// Returns `MempoolPtrUnsetError` if invoked from a non lcore.
+    /// Returns `DpdkError` if the allocation of mbuf fails.
     #[inline]
     pub fn new() -> Result<Self> {
-        let mempool = MEMPOOL.with(|tls| tls.get());
-        let raw =
-            unsafe { ffi::_rte_pktmbuf_alloc(mempool).into_result(|_| MempoolError::Exhausted)? };
+        let mut mp = Mempool::thread_local_ptr()?;
+        let ptr = dpdk::pktmbuf_alloc(&mut mp)?;
 
         Ok(Mbuf {
-            inner: MbufInner::Original(raw),
+            inner: MbufInner::Original(ptr.into()),
         })
     }
 
     /// Creates a new message buffer from a byte array.
     ///
     /// # Errors
     ///
-    /// Returns `MempoolError::Exhausted` if the allocation of mbuf fails.
+    /// Returns `DpdkError` if the allocation of mbuf fails.
     /// Returns `BufferError::NotResized` if the byte array is larger than
     /// the maximum mbuf size.
     #[inline]
@@ -173,21 +172,29 @@ impl Mbuf {
 
     /// Creates a new `Mbuf` from a raw pointer.
     #[inline]
-    pub(crate) unsafe fn from_ptr(ptr: *mut ffi::rte_mbuf) -> Self {
+    pub(crate) fn from_easyptr(ptr: MbufPtr) -> Self {
+        Mbuf {
+            inner: MbufInner::Original(ptr.into()),
+        }
+    }
+
+    /// Creates a new `Mbuf` from a raw pointer.
+    #[inline]
+    pub(crate) unsafe fn from_ptr(ptr: *mut cffi::rte_mbuf) -> Self {
         Mbuf {
             inner: MbufInner::Original(NonNull::new_unchecked(ptr)),
         }
     }
 
     /// Returns the raw struct needed for FFI calls.
     #[inline]
-    fn raw(&self) -> &ffi::rte_mbuf {
+    fn raw(&self) -> &cffi::rte_mbuf {
         unsafe { self.inner.ptr().as_ref() }
     }
 
     /// Returns the raw struct needed for FFI calls.
     #[inline]
-    fn raw_mut(&mut self) -> &mut ffi::rte_mbuf {
+    fn raw_mut(&mut self) -> &mut cffi::rte_mbuf {
         unsafe { self.inner.ptr_mut().as_mut() }
     }
 
@@ -417,7 +424,18 @@ impl Mbuf {
     /// The `Mbuf` is consumed. It is the caller's the responsibility to
     /// free the raw pointer after use. Otherwise the buffer is leaked.
     #[inline]
-    pub(crate) fn into_ptr(self) -> *mut ffi::rte_mbuf {
+    pub(crate) fn into_easyptr(self) -> MbufPtr {
+        let ptr = self.inner.ptr().clone();
+        mem::forget(self);
+        ptr.into()
+    }
+
+    /// Acquires the underlying raw struct pointer.
+    ///
+    /// The `Mbuf` is consumed. It is the caller's the responsibility to
+    /// free the raw pointer after use. Otherwise the buffer is leaked.
+    #[inline]
+    pub(crate) fn into_ptr(self) -> *mut cffi::rte_mbuf {
         let ptr = self.inner.ptr().as_ptr();
         mem::forget(self);
         ptr
@@ -430,25 +448,21 @@ impl Mbuf {
     /// Returns `DpdkError` if the allocation of mbuf fails.
     pub fn alloc_bulk(len: usize) -> Result<Vec<Mbuf>> {
         let mut ptrs = Vec::with_capacity(len);
-        let mempool = MEMPOOL.with(|tls| tls.get());
-
-        let mbufs = unsafe {
-            ffi::_rte_pktmbuf_alloc_bulk(mempool, ptrs.as_mut_ptr(), len as raw::c_uint)
-                .into_result(DpdkError::from_errno)?;
+        let mut mp = Mempool::thread_local_ptr()?;
+        dpdk::pktmbuf_alloc_bulk(&mut mp, &mut ptrs)?;
 
-            ptrs.set_len(len);
-            ptrs.into_iter()
-                .map(|ptr| Mbuf::from_ptr(ptr))
-                .collect::<Vec<_>>()
-        };
+        let mbufs = ptrs.into_iter().map(Mbuf::from_easyptr).collect::<Vec<_>>();
 
         Ok(mbufs)
     }
 
     /// Frees the message buffers in bulk.
     pub(crate) fn free_bulk(mbufs: Vec<Mbuf>) {
-        let ptrs = mbufs.into_iter().map(Mbuf::into_ptr).collect::<Vec<_>>();
-        super::mbuf_free_bulk(ptrs);
+        let mut ptrs = mbufs
+            .into_iter()
+            .map(Mbuf::into_easyptr)
+            .collect::<Vec<_>>();
+        dpdk::pktmbuf_free_bulk(&mut ptrs);
     }
 }
 
@@ -469,9 +483,7 @@ impl Drop for Mbuf {
         match self.inner {
             MbufInner::Original(_) => {
                 trace!("freeing mbuf@{:p}.", self.raw().buf_addr);
-                unsafe {
-                    ffi::_rte_pktmbuf_free(self.raw_mut());
-                }
+                dpdk::pktmbuf_free(self.inner.ptr().clone().into());
             }
             MbufInner::Clone(_) => (),
         }

core/src/ffi/dpdk.rs

@@ -131,7 +131,7 @@ pub(crate) fn mempool_free(ptr: &mut MempoolPtr) {
 }
 
 /// An opaque identifier for a logical execution unit of the processor.
-#[derive(Copy, Clone)]
+#[derive(Copy, Clone, Eq, Hash, PartialEq)]
 pub(crate) struct LcoreId(raw::c_uint);
 
 impl LcoreId {
@@ -397,6 +397,79 @@ pub(crate) fn eth_tx_queue_setup(
     }
 }
 
+/// A `rte_mbuf` pointer.
+pub(crate) type MbufPtr = EasyPtr<cffi::rte_mbuf>;
+
+/// Allocates a new mbuf from a mempool.
+pub(crate) fn pktmbuf_alloc(mp: &mut MempoolPtr) -> Result<MbufPtr> {
+    let ptr =
+        unsafe { cffi::_rte_pktmbuf_alloc(mp.deref_mut()).into_result(|_| DpdkError::new())? };
+
+    Ok(EasyPtr(ptr))
+}
+
+/// Allocates a bulk of mbufs.
+pub(crate) fn pktmbuf_alloc_bulk(mp: &mut MempoolPtr, mbufs: &mut Vec<MbufPtr>) -> Result<()> {
+    let len = mbufs.capacity();
+
+    unsafe {
+        cffi::_rte_pktmbuf_alloc_bulk(
+            mp.deref_mut(),
+            mbufs.as_mut_ptr() as *mut *mut cffi::rte_mbuf,
+            len as raw::c_uint,
+        )
+        .into_result(DpdkError::from_errno)?;
+
+        mbufs.set_len(len);
+    }
+
+    Ok(())
+}
+
+/// Frees a packet mbuf back into its original mempool.
+pub(crate) fn pktmbuf_free(mut m: MbufPtr) {
+    unsafe {
+        cffi::_rte_pktmbuf_free(m.deref_mut());
+    }
+}
+
+/// Frees a bulk of packet mbufs back into their original mempools.
+pub(crate) fn pktmbuf_free_bulk(mbufs: &mut Vec<MbufPtr>) {
+    assert!(!mbufs.is_empty());
+
+    let mut to_free = Vec::with_capacity(mbufs.len());
+    let mut pool = mbufs[0].pool;
+
+    for mbuf in mbufs.drain(..) {
+        if pool == mbuf.pool {
+            to_free.push(mbuf);
+        } else {
+            unsafe {
+                let len = to_free.len();
+                cffi::_rte_mempool_put_bulk(
+                    pool,
+                    to_free.as_ptr() as *const *mut raw::c_void,
+                    len as u32,
+                );
+                to_free.set_len(0);
+            }
+
+            pool = mbuf.pool;
+            to_free.push(mbuf);
+        }
+    }
+
+    unsafe {
+        let len = to_free.len();
+        cffi::_rte_mempool_put_bulk(
+            pool,
+            to_free.as_ptr() as *const *mut raw::c_void,
+            len as u32,
+        );
+        to_free.set_len(0);
+    }
+}
+
 /// An error generated in `libdpdk`.
 ///
 /// When an FFI call fails, the `errno` is translated into `DpdkError`.

core/src/ffi/mod.rs

@@ -160,3 +160,17 @@ impl<T> DerefMut for EasyPtr<T> {
         unsafe { self.0.as_mut() }
     }
 }
+
+// delete later?
+impl<T> From<NonNull<T>> for EasyPtr<T> {
+    fn from(ptr: NonNull<T>) -> Self {
+        EasyPtr(ptr)
+    }
+}
+
+// delete later?
+impl<T> From<EasyPtr<T>> for NonNull<T> {
+    fn from(ptr: EasyPtr<T>) -> Self {
+        ptr.0
+    }
+}

core/src/rt2/mempool.rs

@@ -16,11 +16,14 @@
 * SPDX-License-Identifier: Apache-2.0
 */
 
-use crate::ffi::dpdk::{self, MempoolPtr, SocketId};
+use crate::ffi::dpdk::{self, LcoreId, MempoolPtr, SocketId};
 use crate::ffi::AsStr;
 use crate::{debug, info};
 use anyhow::Result;
+use std::cell::Cell;
 use std::fmt;
+use std::ptr::{self, NonNull};
+use thiserror::Error;
 
 /// A memory pool is an allocator of message buffers, or `Mbuf`. For best
 /// performance, each socket should have a dedicated `Mempool`. However,
@@ -59,6 +62,7 @@ impl Mempool {
     }
 
     /// Returns the raw pointer.
+    #[inline]
     pub(crate) fn ptr_mut(&mut self) -> &mut MempoolPtr {
         &mut self.ptr
     }
@@ -86,6 +90,19 @@ impl Mempool {
     pub(crate) fn socket(&self) -> SocketId {
         self.ptr.socket_id.into()
     }
+
+    /// Returns the thread local mempool pointer.
+    ///
+    /// # Errors
+    ///
+    /// Returns `MempoolPtrUnsetError` if the thread local pointer is not
+    /// set. For example when invoked from a non-EAL thread.
+    pub(crate) fn thread_local_ptr() -> Result<MempoolPtr> {
+        let ptr = MEMPOOL.with(|tls| tls.get());
+        NonNull::new(ptr)
+            .ok_or_else(|| MempoolPtrUnsetError(LcoreId::current()).into())
+            .map(|ptr| ptr.into())
+    }
 }
 
 impl fmt::Debug for Mempool {
@@ -108,6 +125,19 @@ impl Drop for Mempool {
     }
 }
 
+/// The thread local mempool is not set.
+#[derive(Debug, Error)]
+#[error("thread local mempool pointer not set for {0:?}.")]
+pub(crate) struct MempoolPtrUnsetError(LcoreId);
+
+thread_local! {
+    /// The `Mempool` assigned to the core when the core is initialized.
+    /// `Mbuf::new` uses this pool to allocate new buffers when executed on
+    /// the core. For best performance, the `Mempool` and the core should
+    /// share the same socket.
+    pub(crate) static MEMPOOL: Cell<*mut capsule_ffi::rte_mempool> = Cell::new(ptr::null_mut());
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

core/src/testils/mod.rs

@@ -27,11 +27,13 @@ mod rvg;
 pub use self::packet::*;
 pub use self::rvg::*;
 
-use crate::dpdk::{Mempool, SocketId, MEMPOOL};
-use crate::ffi::dpdk;
+use crate::ffi::dpdk::{self, SocketId};
 use crate::metrics;
+use crate::rt2::{Mempool, MEMPOOL};
+use std::ops::DerefMut;
 use std::ptr;
 use std::sync::Once;
+use std::thread;
 
 static TEST_INIT: Once = Once::new();
 
@@ -64,12 +66,11 @@ pub fn cargo_test_init() {
     });
 }
 
-/// A handle that keeps the mempool in scope for the duration of the test. It
-/// will unset the thread-bound mempool on drop.
+/// A RAII guard that keeps the mempool in scope for the duration of the
+/// test. It will unset the thread-bound mempool on drop.
 #[derive(Debug)]
 pub struct MempoolGuard {
-    #[allow(dead_code)]
-    inner: Mempool,
+    _inner: Mempool,
 }
 
 impl Drop for MempoolGuard {
@@ -81,7 +82,8 @@ impl Drop for MempoolGuard {
 /// Creates a new mempool for test that automatically cleans up after the
 /// test completes.
 pub fn new_mempool(capacity: usize, cache_size: usize) -> MempoolGuard {
-    let mut mempool = Mempool::new(capacity, cache_size, SocketId::ANY).unwrap();
-    MEMPOOL.with(|tls| tls.set(mempool.raw_mut()));
-    MempoolGuard { inner: mempool }
+    let name = format!("test-mp-{:?}", thread::current().id());
+    let mut mempool = Mempool::new(name, capacity, cache_size, SocketId::ANY).unwrap();
+    MEMPOOL.with(|tls| tls.set(mempool.ptr_mut().deref_mut()));
+    MempoolGuard { _inner: mempool }
 }