REFACTORING: write_to_csv optimization (#489)

close #491

---------

Co-authored-by: Alexey Zharkov <alexezharkov@yandex.ru>

Author: Az3git

source/connection/flow/tcp/tcp_flow.cpp

@@ -46,8 +46,8 @@ TcpFlow::TcpFlow(Id a_id, std::shared_ptr<IConnection> a_conn,
 }
 
 void TcpFlow::update(Packet packet) {
-    PacketType type =
-        static_cast<PacketType>(packet.flags.get_flag(m_packet_type_label).value_or_throw());
+    PacketType type = static_cast<PacketType>(
+        packet.flags.get_flag(m_packet_type_label).value_or_throw());
     if (m_src.expired()) {
         LOG_ERROR(fmt::format("Sender exprired for flow {}; ignore packet {}",
                               to_string(), packet.to_string()));
@@ -317,7 +317,8 @@ void TcpFlow::process_ack(Packet ack, std::size_t confirm_count) {
 Packet TcpFlow::generate_data_packet(PacketNum packet_num) {
     Packet packet;
     packet.flags = get_flag_manager();
-    packet.flags.set_flag(m_packet_type_label, PacketType::DATA).log_err_if_not_present("Failed to set packet type (DATA)");
+    packet.flags.set_flag(m_packet_type_label, PacketType::DATA)
+        .log_err_if_not_present("Failed to set packet type (DATA)");
 
     set_avg_rtt_if_present(packet);
     packet.size = m_packet_size;
@@ -335,7 +336,8 @@ Packet TcpFlow::generate_data_packet(PacketNum packet_num) {
 void TcpFlow::set_avg_rtt_if_present(Packet& packet) {
     std::optional<TimeNs> avg_rtt = m_rtt_statistics.get_mean();
     if (avg_rtt.has_value()) {
-        set_avg_rtt_flag(packet.flags, avg_rtt.value()).log_err_if_not_present("Failed to set average RTT");
+        set_avg_rtt_flag(packet.flags, avg_rtt.value())
+            .log_err_if_not_present("Failed to set average RTT");
     }
 }
 
@@ -410,21 +412,26 @@ Packet TcpFlow::create_ack(Packet data) {
     ack.congestion_experienced = data.congestion_experienced;
 
     ack.flags = get_flag_manager();
-    ack.flags.set_flag(
-        m_packet_type_label,
-        (M_COLLECTIVE_ACK_SUPPORT ? PacketType::COLLECTIVE_ACK
-                                  : PacketType::ACK)).log_err_if_not_present("Failed to set ACK flag");
-    ack.flags.set_flag(m_ack_ttl_label, data.ttl).log_err_if_not_present("Failed to set TTL flag");
+    ack.flags
+        .set_flag(m_packet_type_label,
+                  (M_COLLECTIVE_ACK_SUPPORT ? PacketType::COLLECTIVE_ACK
+                                            : PacketType::ACK))
+        .log_err_if_not_present("Failed to set ACK flag");
+    ack.flags.set_flag(m_ack_ttl_label, data.ttl)
+        .log_err_if_not_present("Failed to set TTL flag");
 
     utils::StrExpected<TimeNs> exp_avg_rtt = get_avg_rtt_label(data.flags);
     // TODO: use LOG_INFO for this part
-    bool is_not_present = exp_avg_rtt.log_err_if_not_present(fmt::format("avg rtt flag does not set in data packet {} so it "
-                        "will not be set in ack {}", data.to_string(), ack.to_string()));
+    bool is_not_present = exp_avg_rtt.log_err_if_not_present(
+        fmt::format("avg rtt flag does not set in data packet {} so it "
+                    "will not be set in ack {}",
+                    data.to_string(), ack.to_string()));
     if (is_not_present) {
         return ack;
     }
 
-    set_avg_rtt_flag(ack.flags, exp_avg_rtt.value()).log_err_if_not_present("Failed to set average RTT");
+    set_avg_rtt_flag(ack.flags, exp_avg_rtt.value())
+        .log_err_if_not_present("Failed to set average RTT");
     return ack;
 }
 

source/device/hashers/adaptive_flowlet_hasher.cpp

@@ -34,8 +34,9 @@ std::uint32_t AdaptiveFlowletHasher::get_hash(const Packet& packet) {
     TimeNs elapsed_from_last_seen = curr_time - last_seen;
 
     utils::StrExpected<TimeNs> exp_avg_rtt = get_avg_rtt_label(packet.flags);
-    bool is_not_present = exp_avg_rtt.log_err_if_not_present("Adaptive flowlet hasher could not find avg rtt;"
-                        "returned previous hash");
+    bool is_not_present = exp_avg_rtt.log_err_if_not_present(
+        "Adaptive flowlet hasher could not find avg rtt;"
+        "returned previous hash");
     if (is_not_present) {
         return ecmp_hash + shift;
     }

source/metrics/metrics_storage.cpp

@@ -10,7 +10,8 @@ void MetricsStorage::add_record(TimeNs time, double value) {
     m_records.emplace_back(time, value);
 }
 
-std::vector<std::pair<TimeNs, double> > MetricsStorage::get_records() const {
+const std::vector<std::pair<TimeNs, double>>& MetricsStorage::get_records()
+    const {
     return m_records;
 }
 

source/metrics/metrics_storage.hpp

@@ -12,15 +12,15 @@ class MetricsStorage {
 public:
     void add_record(TimeNs time, double value);
 
-    std::vector<std::pair<TimeNs, double> > get_records() const;
+    const std::vector<std::pair<TimeNs, double>>& get_records() const;
     void export_to_file(std::filesystem::path path) const;
     matplot::figure_handle get_picture(PlotMetadata metadata) const;
     void draw_plot(std::filesystem::path path, PlotMetadata metadata) const;
     void draw_on_plot(matplot::figure_handle& fig,
                       std::string_view name = "") const;
 
 private:
-    std::vector<std::pair<TimeNs, double> > m_records;
+    std::vector<std::pair<TimeNs, double>> m_records;
 };
 
 }  // namespace sim

source/metrics/write_to_csv.cpp

@@ -2,54 +2,102 @@
 
 #include <cmath>
 #include <limits>
+#include <span>
 
 namespace sim {
 
 void write_to_csv(
-    const std::vector<std::pair<MetricsStorage, std::string> >& storages,
-    std::filesystem::path output_path) {
-    size_t count_storages = storages.size();
-    // values[time][i] is a value of metric for i-th storage at time time;
-    // If there were no measurement at time, values[time][i] =
-    // std::numeric_limits<double>::quiet_NaN()
-    std::map<TimeNs, std::vector<double> > values;
-    double nan = std::numeric_limits<double>::quiet_NaN();
-    std::vector<double> default_values(count_storages, nan);
-    for (size_t i = 0; i < count_storages; i++) {
-        for (const auto& [time, value] : storages[i].first.get_records()) {
-            if (values.find(time) == values.end()) {
-                values[time] = default_values;
+    const std::vector<std::pair<MetricsStorage, std::string>>& storages,
+    const std::filesystem::path output_path) {
+    static constexpr std::size_t FLUSH_THRESHOLD = (1 << 20);  // 1 MB
+    const size_t count_storages = storages.size();
+
+    using Sample = std::pair<TimeNs, double>;
+    using SeriesSpan = std::span<const Sample>;
+
+    // References to the underlying data vectors
+    std::vector<SeriesSpan> series;
+    series.reserve(count_storages);
+
+    for (const auto& [storage, name] : storages) {
+        const std::vector<Sample>& recs = storage.get_records();
+        series.emplace_back(recs.data(), recs.size());
+    }
+
+    // Indices of the current element for each storage
+    std::vector<size_t> idx(count_storages, 0);
+
+    const double nan = std::numeric_limits<double>::quiet_NaN();
+    // Most recent known values (forward-fill state)
+    std::vector<double> values(count_storages, nan);
+
+    utils::create_all_directories(output_path);
+    std::ofstream out(output_path, std::ios::binary);
+    if (!out) {
+        throw std::runtime_error("Failed to open output file: " +
+                                 output_path.string());
+    }
+
+    fmt::memory_buffer buffer;
+    auto it = std::back_inserter(buffer);
+
+    // Header row
+    fmt::format_to(it, "Time");
+    for (const auto& p : storages) {
+        fmt::format_to(it, ",{}", p.second);
+    }
+    fmt::format_to(it, "\n");
+
+    auto get_next_time = [&]() -> std::optional<TimeNs> {
+        std::optional<TimeNs> result;
+        for (size_t i = 0; i < count_storages; ++i) {
+            const auto& recs = series[i];
+            if (idx[i] < recs.size()) {
+                TimeNs t = recs[idx[i]].first;
+                if (!result || t < result.value()) {
+                    result = t;
+                }
             }
-            values[time][i] = value;
         }
-    }
+        return result;
+    };
 
-    std::vector<double> previous_time_row = default_values;
-    // push values by time using increasing order of keys (time) in std::map
-    for (auto& [time, time_values] : values) {
-        for (size_t i = 0; i < count_storages; i++) {
-            if (std::isnan(time_values[i])) {
-                time_values[i] = previous_time_row[i];
-            } else {
-                previous_time_row[i] = time_values[i];
+    auto flush_buffer = [&]() {
+        out.write(buffer.data(), static_cast<std::streamsize>(buffer.size()));
+        buffer.clear();
+    };
+
+    // Main time-merge loop (k-way merge)
+    for (auto next_time_opt = get_next_time(); next_time_opt.has_value();
+         next_time_opt = get_next_time()) {
+        TimeNs next_time = next_time_opt.value();
+        // 1) Update forward-filled values for all storages that have a record
+        // at next_time
+        for (size_t i = 0; i < count_storages; ++i) {
+            const auto& recs = series[i];
+            if (idx[i] < recs.size() && recs[idx[i]].first == next_time) {
+                values[i] = recs[idx[i]].second;
+                ++idx[i];
             }
         }
-    }
+        // 2) Append the row to the buffer
+        fmt::format_to(it, "{}", next_time.value());
+        for (double v : values) {
+            fmt::format_to(it, ",{}", v);
+        }
+        fmt::format_to(it, "\n");
 
-    utils::create_all_directories(output_path);
-    std::ofstream out(output_path);
-    out << "Time";
-    for (size_t i = 0; i < count_storages; i++) {
-        out << ',' << storages[i].second;
-    }
-    out << '\n';
-    for (const auto& [time, time_values] : values) {
-        out << time;
-        for (auto value : time_values) {
-            out << ',' << value;
+        // 3) Optionally flush the buffer periodically to avoid memory growth
+        // (e.g., every 1–8 MB)
+        if (buffer.size() > FLUSH_THRESHOLD) {
+            flush_buffer();
         }
-        out << '\n';
+    }
+
+    // Final flush
+    if (buffer.size()) {
+        flush_buffer();
     }
 }
 
-}  // namespace sim
+}  // namespace sim

source/packet.hpp

@@ -3,8 +3,8 @@
 #include <string>
 
 #include "connection/flow/i_flow.hpp"
-#include "utils/flag_manager.hpp"
 #include "data.hpp"
+#include "utils/flag_manager.hpp"
 
 namespace sim {
 

source/parser/topology/ecn/ecn_parser.cpp

@@ -10,4 +10,4 @@ ECN EcnParser::parse_ecn(const ConfigNode& node) {
     return ECN(min, max, probability);
 }
 
-}
+}  // namespace sim

source/utils/flag_manager.hpp

@@ -48,7 +48,7 @@ class FlagManager {
     }
 
     [[nodiscard]] utils::StrExpected<void> set_flag(FlagId id,
-                                                            BitStorage value) {
+                                                    BitStorage value) {
         auto it = m_flag_manager.find(id);
         if (it == m_flag_manager.end()) {
             return std::unexpected(

source/utils/str_expected.hpp

@@ -17,7 +17,7 @@ class StrExpected : public std::expected<T, std::string> {
     StrExpected(std::unexpected<U> a_unexpected)
         : std::expected<T, std::string>(a_unexpected) {
         static_assert(std::is_constructible_v<std::expected<T, std::string>,
-                                              std::unexpected<U> >);
+                                              std::unexpected<U>>);
     }
 
     StrExpected(T a_value)
@@ -72,7 +72,8 @@ class StrExpected : public std::expected<T, std::string> {
         return result;
     }
 
-    bool apply_if_not_present(std::function<void(const std::string&)> apply_func) {
+    bool apply_if_not_present(
+        std::function<void(const std::string&)> apply_func) {
         bool result = !this->has_value();
         if (result) {
             apply_func(this->error());
@@ -102,8 +103,7 @@ class StrExpected<void> : public std::expected<void, std::string> {
                                               std::unexpected<U>>);
     }
 
-    StrExpected()
-        : std::expected<void, std::string>(std::in_place) {}
+    StrExpected() : std::expected<void, std::string>(std::in_place) {}
 
     template <typename TErr = std::runtime_error>
     void value_or_throw() const {
@@ -154,7 +154,8 @@ class StrExpected<void> : public std::expected<void, std::string> {
         return result;
     }
 
-    bool apply_if_not_present(std::function<void(const std::string&)> apply_func) {
+    bool apply_if_not_present(
+        std::function<void(const std::string&)> apply_func) {
         bool result = !this->has_value();
         if (result) {
             apply_func(this->error());

test/_mocks/new_conn_mock.cpp

@@ -3,10 +3,10 @@
 namespace test {
 
 NewConnectionMock::NewConnectionMock(
-    Id a_id,
-    [[maybe_unused]] std::shared_ptr<sim::IHost> a_src,
+    Id a_id, [[maybe_unused]] std::shared_ptr<sim::IHost> a_src,
     [[maybe_unused]] std::shared_ptr<sim::IHost> a_dest,
-    std::shared_ptr<sim::INewMPLB> a_mplb) : m_id(std::move(a_id)) {
+    std::shared_ptr<sim::INewMPLB> a_mplb)
+    : m_id(std::move(a_id)) {
     m_context.mplb = std::move(a_mplb);
 }