Selector Packet Fanout - Core implementation

CMakeLists.txt

@@ -51,6 +51,7 @@ option(ENABLE_MODULES "Allow loading third-party modules at runtime" ON)
 option(REQUIRE_MODULES "Require support for loading third-party modules at runtime" OFF)
 option(ENABLE_UNDETERMINISTIC_TESTS "Run undeterministic tests (e.g. queueing) when running tests" ON)
 option(ENABLE_WERROR "Make all compiler warnings fatal" OFF)
+option(ENABLE_PKT_FANOUT "Enable action selector packet fanout support" OFF)
 option(ENABLE_WP4_16_STACKS "Implement stacks strictly as per the P4_16 specification" ON)
 
 # Set compiler flags
@@ -144,6 +145,10 @@ if(ENABLE_LOGGING_MACROS)
   add_definitions(-DLOG_DEBUG_ON -DLOG_TRACE_ON)
 endif()
 
+if(ENABLE_PKT_FANOUT)
+  add_definitions(-DPKT_FANOUT_ON)
+endif()
+
 if(ENABLE_WP4_16_STACKS)
   add_definitions(-DWP4_16_STACKS)
 endif()

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/P4Objects.h

@@ -545,6 +545,8 @@ class P4Objects {
   void enable_arith(header_id_t header_id, int field_offset);
   void enable_arith(header_id_t header_id);
 
+  bool is_selector_fanout(
+      const Json::Value &cfg_next_nodes) const;
   std::unique_ptr<Calculation> process_cfg_selector(
       const Json::Value &cfg_selector) const;
 };

include/bm/bm_sim/action_profile.h

