#216 - Fix engine lifecycle cleanup

docs/api-reference/cpp.md

@@ -33,6 +33,11 @@ auto status = daqiri::daqiri_init(config);
 After `daqiri_init()` returns `Status::SUCCESS`, all memory regions are allocated, NIC
 queues are configured, and worker threads are running.
 
+Only one engine may be active in a process. Calling `daqiri_init()` again before
+`shutdown()` returns `Status::INTERNAL_ERROR` and leaves the running engine unchanged.
+After `shutdown()` completes, a later `daqiri_init()` creates a fresh engine instance and
+allocates/registers its resources again.
+
 If GPU RX `reorder_configs` are configured for Raw Ethernet (`stream_type: "raw"`), set
 one CUDA stream per GPU reorder plan before pulling reordered bursts. CPU reorder configs do not use a
 CUDA stream. See the [Configuration YAML Reference](configuration.md#rx-reorder-configs)

docs/api-reference/python.md

@@ -85,6 +85,10 @@ finally:
     daqiri.shutdown()
 ```
 
+Only one engine may be active at a time. A second `daqiri_init()` before `shutdown()`
+returns `Status.INTERNAL_ERROR` without changing the live engine. After `shutdown()`, a
+later initialization starts with a fresh engine instance and fresh resources.
+
 Initialize from a Python dictionary:
 
 ```python

include/daqiri/common.h

@@ -858,8 +858,8 @@ void set_header(BurstParams *burst, uint16_t port, uint16_t q, int64_t num,
 void format_eth_addr(char *dst, std::string addr);
 
 /**
- * @brief Shut down the daqiri and do any cleanup necessary. Freeing memory is
- * done in the engine's destructor.
+ * @brief Shut down DAQIRI and release the active engine and its resources.
+ * A subsequent daqiri_init() creates a fresh engine instance.
  *
  */
 void shutdown();

src/common.cpp

@@ -47,6 +47,15 @@ static Engine* g_daqiri_engine = nullptr;
 
 namespace {
 
+void reset_active_engine() {
+  Engine* engine = g_daqiri_engine;
+  g_daqiri_engine = nullptr;
+  if (engine != nullptr) {
+    engine->shutdown();
+  }
+  EngineFactory::reset();
+}
+
 constexpr size_t kEthAddrLength = 6;
 constexpr size_t kEthAddrOctetLength = 2;
 
@@ -523,7 +532,7 @@ void* get_packet_ptr(BurstParams* burst, int idx) {
 
 void shutdown() {
   ASSERT_DAQIRI_ENGINE_INITIALIZED();
-  g_daqiri_engine->shutdown();
+  reset_active_engine();
   metrics::shutdown();
 }
 
@@ -580,6 +589,11 @@ void print_stats() {
 }
 
 Status daqiri_init(NetworkConfig& config) {
+  if (g_daqiri_engine != nullptr) {
+    DAQIRI_LOG_ERROR("DAQIRI is already initialized; call shutdown() before daqiri_init()");
+    return Status::INTERNAL_ERROR;
+  }
+
   if (config.common_.engine_type == EngineType::UNKNOWN) {
     if (is_explicit_engine_type(config.common_.engine)) {
       config.common_.engine_type = config.common_.engine;
@@ -653,13 +667,19 @@ Status daqiri_init(NetworkConfig& config) {
 
   auto engine = &(EngineFactory::get_active_engine());
 
-  if (!engine->set_config_and_initialize(config)) { return Status::INTERNAL_ERROR; }
+  if (!engine->set_config_and_initialize(config)) {
+    reset_active_engine();
+    metrics::shutdown();
+    return Status::INTERNAL_ERROR;
+  }
 
   for (const auto& intf : config.ifs_) {
     const auto& rx = intf.rx_;
     auto port = engine->get_port_id(intf.address_);
     if (port < 0) {
       DAQIRI_LOG_ERROR("Failed to get port from name {}", intf.address_);
+      reset_active_engine();
+      metrics::shutdown();
       return Status::INVALID_PARAMETER;
     }
   }

src/engine.cpp

@@ -157,6 +157,58 @@ bool validate_flow_action_config(const FlowAction& action, const std::string& fl
 
 }  // namespace
 
+Engine::~Engine() {
+  free_memory_regions();
+}
+
+void Engine::free_memory_regions() noexcept {
+  // DPDK's external-memory descriptors must be released before their backing
+  // allocations. EAL-backed HUGE allocations themselves are released by
+  // rte_eal_cleanup(), which runs in DpdkEngine before this base destructor.
+  ext_pktmbufs_.clear();
+  // Safety net for partial teardown. Normal DPDK shutdown closes these after
+  // rte_eal_cleanup(), then clears the vector before reaching this destructor.
+  for (const int fd : ext_dmabuf_fds_) {
+    if (fd >= 0) {
+      close(fd);
+    }
+  }
+  ext_dmabuf_fds_.clear();
+
+  for (auto& [name, region] : ar_) {
+    (void)name;
+    if (region.ptr_ == nullptr) {
+      continue;
+    }
+
+    switch (region.deallocator_) {
+      case AllocRegion::Deallocator::FREE:
+        std::free(region.ptr_);
+        break;
+      case AllocRegion::Deallocator::CUDA_HOST:
+        if (region.affinity_ >= 0) {
+          cudaSetDevice(region.affinity_);
+        }
+        cudaFreeHost(region.ptr_);
+        break;
+      case AllocRegion::Deallocator::CUDA_DEVICE:
+        if (region.affinity_ >= 0) {
+          cudaSetDevice(region.affinity_);
+        }
+        cuMemFree(reinterpret_cast<CUdeviceptr>(region.ptr_));
+        break;
+      case AllocRegion::Deallocator::MUNMAP:
+        munmap(region.ptr_, region.size_);
+        break;
+      case AllocRegion::Deallocator::EAL:
+      case AllocRegion::Deallocator::NONE:
+        break;
+    }
+    region.ptr_ = nullptr;
+  }
+  ar_.clear();
+}
+
 std::string Engine::generate_random_string(int len) {
   constexpr char tokens[] = "abcdefghijklmnopqrstuvwxyz";
   if (len <= 0) { return {}; }
@@ -329,16 +381,20 @@ static void bind_region_numa(void* p, size_t bytes, int numa) {
   (void)numa;
 }
 
-void* Engine::alloc_huge(size_t bytes, int numa) {
+void* Engine::alloc_huge(size_t bytes, int numa, AllocRegion::Deallocator* deallocator) {
   // Base (non-DPDK) implementation: try a real hugepage-backed mapping, falling
   // back to ordinary page-aligned host memory if MAP_HUGETLB is unavailable.
   // The DPDK engine overrides this to use rte_malloc_socket (EAL hugepages,
   // IOVA-contiguous for the NIC) -- see DpdkEngine::alloc_huge.
+  if (deallocator == nullptr) {
+    return nullptr;
+  }
 #if defined(MAP_HUGETLB)
   void* p = mmap(nullptr, bytes, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB,
                  -1, 0);
   if (p != MAP_FAILED) {
     bind_region_numa(p, bytes, numa);
+    *deallocator = AllocRegion::Deallocator::MUNMAP;
     return p;
   }
   DAQIRI_LOG_WARN(
@@ -351,6 +407,7 @@ void* Engine::alloc_huge(size_t bytes, int numa) {
     return nullptr;
   }
   bind_region_numa(ptr, bytes, numa);
+  *deallocator = AllocRegion::Deallocator::FREE;
   return ptr;
 }
 
@@ -359,7 +416,10 @@ Status Engine::allocate_memory_regions() {
   for (auto& mr : cfg_.mrs_) {
     void* ptr = nullptr;
     AllocRegion ar;
+    ar.mr_name_ = mr.second.name_;
+    ar.affinity_ = mr.second.affinity_;
     mr.second.ttl_size_ = align_ceil(mr.second.adj_size_ * mr.second.num_bufs_, GPU_PAGE_SIZE);
+    ar.size_ = mr.second.ttl_size_;
 
     if (mr.second.owned_) {
       switch (mr.second.kind_) {
@@ -368,16 +428,18 @@ Status Engine::allocate_memory_regions() {
             DAQIRI_LOG_CRITICAL("Failed to allocate aligned host memory!");
             return Status::NULL_PTR;
           }
+          ar.deallocator_ = AllocRegion::Deallocator::FREE;
           break;
         case MemoryKind::HOST_PINNED:
           cudaSetDevice(mr.second.affinity_);
           if (cudaHostAlloc(&ptr, mr.second.ttl_size_, 0) != cudaSuccess) {
             DAQIRI_LOG_CRITICAL("Failed to allocate CUDA pinned host memory!");
             return Status::NULL_PTR;
           }
+          ar.deallocator_ = AllocRegion::Deallocator::CUDA_HOST;
           break;
         case MemoryKind::HUGE:
-          ptr = alloc_huge(mr.second.ttl_size_, mr.second.affinity_);
+          ptr = alloc_huge(mr.second.ttl_size_, mr.second.affinity_, &ar.deallocator_);
           break;
         case MemoryKind::DEVICE: {
           unsigned int flag = 1;
@@ -402,8 +464,10 @@ Status Engine::allocate_memory_regions() {
               cuPointerSetAttribute(&flag, CU_POINTER_ATTRIBUTE_SYNC_MEMOPS, cuptr);
           if (attr_res != CUDA_SUCCESS) {
             DAQIRI_LOG_CRITICAL("Could not set pointer attributes");
+            cuMemFree(cuptr);
             return Status::NULL_PTR;
           }
+          ar.deallocator_ = AllocRegion::Deallocator::CUDA_DEVICE;
           break;
         }
         default:
@@ -419,6 +483,8 @@ Status Engine::allocate_memory_regions() {
       }
     }
 
+    ar.ptr_ = ptr;
+
     DAQIRI_LOG_INFO(
         "Successfully allocated memory region {} at {} type {} with {} bytes "
         "({} elements @ {} bytes total {})",
@@ -429,7 +495,7 @@ Status Engine::allocate_memory_regions() {
         mr.second.num_bufs_,
         mr.second.adj_size_,
         mr.second.ttl_size_);
-    ar_[mr.second.name_] = {mr.second.name_, ptr};
+    ar_[mr.second.name_] = ar;
   }
   DAQIRI_LOG_INFO("Finished allocating memory regions");
   return Status::SUCCESS;

src/engine.h

@@ -33,8 +33,20 @@ struct rte_pktmbuf_extmem;
 namespace daqiri {
 
 struct AllocRegion {
+  enum class Deallocator {
+    NONE,
+    FREE,
+    CUDA_HOST,
+    CUDA_DEVICE,
+    MUNMAP,
+    EAL,
+  };
+
   std::string mr_name_;
-  void* ptr_;
+  void* ptr_ = nullptr;
+  size_t size_ = 0;
+  int affinity_ = -1;
+  Deallocator deallocator_ = Deallocator::NONE;
 };
 
 /**
@@ -137,7 +149,7 @@ class Engine {
   struct rte_mempool* create_pktmbuf_pool(const std::string& name, const MemoryRegionConfig& mr);
   struct rte_mempool* create_generic_pool(const std::string& name, const MemoryRegionConfig& mr);
 
-  virtual ~Engine() = default;
+  virtual ~Engine();
 
  protected:
   static constexpr int MAX_IFS = 4;
@@ -159,6 +171,9 @@ class Engine {
   // shared_ptr to an incomplete type -- only populated by the DPDK engine
   // (engine_dpdk.cpp). Layout is identical in every build.
   std::unordered_map<std::string, std::shared_ptr<struct rte_pktmbuf_extmem>> ext_pktmbufs_;
+  // rte_extmem_register_dmabuf() retains the caller's numeric fd for deferred
+  // driver mapping. Keep successful exports open until EAL teardown completes.
+  std::vector<int> ext_dmabuf_fds_;
   std::unordered_map<uint32_t, std::vector<std::pair<uint16_t, uint16_t>>> rx_core_q_map;
 
   // EAL lifecycle state, used only by the DPDK engine. Mirrors the
@@ -177,7 +192,8 @@ class Engine {
   // a libdpdk-free implementation (MAP_HUGETLB with a regular-page fallback);
   // DpdkEngine overrides it with rte_malloc_socket so HUGE regions come from EAL
   // hugepages that are IOVA-contiguous for the NIC.
-  virtual void* alloc_huge(size_t bytes, int numa);
+  virtual void* alloc_huge(size_t bytes, int numa, AllocRegion::Deallocator* deallocator);
+  void free_memory_regions() noexcept;
   void init_rx_core_q_map();
   static std::string generate_random_string(int len);
   size_t get_alignment(MemoryKind kind);
@@ -256,6 +272,11 @@ class EngineFactory {
     return *EngineInstance_;
   }
 
+  static void reset() {
+    EngineInstance_.reset();
+    EngineType_ = EngineType::UNKNOWN;
+  }
+
  private:
   EngineFactory() = default;
   ~EngineFactory() = default;

src/engine_dpdk.cpp

@@ -194,6 +194,16 @@ void Engine::cleanup_eal() {
   // behind, so we also do a best-effort unlink targeted at our --file-prefix.
   rte_eal_cleanup();
 
+  // The patched rte_extmem_register_dmabuf() stores caller-provided fd values
+  // for deferred driver DMA mapping; it does not duplicate them. They must
+  // remain valid through EAL teardown and can be closed only afterward.
+  for (const int fd : ext_dmabuf_fds_) {
+    if (fd >= 0) {
+      close(fd);
+    }
+  }
+  ext_dmabuf_fds_.clear();
+
   if (!eal_file_prefix_.empty()) {
     static const char* kHugepageMounts[] = {"/dev/hugepages", "/mnt/huge"};
     for (const char* mount : kHugepageMounts) {
@@ -317,6 +327,11 @@ Status Engine::register_memory_regions() {
 #if RTE_VERSION >= RTE_VERSION_NUM(24, 11, 0, 0)
           ret = rte_extmem_register_dmabuf(ext_mem->buf_ptr, ext_mem->buf_len, dmabuf_fd, offset,
                                            NULL, 0, GPU_PAGE_SIZE);
+          if (ret == 0) {
+            ext_dmabuf_fds_.push_back(dmabuf_fd);
+          } else {
+            close(dmabuf_fd);
+          }
 #else
           DAQIRI_LOG_WARN(
               "rte_extmem_register_dmabuf unavailable in DPDK {}; falling back to peermem "

src/engines/dpdk/daqiri_dpdk_engine.cpp

@@ -1721,6 +1721,7 @@ Status DpdkEngine::set_reorder_cuda_stream(const std::string& interface_name,
 ////////////////////////////////////////////////////////////////////////////////
 bool DpdkEngine::set_config_and_initialize(const NetworkConfig& cfg) {
   if (!this->initialized_) {
+    force_quit.store(false, std::memory_order_relaxed);
     cfg_ = cfg;
 
     if (!validate_config()) {
@@ -1775,8 +1776,12 @@ Status DpdkEngine::get_mac_addr(int port, char* mac) {
   return Status::SUCCESS;
 }
 
-void* DpdkEngine::alloc_huge(size_t bytes, int numa) {
+void* DpdkEngine::alloc_huge(size_t bytes, int numa, AllocRegion::Deallocator* deallocator) {
   // EAL hugepage allocation: IOVA-contiguous and registrable with the NIC.
+  if (deallocator == nullptr) {
+    return nullptr;
+  }
+  *deallocator = AllocRegion::Deallocator::EAL;
   return rte_malloc_socket(nullptr, bytes, 0, numa);
 }
 

src/engines/dpdk/daqiri_dpdk_engine.h

@@ -170,7 +170,7 @@ class DpdkEngine : public Engine {
   void print_stats() override;
   void adjust_memory_regions() override;
   // kind: HUGE regions come from EAL hugepages (IOVA-contiguous for the NIC).
-  void* alloc_huge(size_t bytes, int numa) override;
+  void* alloc_huge(size_t bytes, int numa, AllocRegion::Deallocator* deallocator) override;
   uint64_t get_burst_tot_byte(BurstParams* burst) override;
   BurstParams* create_tx_burst_params() override;
   bool validate_config() const override;

src/engines/dpdk/daqiri_dpdk_stats.cpp

@@ -23,6 +23,7 @@
 namespace daqiri {
 
 void DpdkStats::Init(const NetworkConfig &cfg) {
+  force_quit_.store(false, std::memory_order_relaxed);
   cfg_ = cfg;
   init_ = true;
   port_metrics_.clear();