[WIP] Obtain all possible outputs generated by ActionSelector group members with single input packet

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

@@ -115,7 +115,7 @@ class ActionProfile : public NamedP4Object {
   };
 
   ActionProfile(const std::string &name, p4object_id_t id, bool with_selection);
-
+  
   bool has_selection() const;
 
   void set_hash(std::unique_ptr<Calculation> h) {
@@ -165,6 +165,11 @@ 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(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 +296,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 +316,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,98 @@
+/* Copyright 2013-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 "logger.h"
+#include "packet.h"
+#include "match_tables.h"
+#include "action_profile.h"
+#include <vector>
+
+namespace bm {
+class MatchTableIndirect;
+using bm::ActionProfile;
+using EntryVec = const std::vector<const ActionEntry*>;
+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};
+};
+
+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_;
+    }
+
+    inline void register_thread(std::thread::id thread_id) {
+      BMLOG_DEBUG("Registering thread {}", thread_id);
+      fanout_ctx_map.emplace(thread_id, FanoutCtx());
+    }
+    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);
+
+    std::mutex fanout_pkt_mutex;
+    // TODO(Hao): deduplicate packets fanout, optional
+
+
+private:
+    FanoutPktMgr() = default;
+    std::unordered_map<std::thread::id, FanoutCtx> fanout_ctx_map; 
+    FanoutPktSelection fanout_selection;
+    SelectorIface* grp_selector{&fanout_selection};
+};
+
+} // 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,7 @@ 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,20 @@ 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));
+        // 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"]);
+          action_profile->set_hash(std::move(calc));
+        }
       }
       add_action_profile(act_prof_name, std::move(action_profile));
     }
@@ -2499,6 +2513,11 @@ P4Objects::check_hash(const std::string &name) const {
   return nullptr;
 }
 
+bool P4Objects::is_selector_fanout(const Json::Value &cfg_selector) const {
+  return cfg_selector.isMember("algo") && // not to hardcode names, fix later
+         cfg_selector["algo"].asString() == "selector_fanout";
+}
+
 std::unique_ptr<Calculation>
 P4Objects::process_cfg_selector(const Json::Value &cfg_selector) const {
   const auto selector_algo = cfg_selector["algo"].asString();

src/bm_sim/action_profile.cpp

@@ -22,12 +22,14 @@
 #include <bm/bm_sim/action_profile.h>
 #include <bm/bm_sim/logger.h>
 #include <bm/bm_sim/packet.h>
+#include <bm/bm_sim/fanout_pkt_mgr.h>
 
 #include <iostream>
 #include <memory>
 #include <string>
 #include <vector>
 
+#include <boost/stacktrace.hpp> // remove 
 namespace bm {
 
 void
@@ -126,6 +128,8 @@ ActionProfile::mbr_hdl_t
 ActionProfile::GroupMgr::get_from_hash(grp_hdl_t grp, hash_t h) const {
   const auto &group_info = groups.at(grp);
   auto s = group_info.size();
+  BMLOG_DEBUG("Choosing member from group {} with hash {} (size {})",
+                grp, h, s);
   return group_info.get_nth(h % s);
 }
 
@@ -184,6 +188,28 @@ ActionProfile::lookup(const Packet &pkt, const IndirectIndex &index) const {
   return action_entries[mbr];
 }
 
+
+std::vector<ActionProfile::mbr_hdl_t>
+ActionProfile::get_all_mbrs_from_grp(grp_hdl_t grp) const {
+  assert(is_valid_grp(grp));
+  Group group;
+  MatchErrorCode rc = get_group(grp, &group);
+  _BM_UNUSED(rc);
+  assert(rc == MatchErrorCode::SUCCESS);
+  return group.mbr_handles;
+}
+
+std::vector<const ActionEntry*>
+ActionProfile::get_entries_with_mbrs(const std::vector<ActionProfile::mbr_hdl_t> &mbrs) const {
+  std::vector<const ActionEntry*> entries;
+  entries.reserve(mbrs.size());
+  for (auto m : mbrs) {
+    assert(is_valid_mbr(m));
+    entries.push_back(&action_entries[m]);
+  }
+  return entries;
+}
+
 bool
 ActionProfile::has_selection() const { return with_selection; }
 
@@ -524,6 +550,10 @@ void
 ActionProfile::set_group_selector(
     std::shared_ptr<GroupSelectionIface> selector) {
   WriteLock lock = lock_write();
+  BMLOG_DEBUG("Setting group selector for action profile '{}'",
+                get_name());
+  // output stack trace, remove later
+  BMLOG_DEBUG("Stack trace: {}", boost::stacktrace::stacktrace());
   grp_selector_ = selector;
   grp_selector = grp_selector_.get();
 }
@@ -612,9 +642,18 @@ ActionProfile::ref_count_decrease(const IndirectIndex &index) {
 
 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) {
+    return FanoutPktMgr::instance().get_grp_selector()->get_from_hash(grp, 0);
+  }
   if (!hash) return grp_selector->get_from_hash(grp, 0);
   hash_t h = static_cast<hash_t>(hash->output(pkt));
   return grp_selector->get_from_hash(grp, h);
 }
 
+void ActionProfile::set_selector_fanout(){
+  selector_fanout_enabled = true;
+}
+
 }  // namespace bm