@@ -165,6 +165,12 @@ class ActionProfile : public NamedP4Object {
   void serialize(std::ostream *out) const;
   void deserialize(std::istream *in, const P4Objects &objs);
 
+  void set_selector_fanout();
+
+  std::vector<mbr_hdl_t> get_all_mbrs_from_grp(const grp_hdl_t &grp) const;
+  std::vector<const ActionEntry*>
+  get_entries_with_mbrs(const std::vector<mbr_hdl_t> &mbrs) const;
+
  private:
   using ReadLock = boost::shared_lock<boost::shared_mutex>;
   using WriteLock = boost::unique_lock<boost::shared_mutex>;
@@ -291,6 +297,10 @@ class ActionProfile : public NamedP4Object {
 
   bool group_is_empty(grp_hdl_t grp) const;
 
+  bool is_selector_fanout_enabled() const {
+    return selector_fanout_enabled;
+  }
+
   const ActionEntry &lookup(const Packet &pkt,
                             const IndirectIndex &index) const;
 
@@ -307,6 +317,7 @@ class ActionProfile : public NamedP4Object {
   std::shared_ptr<GroupSelectionIface> grp_selector_{nullptr};
   GroupSelectionIface *grp_selector{&grp_mgr};
   std::unique_ptr<Calculation> hash{nullptr};
+  bool selector_fanout_enabled{false};
 };
 
 }  // namespace bm

include/bm/bm_sim/event_logger.h

@@ -99,7 +99,8 @@ class EventLogger {
 
   void action_execute(const Packet &packet,
                       const ActionFn &action_fn, const ActionData &action_data);
-
+  void fanout_gen(const Packet &packet, uint64_t table_id,
+                  uint64_t parent_pkt_copy_id);
   void config_change();
 
   static EventLogger *get() {

include/bm/bm_sim/fanout_pkt_mgr.h

@@ -0,0 +1,112 @@
+/* Copyright 2025-present Contributors to the P4 Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef BM_BM_SIM_FANOUT_PKT_MGR_H_
+#define BM_BM_SIM_FANOUT_PKT_MGR_H_
+
+#include <memory>
+#include <unordered_map>
+#include <vector>
+#include "logger.h"
+#include "packet.h"
+#include "match_tables.h"
+#include "action_profile.h"
+
+namespace bm {
+class MatchTableIndirect;
+using bm::ActionProfile;
+using EntryVec = const std::vector<const ActionEntry*>;
+using SelectorIface = ActionProfile::GroupSelectionIface;
+
+struct FanoutCtx {
+  bool hit{false};
+  const Packet * cur_pkt{nullptr};
+  ActionProfile *action_profile{nullptr};
+  MatchTableIndirect *cur_table{nullptr};
+  std::function<void(const bm::Packet *)> buffer_push_fn;
+
+  explicit FanoutCtx(
+    const std::function<void(const bm::Packet *)> &buffer_push_fn)
+      : buffer_push_fn(buffer_push_fn) { }
+};
+
+class FanoutPktSelection: public SelectorIface{
+ public:
+    using grp_hdl_t = ActionProfile::grp_hdl_t;
+    using mbr_hdl_t = ActionProfile::mbr_hdl_t;
+    using hash_t = ActionProfile::hash_t;
+    using MatchErrorCode = bm::MatchErrorCode;
+
+    FanoutPktSelection() = default;
+
+    // callbacks after member op, not actual member/group ops
+    void add_member_to_group(grp_hdl_t grp, mbr_hdl_t mbr) override;
+
+    void remove_member_from_group(grp_hdl_t grp, mbr_hdl_t mbr) override;
+
+    mbr_hdl_t get_from_hash(grp_hdl_t grp, hash_t h) const override;
+
+    void reset() override {}
+
+ private:
+    std::unordered_map<grp_hdl_t, std::vector<mbr_hdl_t>> groups;
+};
+
+class FanoutPktMgr {
+ public:
+    FanoutPktMgr(const FanoutPktMgr&) = delete;
+    FanoutPktMgr& operator=(const FanoutPktMgr&) = delete;
+    static FanoutPktMgr& instance() {
+      static FanoutPktMgr instance_;
+      return instance_;
+    }
+
+    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);
+
+    // PI overwrite selector specified during P4Object init,
+    // so we need to set the selector in switch start_and_return_
+    void set_grp_selector() {
+      for (const auto &ap : act_profs) {
+        ap->set_group_selector(grp_selector);
+      }
+    }
+    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(buffer_push_fn));
+    }
+
+    // TODO(Hao): deduplicate packets fanout, optional
+#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;
+    std::vector<ActionProfile*> act_profs;
+
+ private:
+    FanoutPktMgr() = default;
+    std::unordered_map<std::thread::id, FanoutCtx> fanout_ctx_map;
+    std::shared_ptr<SelectorIface>
+      grp_selector{std::make_shared<FanoutPktSelection>()};
+};
+
+}  // namespace bm
+
+#endif  // BM_BM_SIM_FANOUT_PKT_MGR_H_

include/bm/bm_sim/logger.h

@@ -155,4 +155,10 @@ class Logger {
   bm::Logger::get()->error("[{}] [cxt {}] " s, (pkt).get_unique_id(),   \
                            (pkt).get_context(), ##__VA_ARGS__)
 
+#define BMLOG_WARN(...) bm::Logger::get()->warn(__VA_ARGS__)
+
+#define BMLOG_WARN_PKT(pkt, s, ...)                                    \
+  bm::Logger::get()->warn("[{}] [cxt {}] " s, (pkt).get_unique_id(),   \
+                           (pkt).get_context(), ##__VA_ARGS__)
+
 #endif  // BM_BM_SIM_LOGGER_H_

include/bm/bm_sim/match_tables.h

@@ -39,6 +39,7 @@
 #include "lookup_structures.h"
 #include "action_entry.h"
 #include "action_profile.h"
+#include "fanout_pkt_mgr.h"
 
 namespace bm {
 
@@ -385,6 +386,7 @@ class MatchTable : public MatchTableAbstract {
 
 class MatchTableIndirect : public MatchTableAbstract {
  public:
+  friend class FanoutPktMgr;
   using mbr_hdl_t = ActionProfile::mbr_hdl_t;
 
   using IndirectIndex = ActionProfile::IndirectIndex;

include/bm/bm_sim/packet.h

@@ -32,6 +32,7 @@
 #include <atomic>
 #include <algorithm>  // for std::min
 #include <limits>
+#include <optional>
 
 #include <cassert>
 
@@ -40,6 +41,7 @@
 #include "parser_error.h"
 #include "phv_source.h"
 #include "phv_forward.h"
+#include "control_flow.h"
 
 namespace bm {
 
@@ -301,6 +303,11 @@ class Packet final {
   //! @copydoc clone_with_phv_reset_metadata
   std::unique_ptr<Packet> clone_with_phv_reset_metadata_ptr() const;
 
+  //! Clone the current packet, along with its PHV and registers.
+  Packet clone_with_phv_and_registers() const;
+  //! @copydoc clone_with_phv_and_registers
+  std::unique_ptr<Packet> clone_with_phv_and_registers_ptr() const;
+
   //! Clone the current packet, without the PHV. The value of the fields in the
   //! clone will be undefined and should not be accessed before setting it
   //! first.
@@ -315,6 +322,17 @@ class Packet final {
   //! @copydoc clone_choose_context
   std::unique_ptr<Packet> clone_choose_context_ptr(cxt_id_t new_cxt) const;
 
+  // Packet fanout related methods
+  //! Returns true if the packet has a next node set
+  bool has_next_node() const;
+  //! Get the next node, if it exists
+  const ControlFlowNode *get_next_node() const;
+  //! Set the next node, which is used to next the packet processing
+  void set_next_node(const ControlFlowNode *node);
+  //! Reset the next node
+  void reset_next_node();
+
+
   //! Deleted copy constructor
   Packet(const Packet &other) = delete;
   //! Deleted copy assignment operator
@@ -385,6 +403,8 @@ class Packet final {
 
   bool checksum_error{false};
 
+  std::optional<const ControlFlowNode *> next_node{std::nullopt};
+
  private:
   static CopyIdGenerator *copy_id_gen;
 };

include/bm/bm_sim/pipeline.h

@@ -40,10 +40,12 @@ class Pipeline : public NamedP4Object {
     : NamedP4Object(name, id), first_node(first_node) {}
 
   //! Sends the \p pkt through the correct match-action tables and
-  //! condiitons. Each step is determined based on the result of the previous
+  //! conditions. Each step is determined based on the result of the previous
   //! step (table lookup or condition evaluation), according to the P4 control
   //! flow graph.
   void apply(Packet *pkt);
+  // Start from next_node instead of first_node
+  void apply_from_next_node(Packet *pkt);
 
   //! Deleted copy constructor
   Pipeline(const Pipeline &other) = delete;

src/bm_sim/CMakeLists.txt

@@ -28,6 +28,7 @@ add_library(bmsim OBJECT
   event_logger.cpp
   expressions.cpp
   extern.cpp
+  fanout_pkt_mgr.cpp
   fields.cpp
   headers.cpp
   header_unions.cpp

src/bm_sim/Makefile.am

@@ -37,6 +37,7 @@ event_logger.cpp \
 expressions.cpp \
 extern.cpp \
 extract.h \
+fanout_pkt_mgr.cpp \
 fields.cpp \
 headers.cpp \
 header_unions.cpp \

src/bm_sim/P4Objects.cpp

@@ -22,6 +22,8 @@
 #include <bm/bm_sim/P4Objects.h>
 #include <bm/bm_sim/phv.h>
 #include <bm/bm_sim/actions.h>
+#include <bm/bm_sim/logger.h>
+#include <bm/bm_sim/fanout_pkt_mgr.h>
 
 #include <iostream>
 #include <istream>
@@ -1655,8 +1657,13 @@ 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) {
-        auto calc = process_cfg_selector(cfg_act_prof["selector"]);
-        action_profile->set_hash(std::move(calc));
+        if (is_selector_fanout(cfg_act_prof["selector"])) {
+          action_profile->set_selector_fanout();
+          FanoutPktMgr::instance().act_profs.push_back(action_profile.get());
+        } else {
+          auto calc = process_cfg_selector(cfg_act_prof["selector"]);
+          action_profile->set_hash(std::move(calc));
+        }
       }
       add_action_profile(act_prof_name, std::move(action_profile));
     }
@@ -2499,6 +2506,16 @@ P4Objects::check_hash(const std::string &name) const {
   return nullptr;
 }
 
+bool P4Objects::is_selector_fanout(const Json::Value &cfg_selector) const {
+  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>
 P4Objects::process_cfg_selector(const Json::Value &cfg_selector) const {
   const auto selector_algo = cfg_selector["algo"].asString();