thea/lcores_launch_at_conf

src/IfaceWrapper.cpp

@@ -54,6 +54,7 @@ IfaceWrapper::IfaceWrapper(uint16_t iface_id, source_to_sink_map_t& sources, std
     , m_mbuf_cache_size(0)
     , m_sources(sources)
     , m_run_marker(run_marker)
+    , m_push_marker{false} 
 { 
   m_iface_id_str = "iface-" + std::to_string(m_iface_id);
 }
@@ -246,48 +247,66 @@ IfaceWrapper::conf(const iface_conf_t& args)
     TLOG() << "Append TX_Q=0 for ARP responses.";
     m_tx_qs.insert(0);
 
-
   }
-}
 
-void
-IfaceWrapper::start()
-{
+  // disable pushing
+  m_push_marker.store(false);
+
+  // Reset counters
   for (auto const& [rx_q, _] : m_num_frames_rxq ) {
     m_num_frames_rxq[rx_q] = { 0 };
     m_num_bytes_rxq[rx_q] = { 0 };
     m_num_full_bursts[rx_q] = { 0 };
     m_max_burst_size[rx_q] = { 0 };
   }
 
-
+  // Launch GARP
   m_lcore_quit_signal.store(false);
   TLOG() << "Launching GARP thread with garp_func...";
   m_garp_thread = std::thread(&IfaceWrapper::garp_func, this);
 
-
+
+  // Launch LCores
   TLOG() << "Interface id=" << m_iface_id << " starting LCore processors:";
   for (auto const& [lcoreid, _] : m_rx_core_map) {
     int ret = rte_eal_remote_launch((int (*)(void*))(&IfaceWrapper::rx_runner), this, lcoreid);
     TLOG() << "  -> LCore[" << lcoreid << "] launched with return code=" << ret;
   }
 }
 
+void
+IfaceWrapper::start()
+{
+  // Start pushing
+  m_lcore_quit_signal.store(true);
+  for (auto const& [rx_q, _] : m_num_frames_rxq ) {
+    m_num_frames_rxq[rx_q] = { 0 };
+    m_num_bytes_rxq[rx_q] = { 0 };
+    m_num_full_bursts[rx_q] = { 0 };
+    m_max_burst_size[rx_q] = { 0 };
+  }
+}
+
 void
 IfaceWrapper::stop()
 {
+  // Stop Pushing
+  m_push_marker.store(false);
+}
+
+void
+IfaceWrapper::scrap()
+{
+  // Nuke lcores
   m_lcore_quit_signal.store(true);
+
   // Stop GARP sender thread  
   if (m_garp_thread.joinable()) {
     m_garp_thread.join();
   } else {
     TLOG() << "GARP thrad is not joinable!";
   }
-}
 
-void
-IfaceWrapper::scrap()
-{
   struct rte_flow_error error;
   rte_flow_flush(m_iface_id, &error);
 }

src/IfaceWrapper.hpp

@@ -113,6 +113,7 @@ class IfaceWrapper
 
   // Run marker
   std::atomic<bool>& m_run_marker;
+  std::atomic<bool> m_push_marker;
 
   // GARP
   std::unique_ptr<rte_mempool> m_garp_mbuf_pool;

src/detail/IfaceWrapper.hxx

@@ -79,10 +79,12 @@ IfaceWrapper::rx_runner(void *arg __rte_unused) {
           //if (pkt_type == RTE_PTYPE_L4_UDP) { // RS FIXME: doesn't work. Why? What is the PKT_TYPE in our ETH frames?
           // Check for JUMBO frames
           if (q_bufs[i_b]->pkt_len > 7000) { // RS FIXME: do proper check on data length later
-            // Handle them!
             std::size_t data_len = q_bufs[i_b]->data_len;
-            char* message = udp::get_udp_payload(q_bufs[i_b]);
-            handle_eth_payload(src_rx_q, message, data_len);
+            if (m_push_marker.load()) {
+              // Handle them!
+              char* message = udp::get_udp_payload(q_bufs[i_b]);
+              handle_eth_payload(src_rx_q, message, data_len);
+            }
             ++m_num_frames_rxq[src_rx_q];
             m_num_bytes_rxq[src_rx_q] += data_len;
           }