Clean up fast path

Signed-off-by: Hoooao <hao021014@163.com>

Author: Hoooao

configure.ac

@@ -63,6 +63,15 @@ AS_IF([test "x$enable_debugger" = "xyes"], [
     ])
 ])
 
+# TODO: add this to cmake build as well
+pkt_fanout_enabled=no
+AC_ARG_ENABLE([pkt_fanout],
+    AS_HELP_STRING([--enable-pkt-fanout], [Enable packet fanout support]))
+AS_IF([test "x$enable_pkt_fanout" = "xyes"], [
+    pkt_fanout_enabled=yes
+    AC_DEFINE([PKT_FANOUT_ON], [], [Enable packet fanout support])
+])
+
 logging_macros_enabled=no
 AC_ARG_ENABLE([logging_macros],
     AS_HELP_STRING([--disable-logging-macros],

include/bm/bm_sim/fanout_pkt_mgr.h

@@ -30,10 +30,13 @@ using SelectorIface = ActionProfile::GroupSelectionIface;
 
 struct FanoutCtx {
   bool hit{false};
-  std::vector<Packet *> fanout_pkts;
   const Packet * cur_pkt{nullptr};
   ActionProfile *action_profile{nullptr};
   MatchTableIndirect *cur_table{nullptr};
+  std::function<void(const bm::Packet *)> buffer_push_fn;
+
+  FanoutCtx(const std::function<void(const bm::Packet *)> &buffer_push_fn)
+      : buffer_push_fn(buffer_push_fn) { }
 };
 
 class FanoutPktSelection: public SelectorIface{
@@ -67,23 +70,26 @@ class FanoutPktMgr {
       return instance_;
     }
 
-    inline void register_thread(std::thread::id thread_id) {
+    inline void register_thread(std::thread::id thread_id,
+          const std::function<void(const bm::Packet *)> &buffer_push_fn) {
       BMLOG_DEBUG("Registering thread {}", thread_id);
-      fanout_ctx_map.emplace(thread_id, FanoutCtx());
+      fanout_ctx_map.emplace(thread_id, FanoutCtx(buffer_push_fn));
     }
     inline SelectorIface* get_grp_selector() {
       return grp_selector;
     }
 
-
-
-    std::vector<Packet *>& get_fanout_pkts();
     FanoutCtx& get_fanout_ctx();
     void set_ctx(MatchTableIndirect *table, const Packet &pkt,
                  ActionProfile *action_profile, bool hit);
     void reset_ctx();
     void replicate_for_entries(const std::vector<const ActionEntry*> &entries);
 
+#ifdef BM_PKT_FANOUT_ON
+    static constexpr bool pkt_fanout_on = true;
+#else
+    static constexpr bool pkt_fanout_on = false;
+#endif
     std::mutex fanout_pkt_mutex;
     // TODO(Hao): deduplicate packets fanout, optional
 

src/bm_sim/P4Objects.cpp

@@ -1657,15 +1657,7 @@ P4Objects::init_pipelines(const Json::Value &cfg_root,
       std::unique_ptr<ActionProfile> action_profile(
           new ActionProfile(act_prof_name, act_prof_id, with_selection));
       if (with_selection) {
-        // TODO(Hao): clean debug logs
-        BMLOG_DEBUG(
-            "Action profile '{}' with id {} has a selector",
-            act_prof_name, act_prof_id);
-
         if (is_selector_fanout(cfg_act_prof["selector"])) {
-          BMLOG_DEBUG(
-              "Action profile '{}' with id {} enabled selector fanout",
-              act_prof_name, act_prof_id);
           action_profile->set_selector_fanout();
         } else {
           auto calc = process_cfg_selector(cfg_act_prof["selector"]);
@@ -2514,8 +2506,13 @@ P4Objects::check_hash(const std::string &name) const {
 }
 
 bool P4Objects::is_selector_fanout(const Json::Value &cfg_selector) const {
-  return cfg_selector.isMember("algo") &&
+  bool is_fanout = cfg_selector.isMember("algo") &&
          cfg_selector["algo"].asString() == "selector_fanout";
+  if(is_fanout && !FanoutPktMgr::pkt_fanout_on){
+    throw std::runtime_error("Selector fanout is not enabled, but"
+                             " found selector_fanout mode used");
+  }
+  return is_fanout;
 }
 
 std::unique_ptr<Calculation>

src/bm_sim/action_profile.cpp

@@ -640,7 +640,7 @@ ActionProfile::mbr_hdl_t
 ActionProfile::choose_from_group(grp_hdl_t grp, const Packet &pkt) const {
   // TODO(Hao): PI resets to it own grp_selector, might be a bug, so I just sets
   //  here for now..
-  if (selector_fanout_enabled) {
+  if (FanoutPktMgr::pkt_fanout_on && selector_fanout_enabled) {
     return FanoutPktMgr::instance().get_grp_selector()->get_from_hash(grp, 0);
   }
   if (!hash) return grp_selector->get_from_hash(grp, 0);

src/bm_sim/fanout_pkt_mgr.cpp

@@ -53,20 +53,6 @@ FanoutPktSelection::get_from_hash(grp_hdl_t grp, hash_t h) const {
   return selected_mbr;
 }
 
-
-std::vector<Packet *>& FanoutPktMgr::get_fanout_pkts() {
-    std::thread::id thread_id = std::this_thread::get_id();
-    BMLOG_DEBUG("Getting fanout packets for thread {}", thread_id);
-
-    std::lock_guard<std::mutex> lock(fanout_pkt_mutex);
-    auto it = fanout_ctx_map.find(thread_id);
-    if (it == fanout_ctx_map.end()) {
-        BMLOG_ERROR("No fanout vector registered for thread {}", thread_id);
-        throw std::runtime_error("Fanout vector not found for thread");
-    }
-    return it->second.fanout_pkts;
-}
-
 FanoutCtx& FanoutPktMgr::get_fanout_ctx() {
     std::thread::id thread_id = std::this_thread::get_id();
     auto it = fanout_ctx_map.find(thread_id);
@@ -98,7 +84,6 @@ void FanoutPktMgr::reset_ctx() {
 
 void FanoutPktMgr::replicate_for_entries(
     const std::vector<const ActionEntry*> &entries) {
-    auto &fanout_pkts = get_fanout_pkts();
     auto &ctx = get_fanout_ctx();
     auto *match_table = ctx.cur_table;
     const Packet &pkt = *ctx.cur_pkt;
@@ -126,7 +111,7 @@ void FanoutPktMgr::replicate_for_entries(
                             act_id, next_node->get_name());
         }
         rep_pkt->set_next_node(next_node);
-        fanout_pkts.push_back(rep_pkt);
+        ctx.buffer_push_fn(rep_pkt);
 
         BMELOG(fanout_gen, *rep_pkt, table_id, parent_pkt_copy_id);
     }

src/bm_sim/match_tables.cpp

@@ -715,10 +715,14 @@ MatchTableIndirect::lookup(const Packet &pkt,
         pkt, "Lookup in table '{}' yielded empty group", get_name());
     return empty_action;
   }
-  // A bit hacky, in order to get the next_table for fanout.
-  if (action_profile->is_selector_fanout_enabled()) {
-    FanoutPktMgr::instance().set_ctx(this, pkt, action_profile, *hit);
+
+  if (FanoutPktMgr::pkt_fanout_on) {
+    // A bit hacky, in order to get the next_table for fanout.
+    if (action_profile->is_selector_fanout_enabled()) {
+      FanoutPktMgr::instance().set_ctx(this, pkt, action_profile, *hit);
+    }
   }
+  
   const auto &entry = action_profile->lookup(pkt, index);
   // Unfortunately this has to be done at this stage and cannot be done when
   // inserting a member because for 2 match tables sharing the same action

targets/simple_switch/simple_switch.cpp

@@ -278,23 +278,37 @@ SimpleSwitch::receive_(port_t port_num, const char *buffer, int len) {
 void
 SimpleSwitch::start_and_return_() {
   check_queueing_metadata();
-#ifdef PKT_FANOUT_ON
-  auto ingress_thread = std::thread(&SimpleSwitch::ingress_thread, this);
-  FanoutPktMgr::instance().register_thread(
-      ingress_thread.get_id());
-  threads_.push_back(std::move(ingress_thread));
-  for (size_t i = 0; i < nb_egress_threads; i++) {
-    auto egress_thread = std::thread(&SimpleSwitch::egress_thread, this, i);
+  if (FanoutPktMgr::pkt_fanout_on) {
+    auto ingress_thread = std::thread(&SimpleSwitch::ingress_thread, this);
     FanoutPktMgr::instance().register_thread(
-        egress_thread.get_id());
-    threads_.push_back(std::move(egress_thread));
-  }
-#else
-  threads_.push_back(std::thread(&SimpleSwitch::ingress_thread, this));
-  for (size_t i = 0; i < nb_egress_threads; i++) {
-    threads_.push_back(std::thread(&SimpleSwitch::egress_thread, this, i));
+        ingress_thread.get_id(), [&](const bm::Packet *pkt) {
+          this->input_buffer->push_front(
+              InputBuffer::PacketType::SELECTOR_FANOUT,
+              std::unique_ptr<bm::Packet>(const_cast<bm::Packet *>(pkt)));
+         BMLOG_DEBUG_PKT(*pkt,
+           "SELECTOR_FANOUT packet pushed to ingress_buffer");
+        });
+        
+    threads_.push_back(std::move(ingress_thread));
+    for (size_t i = 0; i < nb_egress_threads; i++) {
+      auto egress_thread = std::thread(&SimpleSwitch::egress_thread, this, i);
+
+      FanoutPktMgr::instance().register_thread(
+          egress_thread.get_id(), [&](const bm::Packet *pkt) {
+            this->egress_buffers.push_front(i, 0,
+              std::unique_ptr<bm::Packet>(const_cast<bm::Packet *>(pkt)));
+            BMLOG_DEBUG_PKT(*pkt, 
+              "SELECTOR_FANOUT packet pushed to egress_buffer");
+          });
+
+      threads_.push_back(std::move(egress_thread));
+    }
+  } else {
+    threads_.push_back(std::thread(&SimpleSwitch::ingress_thread, this));
+    for (size_t i = 0; i < nb_egress_threads; i++) {
+      threads_.push_back(std::thread(&SimpleSwitch::egress_thread, this, i));
+    }
   }
-#endif
   threads_.push_back(std::thread(&SimpleSwitch::transmit_thread, this));
 }
 
@@ -524,9 +538,9 @@ SimpleSwitch::ingress_thread() {
        deparser. TODO? */
     const Packet::buffer_state_t packet_in_state = packet->save_buffer_state();
 
-    // Check if the packet has an optional continue node
+    // Check if the packet has an optional continue node for pkt fanout
     // TODO(Hao): update the doc/simple_switch.md
-    if (packet->has_next_node()) {
+    if (FanoutPktMgr::pkt_fanout_on && packet->has_next_node()) {
       ingress_mau->apply_from_next_node(packet.get());
     } else {
       parser->parse(packet.get());
@@ -542,20 +556,6 @@ SimpleSwitch::ingress_thread() {
 
       ingress_mau->apply(packet.get());
     }
-
-#ifdef PKT_FANOUT_ON
-    {
-      auto &fanout_pkts = FanoutPktMgr::instance().get_fanout_pkts();
-      for (auto pkt : fanout_pkts) {
-        input_buffer->push_front(InputBuffer::PacketType::SELECTOR_FANOUT,
-          std::unique_ptr<bm::Packet>(pkt));
-        BMLOG_DEBUG_PKT(*pkt,
-          "SELECTOR_FANOUT packet pushed to ingress_buffer");
-      }
-      fanout_pkts.clear();
-    }
-#endif
-
     packet->reset_exit();
 
     Field &f_egress_spec = phv->get_field("standard_metadata.egress_spec");
@@ -694,7 +694,8 @@ SimpleSwitch::egress_thread(size_t worker_id) {
     Pipeline *egress_mau = this->get_pipeline("egress");
 
     phv = packet->get_phv();
-
+    Field &f_egress_spec = phv->get_field("standard_metadata.egress_spec");
+    
     if (packet->has_next_node()) {
       egress_mau->apply_from_next_node(packet.get());
     } else {
@@ -717,7 +718,6 @@ SimpleSwitch::egress_thread(size_t worker_id) {
       }
 
       phv->get_field("standard_metadata.egress_port").set(port);
-      Field &f_egress_spec = phv->get_field("standard_metadata.egress_spec");
       // When egress_spec == drop_port the packet will be dropped, thus
       // here we initialize egress_spec to a value different from drop_port.
       f_egress_spec.set(drop_port + 1);
@@ -728,17 +728,6 @@ SimpleSwitch::egress_thread(size_t worker_id) {
       egress_mau->apply(packet.get());
     }
 
-#ifdef PKT_FANOUT_ON
-    {
-      auto &fanout_pkts = FanoutPktMgr::instance().get_fanout_pkts();
-      for (auto pkt : fanout_pkts) {
-        egress_buffers.push_front(worker_id, 0, std::unique_ptr<Packet>(pkt));
-        BMLOG_DEBUG_PKT(*pkt, "SELECTOR_FANOUT packet pushed to egress_buffer");
-      }
-      fanout_pkts.clear();
-    }
-#endif
-
     auto clone_mirror_session_id =
         RegisterAccess::get_clone_mirror_session_id(packet.get());
     auto clone_field_list = RegisterAccess::get_clone_field_list(packet.get());