diff --git a/core/modules/flow_measure.cc b/core/modules/flow_measure.cc index 71dc9569..1f9bc9bd 100644 --- a/core/modules/flow_measure.cc +++ b/core/modules/flow_measure.cc @@ -5,8 +5,8 @@ #include "flow_measure.h" -#include -#include +#include +#include #include "../core/utils/common.h" @@ -31,6 +31,7 @@ CommandResponse FlowMeasure::Init(const bess::pb::FlowMeasureArg &arg) { leader_ = false; buffer_flag_attr_id_ = AddMetadataAttr(arg.flag_attr_name(), sizeof(uint64_t), AccessMode::kRead); + current_flag_value_ = Flag::FLAG_VALUE_INVALID; } if (buffer_flag_attr_id_ < 0) return CommandFailure(EINVAL, "invalid flag attribute name"); @@ -46,40 +47,11 @@ CommandResponse FlowMeasure::Init(const bess::pb::FlowMeasureArg &arg) { if (pdr_attr_id_ < 0) return CommandFailure(EINVAL, "invalid metadata declaration"); - rte_hash_parameters hash_params = {}; - hash_params.entries = kDefaultNumEntries; - hash_params.key_len = sizeof(TableKey); - hash_params.hash_func = rte_jhash; - hash_params.socket_id = static_cast(rte_socket_id()); - hash_params.extra_flag = RTE_HASH_EXTRA_FLAGS_RW_CONCURRENCY; - if (arg.entries()) { - hash_params.entries = arg.entries(); - } - // Create both hash tables. - std::string name_a = name() + "Ta" + std::to_string(hash_params.socket_id); - if (name_a.length() > 26 /*RTE_HASH_NAMESIZE - 1*/) { - return CommandFailure(EINVAL, "invalid hash name A"); - } - hash_params.name = name_a.c_str(); - table_a_ = rte_hash_create(&hash_params); - if (!table_a_) { - return CommandFailure(rte_errno, "could not create hashmap A"); - } - std::string name_b = name() + "Tb" + std::to_string(hash_params.socket_id); - if (name_b.length() > 26 /*RTE_HASH_NAMESIZE - 1*/) { - return CommandFailure(EINVAL, "invalid hash name B"); - } - hash_params.name = name_b.c_str(); - table_b_ = rte_hash_create(&hash_params); - if (!table_b_) { - return CommandFailure(rte_errno, "could not create hashmap B"); - } + // Initialized dynamic buffers using smart pointers to replace manual DPDK + // hash creation. + buf_a_ = std::make_unique(); + buf_b_ = std::make_unique(); - // resize() would require a copyable object. - std::vector tmp_a(hash_params.entries); - std::vector tmp_b(hash_params.entries); - table_data_a_.swap(tmp_a); - table_data_b_.swap(tmp_b); VLOG(1) << name() << ": Tables created successfully."; return CommandSuccess(); @@ -95,66 +67,57 @@ void FlowMeasure::ProcessBatch(Context *ctx, bess::PacketBatch *batch) { static_cast(current_flag_value_)); cached_current_flag = current_flag_value_; } else { - const std::lock_guard lock(flag_mutex_); uint64_t flag = get_attr(this, buffer_flag_attr_id_, batch->pkts()[i]); if (!Flag_IsValid(flag)) { - LOG_EVERY_N(WARNING, 100'001) << "Encountered invalid flag: " << flag; + LOG_EVERY_N(WARNING, 100'001) + << name() << ": encountered invalid flag: " << flag; continue; - } else { - current_flag_value_ = static_cast(flag); - cached_current_flag = current_flag_value_; } + const std::lock_guard lock(flag_mutex_); + current_flag_value_ = static_cast(flag); + cached_current_flag = current_flag_value_; } uint64_t ts_ns = get_attr(this, ts_attr_id_, batch->pkts()[i]); uint64_t fseid = get_attr(this, fseid_attr_id_, batch->pkts()[i]); uint32_t pdr = get_attr(this, pdr_attr_id_, batch->pkts()[i]); - // Discard invalid timestamps. - if (!ts_ns || now_ns < ts_ns) { - continue; + + // Added fallback to current time for missing timestamps to ensure + // packet/byte counts remain accurate. + if (ts_ns == 0 || now_ns < ts_ns) { + ts_ns = now_ns; } - // Pick current side. - rte_hash *current_hash = nullptr; - std::vector *current_data = nullptr; + + Buffer *buf = nullptr; switch (cached_current_flag) { case Flag::FLAG_VALUE_A: - current_hash = table_a_; - current_data = &table_data_a_; + buf = buf_a_.get(); break; case Flag::FLAG_VALUE_B: - current_hash = table_b_; - current_data = &table_data_b_; + buf = buf_b_.get(); break; default: LOG_EVERY_N(ERROR, 100'001) - << "Unknown flag value: " << Flag_Name(cached_current_flag) << "."; + << name() + << ": unknown flag value: " << Flag_Name(cached_current_flag); continue; } // Find or create session. TableKey key(fseid, pdr); - int32_t ret = rte_hash_lookup(current_hash, &key); - if (ret == -ENOENT) { - ret = rte_hash_add_key(current_hash, &key); - } - if (ret < 0) { - LOG(ERROR) << "Failed to lookup or insert session stats for key " - << key.ToString() << ": " << ret << ", " << rte_strerror(-ret); - continue; - } - // Update stats. - SessionStats &stat = current_data->at(ret); - const std::lock_guard lock(stat.mutex); + // Replaced index-based lookup with a thread-safe get_or_create pattern for + // dynamic flow tracking. + SessionStats *stat = buf->get_or_create(key); + const std::lock_guard lock(stat->mutex); uint64_t diff_ns = now_ns - ts_ns; - if (stat.last_latency == 0) { - stat.last_latency = diff_ns; - } - uint64_t jitter_ns = absdiff(stat.last_latency, diff_ns); - stat.last_latency = diff_ns; - stat.latency_histogram.Insert(diff_ns); - stat.jitter_histogram.Insert(jitter_ns); - stat.pkt_count += 1; - stat.byte_count += batch->pkts()[i]->total_len(); + if (stat->last_latency == 0) + stat->last_latency = diff_ns; + uint64_t jitter_ns = absdiff(stat->last_latency, diff_ns); + stat->last_latency = diff_ns; + stat->latency_histogram.Insert(diff_ns); + stat->jitter_histogram.Insert(jitter_ns); + stat->pkt_count += 1; + stat->byte_count += batch->pkts()[i]->total_len(); } RunNextModule(ctx, batch); @@ -163,7 +126,7 @@ void FlowMeasure::ProcessBatch(Context *ctx, bess::PacketBatch *batch) { CommandResponse FlowMeasure::CommandReadStats( const bess::pb::FlowMeasureCommandReadArg &arg) { Flag flag_to_read = static_cast(arg.flag_to_read()); - if (!Flag_IsValid(flag_to_read)) { + if (!Flag_IsValid(flag_to_read) && flag_to_read != Flag::FLAG_VALUE_INVALID) { return CommandFailure(EINVAL, "invalid flag value"); } // Cache current flag so we don't block the dataplane while reading the stats. @@ -175,75 +138,68 @@ CommandResponse FlowMeasure::CommandReadStats( VLOG(1) << name() << ": " << (leader_ ? "leader" : "follower") << " last saw buffer flag " << Flag_Name(cached_current_flag) << ", now reading from " << Flag_Name(flag_to_read); - VLOG_IF(1, cached_current_flag == flag_to_read) - << name() - << ": requested to read active buffer flag. Either there is no " - "traffic or the controller is performing invalid requests."; + VLOG_IF(1, cached_current_flag == flag_to_read && + flag_to_read != Flag::FLAG_VALUE_INVALID) + << name() << ": reading from active buffer — either no traffic yet" + << " or controller is performing invalid requests."; + + if (flag_to_read == Flag::FLAG_VALUE_INVALID) { + return CommandSuccess(bess::pb::FlowMeasureReadResponse{}); + } + + Buffer *buf = + (flag_to_read == Flag::FLAG_VALUE_A) ? buf_a_.get() : buf_b_.get(); + if (!buf) + return CommandFailure(EINVAL, "buffer not initialized"); bess::pb::FlowMeasureReadResponse resp; auto t_start = std::chrono::high_resolution_clock::now(); - rte_hash *current_hash = nullptr; - std::vector *current_data = nullptr; - switch (flag_to_read) { - case Flag::FLAG_VALUE_INVALID: - return CommandSuccess(resp); // return empty stats when no traffic - case Flag::FLAG_VALUE_A: - current_hash = table_a_; - current_data = &table_data_a_; - break; - case Flag::FLAG_VALUE_B: - current_hash = table_b_; - current_data = &table_data_b_; - break; - default: - return CommandFailure(EINVAL, "invalid flag value"); - } - const void *key = nullptr; - void *data = nullptr; - uint32_t next = 0; - int32_t ret = 0; - while (ret = rte_hash_iterate(current_hash, &key, &data, &next), ret >= 0) { - const TableKey *table_key = reinterpret_cast(key); - const SessionStats &session_stat = current_data->at(ret); - const std::lock_guard lock(session_stat.mutex); - const std::vector lat_percs(arg.latency_percentiles().begin(), - arg.latency_percentiles().end()); - const std::vector jitter_percs(arg.jitter_percentiles().begin(), - arg.jitter_percentiles().end()); - const auto lat_summary = - session_stat.latency_histogram.Summarize(lat_percs); - const auto jitter_summary = - session_stat.jitter_histogram.Summarize(jitter_percs); - bess::pb::FlowMeasureReadResponse::Statistic stat; - stat.set_fseid(table_key->fseid); - stat.set_pdr(table_key->pdr); - for (const auto &lat_perc : lat_summary.percentile_values) { - stat.mutable_latency()->add_percentile_values_ns(lat_perc); - } - for (const auto &jitter_perc : jitter_summary.percentile_values) { - stat.mutable_jitter()->add_percentile_values_ns(jitter_perc); + + const std::vector lat_percs(arg.latency_percentiles().begin(), + arg.latency_percentiles().end()); + const std::vector jitter_percs(arg.jitter_percentiles().begin(), + arg.jitter_percentiles().end()); + + { + // Used shared_lock to allow concurrent dataplane lookups while the control + // plane iterates through the map. + std::shared_lock map_lk(buf->map_mutex); + + for (auto &[key, stat_ptr] : buf->map) { + if (!stat_ptr) + continue; + + const std::lock_guard stat_lk(stat_ptr->mutex); + if (stat_ptr->pkt_count == 0) + continue; + + const auto lat_summary = stat_ptr->latency_histogram.Summarize(lat_percs); + const auto jitter_summary = + stat_ptr->jitter_histogram.Summarize(jitter_percs); + bess::pb::FlowMeasureReadResponse::Statistic stat; + stat.set_fseid(key.fseid); + stat.set_pdr(key.pdr); + for (const auto &v : lat_summary.percentile_values) + stat.mutable_latency()->add_percentile_values_ns(v); + for (const auto &v : jitter_summary.percentile_values) + stat.mutable_jitter()->add_percentile_values_ns(v); + stat.set_total_packets(stat_ptr->pkt_count); + stat.set_total_bytes(stat_ptr->byte_count); + *resp.add_statistics() = stat; } - stat.set_total_packets(session_stat.pkt_count); - stat.set_total_bytes(session_stat.byte_count); - *resp.add_statistics() = stat; - } + } // shared_lock released before clear if (arg.clear()) { - VLOG(1) << name() << ": starting hash table clear..."; - rte_hash_reset(current_hash); - // TODO: this is quite slow - VLOG(1) << name() << ": hash table clear done, clearing table data..."; - for (auto &stat : *current_data) { - const std::lock_guard lock(stat.mutex); - stat.reset(); - } - VLOG(1) << name() << ": table data clear done."; + // clear() acquires unique_lock internally. + // Safe: ProcessBatch is writing to the other buffer after the flag flip. + buf->clear(); } - auto t_done = std::chrono::high_resolution_clock::now(); if (VLOG_IS_ON(1)) { - std::chrono::duration diff = t_done - t_start; - VLOG(1) << name() << ": CommandReadStats took " << diff.count() << "s."; + auto elapsed = std::chrono::duration( + std::chrono::high_resolution_clock::now() - t_start) + .count(); + VLOG(1) << name() << ": CommandReadStats took " << elapsed << "s."; } return CommandSuccess(resp); @@ -252,15 +208,14 @@ CommandResponse FlowMeasure::CommandReadStats( CommandResponse FlowMeasure::CommandFlipFlag( const bess::pb::FlowMeasureCommandFlipArg &) { // Can only flip the flag if leader module. - if (!leader_) { + if (!leader_) return CommandFailure(EINVAL, "only leaders can flip the flag"); - } // Cache flags so we block the dataplane as short as possible. Flag cached_old_flag, cached_current_flag; { const std::lock_guard lock(flag_mutex_); cached_old_flag = current_flag_value_; - current_flag_value_ = current_flag_value_ == Flag::FLAG_VALUE_A + current_flag_value_ = (current_flag_value_ == Flag::FLAG_VALUE_A) ? Flag::FLAG_VALUE_B : Flag::FLAG_VALUE_A; cached_current_flag = current_flag_value_; @@ -276,8 +231,8 @@ CommandResponse FlowMeasure::CommandFlipFlag( } void FlowMeasure::DeInit() { - rte_hash_free(table_a_); - rte_hash_free(table_b_); + buf_a_.reset(); + buf_b_.reset(); } /*----------------------------------------------------------------------------------*/ diff --git a/core/modules/flow_measure.h b/core/modules/flow_measure.h index 374d2892..51ed93d8 100644 --- a/core/modules/flow_measure.h +++ b/core/modules/flow_measure.h @@ -6,9 +6,12 @@ #ifndef BESS_MODULES_QOS_MEASURE_H_ #define BESS_MODULES_QOS_MEASURE_H_ -#include - +#include #include +#include +#include +#include +#include #include "../core/utils/histogram.h" #include "../module.h" @@ -17,43 +20,43 @@ class FlowMeasure final : public Module { public: FlowMeasure() : leader_(false), - current_flag_value_(), - table_a_(nullptr), - table_b_(nullptr), + current_flag_value_(Flag::FLAG_VALUE_INVALID), + buf_a_(nullptr), + buf_b_(nullptr), ts_attr_id_(-1), fseid_attr_id_(-1), - pdr_attr_id_(-1) { - // Multi-writer support is not enabled on the hash maps. + pdr_attr_id_(-1), + buffer_flag_attr_id_(-1) { max_allowed_workers_ = 1; } - static constexpr uint32_t kDefaultNumEntries = 1 << 15; static const Commands cmds; + CommandResponse Init(const bess::pb::FlowMeasureArg &arg); void DeInit() override; void ProcessBatch(Context *ctx, bess::PacketBatch *batch) override; - std::string GetDesc() const override { return ""; }; + std::string GetDesc() const override { return ""; } + CommandResponse CommandReadStats( const bess::pb::FlowMeasureCommandReadArg &arg); CommandResponse CommandFlipFlag( const bess::pb::FlowMeasureCommandFlipArg &arg); private: - // Flag represents a collection of possible values to select buffer sides. enum class Flag { FLAG_VALUE_INVALID = 0, - FLAG_VALUE_A, - FLAG_VALUE_B, + FLAG_VALUE_A = 1, + FLAG_VALUE_B = 2, FLAG_VALUE_MAX = FLAG_VALUE_B, }; template static constexpr bool Flag_IsValid(T value) { - Flag flag = static_cast(value); - return flag > Flag::FLAG_VALUE_INVALID && flag <= Flag::FLAG_VALUE_MAX; + Flag f = static_cast(value); + return f > Flag::FLAG_VALUE_INVALID && f <= Flag::FLAG_VALUE_MAX; } - static const std::string Flag_Name(const Flag &flag) { + static std::string Flag_Name(const Flag &flag) { switch (flag) { case Flag::FLAG_VALUE_INVALID: return "FLAG_VALUE_INVALID"; @@ -62,48 +65,74 @@ class FlowMeasure final : public Module { case Flag::FLAG_VALUE_B: return "FLAG_VALUE_B"; default: - return ""; + return ""; } } - // TableKey encapsulates all information used to identify a flow and is used - // as the lookup key in the hash tables. It is packed and aligned to - // calculating a hash over the raw bytes of the struct is ok. struct __attribute__((packed, aligned(16))) TableKey { uint64_t fseid; uint64_t pdr; + TableKey(uint64_t fseid, uint64_t pdr) : fseid(fseid), pdr(pdr) {} TableKey() : fseid(0), pdr(0) {} + + // Overloaded equality operator to support TableKey lookup in + // std::unordered_map. + bool operator==(const TableKey &o) const { + return fseid == o.fseid && pdr == o.pdr; + } + std::string ToString() const { std::stringstream ss; ss << "{ fseid: " << fseid << ", pdr: " << pdr << " }"; return ss.str(); } }; + static_assert(std::is_trivially_copyable::value, - "TableKey must be is_trivially_copyable."); + "TableKey must be trivially copyable."); + + // Custom hash function to enable TableKey use within standard C++ associative + // containers. + struct TableKeyHash { + std::size_t operator()(const TableKey &k) const noexcept { + std::size_t h = k.fseid; + h ^= h >> 33; + h *= 0xff51afd7ed558ccdULL; + h ^= h >> 33; + h *= 0xc4ceb9fe1a85ec53ULL; + h ^= h >> 33; + h ^= k.pdr * 0x9e3779b97f4a7c15ULL; + return h; + } + }; - // SessionStats ... struct SessionStats { uint64_t pkt_count; uint64_t byte_count; uint64_t last_latency; + + static constexpr uint64_t kBucketWidthNs = 1000; + static constexpr uint64_t kNumBuckets = 100; + Histogram latency_histogram; Histogram jitter_histogram; mutable std::mutex mutex; - static constexpr uint64_t kBucketWidthNs = 1000; // accuracy: 1 us - static constexpr uint64_t kNumBuckets = 100; // range: 0 - 100 us + SessionStats() : pkt_count(0), byte_count(0), last_latency(0), latency_histogram(kNumBuckets, kBucketWidthNs), jitter_histogram(kNumBuckets, kBucketWidthNs) {} - // Move allowed, copy not allowed. + + // Explicitly deleted copy/move operations to ensure stat objects remain + // stationary in memory. SessionStats(const SessionStats &) = delete; - SessionStats(SessionStats &&) noexcept = default; SessionStats &operator=(const SessionStats &) = delete; - SessionStats &operator=(SessionStats &&) = default; + SessionStats(SessionStats &&) = delete; + SessionStats &operator=(SessionStats &&) = delete; + void reset() { pkt_count = 0; byte_count = 0; @@ -112,17 +141,60 @@ class FlowMeasure final : public Module { jitter_histogram.Reset(); } }; + + // Encapsulated map and shared_mutex into a Buffer struct to manage + // side-specific locking and lifecycle. + struct Buffer { + // Uses shared_lock for high-performance concurrent lookups and unique_lock + // for flow insertion/clearing. + mutable std::shared_mutex map_mutex; + // Replaced fixed-size DPDK hash with dynamic unordered_map to support an + // unlimited number of flows. + std::unordered_map map; + + Buffer() = default; + ~Buffer() { clear(); } + + Buffer(const Buffer &) = delete; + Buffer &operator=(const Buffer &) = delete; + + void clear() { + std::unique_lock lk(map_mutex); + for (auto &kv : map) + delete kv.second; + map.clear(); + } + + // Stored pointers instead of objects to prevent pointer invalidation during + // map rehashing. + SessionStats *get_or_create(const TableKey &key) { + { + std::shared_lock lk(map_mutex); + auto it = map.find(key); + if (it != map.end()) + return it->second; + } + std::unique_lock lk(map_mutex); + auto [it, inserted] = map.emplace(key, nullptr); + if (inserted) + it->second = new SessionStats(); + return it->second; + } + }; + bool leader_; - Flag current_flag_value_; // protected by flag_mutex_ + Flag current_flag_value_; mutable std::mutex flag_mutex_; - rte_hash *table_a_; - rte_hash *table_b_; - std::vector table_data_a_; - std::vector table_data_b_; + + // Switched to unique_ptr for buffers to ensure automatic and safe memory + // cleanup during DeInit. + std::unique_ptr buf_a_; + std::unique_ptr buf_b_; + int ts_attr_id_; int fseid_attr_id_; int pdr_attr_id_; int buffer_flag_attr_id_; }; -#endif // BESS_MODULES_QOS_MEASURE_H_ +#endif // BESS_MODULES_QOS_MEASURE_H_ \ No newline at end of file