changes by review

Author: Alexey Zharkov

.github/workflows/load_testing.yml

@@ -8,10 +8,19 @@ on:
 
 env:
   BUILD_DIR: build
-  GENERATORS_DIR: scripts/generators 
-  FLAMEGRAPH_ARTIFACT_DIR: flamegraph
+  EXECUTABLE_PATH: build/nons
+
+  GENERATORS_DIR: scripts/generators
+  FAT_TREE_GENERATOR: scripts/generators/topology/fat_tree/fat_tree.py
+  FAT_TREE_CONFIGS_DIR: scripts/generators/topology/fat_tree
+  FAT_TREE_SMALL_CONFIG: scripts/generators/topology/fat_tree/small_fat_tree_config.yml
+  FAT_TREE_LARGE_CONFIG: scripts/generators/topology/fat_tree/large_fat_tree_config.yml
+  ALL_TO_ALL_GENERATOR: scripts/generators/ti_simulation/all-to-all/all-to-all.py
+
   FLAMEGRAPH_TOPOLOGY_NAME: fat_tree_topology.yml
   FLAMEGRAPH_SIMULATION_NAME: fat_tree_simulation.yml
+  FLAMEGRAPH_ARTIFACT_DIR: flamegraph
+  FLAMEGRAPH_OUTPUT_DIR: flamegraph/flamegraph
 
 jobs:
   load_testing:
@@ -23,29 +32,19 @@ jobs:
     - name: Update submodules
       run: git submodule update --init --recursive
 
-    - name: Derive env vars
-      run: |
-        echo "EXECUTABLE_PATH=$BUILD_DIR/nons" >> $GITHUB_ENV
-        echo "FAT_TREE_GENERATOR=$GENERATORS_DIR/topology/fat_tree/fat_tree.py" >> $GITHUB_ENV
-        echo "ALL_TO_ALL_GENERATOR=$GENERATORS_DIR/ti_simulation/all-to-all/all-to-all.py" >> $GITHUB_ENV
-        echo "FAT_TREE_CONFIGS_DIR=$GENERATORS_DIR/topology/fat_tree" >> $GITHUB_ENV
-        echo "FAT_TREE_SMALL_CONFIG=$GENERATORS_DIR/topology/fat_tree/small_fat_tree_config.yml" >> $GITHUB_ENV
-        echo "FAT_TREE_LARGE_CONFIG=$GENERATORS_DIR/topology/fat_tree/large_fat_tree_config.yml" >> $GITHUB_ENV
-        echo "FLAMEGRAPH_OUTPUT_DIR=$FLAMEGRAPH_ARTIFACT_DIR/flamegraph" >> $GITHUB_ENV
-
     - name: Configure CMake for building project
-      run: cmake -B $BUILD_DIR -DCMAKE_BUILD_TYPE=Release -DBUILD_PROJECT=ON -DBUILD_TESTS=OFF -DLOG_LEVEL=LOG_LEVEL_ERROR
+      run: cmake -B ${{env.BUILD_DIR}} -DCMAKE_BUILD_TYPE=Release -DBUILD_PROJECT=ON -DBUILD_TESTS=OFF -DLOG_LEVEL=LOG_LEVEL_ERROR
 
     - name: Build project
-      run: cmake --build $BUILD_DIR -j $(nproc)
+      run: cmake --build ${{env.BUILD_DIR}} -j $(nproc)
 
     - name: Update PYTHONPATH
       run: echo "PYTHONPATH=${{ github.workspace }}/scripts" >> $GITHUB_ENV
 
     - name: Run simulations
       run: | 
-        python3 scripts/generate_configs_and_run.py -e $EXECUTABLE_PATH -t $FAT_TREE_GENERATOR -tc $FAT_TREE_SMALL_CONFIG
-        python3 scripts/generate_configs_and_run.py -e $EXECUTABLE_PATH -t $FAT_TREE_GENERATOR -tc $FAT_TREE_LARGE_CONFIG
+        python3 scripts/generate_configs_and_run.py -e ${{env.EXECUTABLE_PATH}} -t ${{env.FAT_TREE_GENERATOR}} -tc ${{env.FAT_TREE_SMALL_CONFIG}}
+        python3 scripts/generate_configs_and_run.py -e ${{env.EXECUTABLE_PATH}} -t ${{env.FAT_TREE_GENERATOR}} -tc ${{env.FAT_TREE_LARGE_CONFIG}}
       
 
   profiling_load_testing:
@@ -56,34 +55,24 @@ jobs:
     - name: Update submodules
       run: git submodule update --init --recursive
 
-    - name: Derive env vars
-      run: |
-        echo "EXECUTABLE_PATH=$BUILD_DIR/nons" >> $GITHUB_ENV
-        echo "FAT_TREE_GENERATOR=$GENERATORS_DIR/topology/fat_tree/fat_tree.py" >> $GITHUB_ENV
-        echo "ALL_TO_ALL_GENERATOR=$GENERATORS_DIR/ti_simulation/all-to-all/all-to-all.py" >> $GITHUB_ENV
-        echo "FAT_TREE_CONFIGS_DIR=$GENERATORS_DIR/topology/fat_tree" >> $GITHUB_ENV
-        echo "FAT_TREE_SMALL_CONFIG=$GENERATORS_DIR/topology/fat_tree/small_fat_tree_config.yml" >> $GITHUB_ENV
-        echo "FAT_TREE_LARGE_CONFIG=$GENERATORS_DIR/topology/fat_tree/large_fat_tree_config.yml" >> $GITHUB_ENV
-        echo "FLAMEGRAPH_OUTPUT_DIR=$FLAMEGRAPH_ARTIFACT_DIR/flamegraph" >> $GITHUB_ENV
-
     - name: Configure CMake for profiling
-      run: cmake -B $BUILD_DIR -DCMAKE_BUILD_TYPE=Release -DBUILD_PROJECT=ON -DBUILD_TESTS=OFF -DPROFILING=ON -DLOG_LEVEL=LOG_LEVEL_ERROR
+      run: cmake -B ${{env.BUILD_DIR}} -DCMAKE_BUILD_TYPE=Release -DBUILD_PROJECT=ON -DBUILD_TESTS=OFF -DPROFILING=ON -DLOG_LEVEL=LOG_LEVEL_ERROR
 
     - name: Build project with profiling
-      run: cmake --build $BUILD_DIR -j $(nproc)
+      run: cmake --build ${{env.BUILD_DIR}} -j $(nproc)
 
     - name: Update PYTHONPATH
       run: echo "PYTHONPATH=${{ github.workspace }}/scripts" >> $GITHUB_ENV
 
     - name: Run simulations with profiling
       run: | 
-        python3 scripts/generate_configs_and_run.py -e $EXECUTABLE_PATH -t $FAT_TREE_GENERATOR -tc $FAT_TREE_SMALL_CONFIG
-        gprof $EXECUTABLE_PATH > profile.txt || true
-        head profile.txt || true
+        python3 scripts/generate_configs_and_run.py -e ${{env.EXECUTABLE_PATH}} -t ${{env.FAT_TREE_GENERATOR}} -tc ${{env.FAT_TREE_SMALL_CONFIG}}
+        gprof ${{env.EXECUTABLE_PATH}} > profile.txt
+        head profile.txt
         
-        python3 scripts/generate_configs_and_run.py -e $EXECUTABLE_PATH -t $FAT_TREE_GENERATOR -tc $FAT_TREE_LARGE_CONFIG
-        gprof $EXECUTABLE_PATH > profile.txt || true
-        head profile.txt || true
+        python3 scripts/generate_configs_and_run.py -e ${{env.EXECUTABLE_PATH}} -t ${{env.FAT_TREE_GENERATOR}} -tc ${{env.FAT_TREE_LARGE_CONFIG}}
+        gprof ${{env.EXECUTABLE_PATH}} > profile.txt
+        head profile.txt
 
   flamegraph:
     runs-on: ubuntu-latest
@@ -96,50 +85,41 @@ jobs:
     - name: Update submodules
       run: git submodule update --init --recursive
 
-    - name: Derive env vars
-      run: |
-        echo "EXECUTABLE_PATH=$BUILD_DIR/nons" >> $GITHUB_ENV
-        echo "FAT_TREE_GENERATOR=$GENERATORS_DIR/topology/fat_tree/fat_tree.py" >> $GITHUB_ENV
-        echo "ALL_TO_ALL_GENERATOR=$GENERATORS_DIR/ti_simulation/all-to-all/all-to-all.py" >> $GITHUB_ENV
-        echo "FAT_TREE_CONFIGS_DIR=$GENERATORS_DIR/topology/fat_tree" >> $GITHUB_ENV
-        echo "FAT_TREE_LARGE_CONFIG=$GENERATORS_DIR/topology/fat_tree/large_fat_tree_config.yml" >> $GITHUB_ENV
-        echo "FLAMEGRAPH_OUTPUT_DIR=$FLAMEGRAPH_ARTIFACT_DIR/flamegraph" >> $GITHUB_ENV
-
     - name: Get dependencies  
       run: git clone https://github.com/brendangregg/FlameGraph.git
 
     - name: Configure CMake for perf
-      run: cmake -B $BUILD_DIR -DCMAKE_BUILD_TYPE=Release -DBUILD_PROJECT=ON -DBUILD_TESTS=OFF -DLOG_LEVEL=LOG_LEVEL_ERROR -DCMAKE_CXX_FLAGS="-g -fno-omit-frame-pointer"
+      run: cmake -B ${{env.BUILD_DIR}} -DCMAKE_BUILD_TYPE=Release -DBUILD_PROJECT=ON -DBUILD_TESTS=OFF -DLOG_LEVEL=LOG_LEVEL_ERROR -DCMAKE_CXX_FLAGS="-g -fno-omit-frame-pointer"
 
     - name: Build project with profiling
-      run: cmake --build $BUILD_DIR -j $(nproc)
+      run: cmake --build ${{env.BUILD_DIR}} -j $(nproc)
 
     - name: Update PYTHONPATH
       run: echo "PYTHONPATH=${{ github.workspace }}/scripts" >> $GITHUB_ENV
 
     - name: Generate topology config  
-      run: python3 $FAT_TREE_GENERATOR -c $FAT_TREE_LARGE_CONFIG -o $FLAMEGRAPH_TOPOLOGY_NAME
+      run: python3 ${{env.FAT_TREE_GENERATOR}} -c ${{env.FAT_TREE_LARGE_CONFIG}} -o ${{env.FLAMEGRAPH_TOPOLOGY_NAME}}
 
     - name: Generate simulation config
-      run: python3 $ALL_TO_ALL_GENERATOR -t $FLAMEGRAPH_TOPOLOGY_NAME -o $FLAMEGRAPH_SIMULATION_NAME
+      run: python3 ${{env.ALL_TO_ALL_GENERATOR}} -t ${{env.FLAMEGRAPH_TOPOLOGY_NAME}} -o ${{env.FLAMEGRAPH_SIMULATION_NAME}}
 
     - name: Run perf
       run: |
-        sudo perf record -F 99 -g -- ./$EXECUTABLE_PATH -c $FLAMEGRAPH_SIMULATION_NAME --no-logs --no-plots > perf.data
+        sudo perf record -F 99 -g -- ./${{env.EXECUTABLE_PATH}} -c ${{env.FLAMEGRAPH_SIMULATION_NAME}} --no-logs --no-plots > perf.data
         perf report -i perf.data --header-only
-        perf script -i perf.data -v > out.perf 2>&1 || true
+        perf script -i perf.data -v > out.perf 2>&1
 
     - name: Prepare FlameGraph
       run: |
-        mkdir -p $FLAMEGRAPH_OUTPUT_DIR
+        mkdir -p ${{ env.FLAMEGRAPH_OUTPUT_DIR }}
 
     - name: Generate flamegraph
       run: |
         ./FlameGraph/stackcollapse-perf.pl out.perf > out.folded
-        ./FlameGraph/flamegraph.pl out.folded > $FLAMEGRAPH_OUTPUT_DIR/flamegraph.svg
+        ./FlameGraph/flamegraph.pl out.folded > ${{ env.FLAMEGRAPH_OUTPUT_DIR }}/flamegraph.svg
 
     - name: Create HTML    
-      run: python3 scripts/generate-html.py ./$FLAMEGRAPH_ARTIFACT_DIR
+      run: python3 scripts/generate-html.py ./${{ env.FLAMEGRAPH_ARTIFACT_DIR }}
 
     - name: Put github context to json file
       run: echo '${{ toJson(github) }}' > github_context.json    

scripts/common.py

@@ -30,13 +30,14 @@ def run_subprocess(subprocess_args : list[str], log_commands : bool = True, chec
     Runs subprocess with given args. If command fails, prints its stderr to stderr
     Returns result of subprocess.
     """
+    command_str = ' '.join(subprocess_args)
     if log_commands:
-        print(f"Run command: `{' '.join(subprocess_args)}`")
+        print(f"Run command: `{command_str}`")
 
     result = subprocess.run(subprocess_args, capture_output=True, text=True)
 
     if result.returncode != 0 and check_fails:
-        print(f"Command `{' '.join(subprocess_args)}` failed; stderr:", file=sys.stderr)
+        print(f"Command `{command_str}` failed; stderr:", file=sys.stderr)
         print(result.stderr, file=sys.stderr)
 
     return result

source/connection/flow/i_flow.hpp

@@ -17,6 +17,7 @@ class IFlow : public Identifiable {
     virtual void send_data(SizeByte data) = 0;
 
     virtual SizeByte get_packet_size() const = 0;
+    virtual SizeByte get_total_data_size_added_from_conn() const = 0;
     virtual SizeByte get_sent_data_size() const = 0;
     virtual SizeByte get_delivered_data_size() const = 0;
     virtual uint32_t retransmit_count() const = 0;
@@ -27,7 +28,7 @@ class IFlow : public Identifiable {
     virtual TimeNs get_fct() const = 0;
 
     virtual const BaseFlagManager& get_flag_manager() const = 0;
-    
+
     virtual std::shared_ptr<IHost> get_sender() const = 0;
     virtual std::shared_ptr<IHost> get_receiver() const = 0;
 

source/connection/flow/tcp/tcp_flow.cpp

@@ -32,6 +32,7 @@ TcpFlow::TcpFlow(Id a_id, std::shared_ptr<IConnection> a_conn,
       m_rto_steady(false),
       m_retransmit_count(0),
       m_packets_in_flight(0),
+      m_total_data_from_conn(0),
       m_delivered_data_size(0),
       m_sent_data_size(0),
       m_next_packet_num(0) {
@@ -95,6 +96,7 @@ void TcpFlow::update(Packet packet) {
 }
 
 void TcpFlow::send_data(SizeByte data) {
+    m_total_data_from_conn += data;
     TimeNs now = Scheduler::get_instance().get_current_time();
 
     if (!m_sending_started) {
@@ -125,6 +127,10 @@ void TcpFlow::send_data(SizeByte data) {
 
 SizeByte TcpFlow::get_packet_size() const { return m_packet_size; }
 
+SizeByte TcpFlow::get_total_data_size_added_from_conn() const {
+    return m_total_data_from_conn;
+}
+
 SizeByte TcpFlow::get_sent_data_size() const { return m_sent_data_size; }
 
 SizeByte TcpFlow::get_delivered_data_size() const {

source/connection/flow/tcp/tcp_flow.hpp

@@ -23,6 +23,7 @@ class TcpFlow : public IFlow, public std::enable_shared_from_this<TcpFlow> {
     void send_data(SizeByte data) final;
 
     SizeByte get_packet_size() const final;
+    SizeByte get_total_data_size_added_from_conn() const final;
     SizeByte get_sent_data_size() const final;
     SizeByte get_delivered_data_size() const final;
     uint32_t retransmit_count() const final;
@@ -98,6 +99,7 @@ class TcpFlow : public IFlow, public std::enable_shared_from_this<TcpFlow> {
     std::uint32_t m_retransmit_count;
 
     std::uint32_t m_packets_in_flight;
+    SizeByte m_total_data_from_conn;
     SizeByte m_delivered_data_size;
     SizeByte m_sent_data_size;
     PacketNum m_next_packet_num;

source/scenario/action/send_data_action.cpp

@@ -6,7 +6,8 @@ namespace sim {
 
 SendDataAction::SendDataAction(TimeNs a_when, SizeByte a_size,
                                std::vector<std::weak_ptr<IConnection>> a_conns,
-                               int a_repeat_count, TimeNs a_repeat_interval, TimeNs a_jitter)
+                               int a_repeat_count, TimeNs a_repeat_interval,
+                               TimeNs a_jitter)
     : m_when(a_when),
       m_size(a_size),
       m_conns(std::move(a_conns)),
@@ -21,9 +22,10 @@ void SendDataAction::schedule() {
         auto conn = weak.lock();
         if (!conn) throw std::runtime_error("Expired connection in action");
 
-        std::uint64_t seed = std::hash<std::string>{}(conn->get_id()); 
+        std::uint64_t seed = std::hash<std::string>{}(conn->get_id());
         std::mt19937_64 rng(seed);
-        std::uniform_int_distribution<uint64_t> dist(0, m_jitter.value_nanoseconds());
+        std::uniform_int_distribution<uint64_t> dist(
+            0, m_jitter.value_nanoseconds());
 
         for (size_t i = 0; i < m_repeat_count; ++i) {
             TimeNs jitter_gap = use_jitter ? TimeNs(dist(rng)) : TimeNs(0);

source/scenario/action/send_data_action.hpp

@@ -1,5 +1,6 @@
 #pragma once
 #include <random>
+
 #include "connection/i_connection.hpp"
 #include "i_action.hpp"
 #include "scheduler.hpp"
@@ -11,7 +12,8 @@ class SendDataAction : public IAction {
 public:
     SendDataAction(TimeNs a_when, SizeByte a_size,
                    std::vector<std::weak_ptr<IConnection>> a_conns,
-                   int a_repeat_count, TimeNs a_repeat_interval, TimeNs a_jitter);
+                   int a_repeat_count, TimeNs a_repeat_interval,
+                   TimeNs a_jitter);
 
     void schedule() final;
 

source/utils/summary.cpp

@@ -15,16 +15,26 @@ Summary::Summary(
         Id conn_id = conn->get_id();
         SizeByte expt_data_delivery = conn->get_total_data_added();
         if (expt_data_delivery == SizeByte(0)) {
-            m_warnings.emplace_back(
-                fmt::format("Connection {} has 0 expected data delivery", conn_id));
+            m_warnings.emplace_back(fmt::format(
+                "Connection {} has 0 expected data delivery", conn_id));
         }
         SizeByte real_data_delivery(0);
         auto flows = conn->get_flows();
 
         for (const auto& flow : flows) {
+            const SizeByte added_from_conn =
+                flow->get_total_data_size_added_from_conn();
             const SizeByte delivered = flow->get_delivered_data_size();
             real_data_delivery += delivered;
 
+            if (added_from_conn > delivered) {
+                m_errors.emplace_back(fmt::format(
+                    "For flow {} of connection {} added from connection {} "
+                    "bytes but delivered {} bytes",
+                    flow->get_id(), conn_id, added_from_conn.value(),
+                    delivered.value()));
+            }
+
             const SizeByte sent = flow->get_sent_data_size();
             if (sent < delivered) {
                 m_errors.emplace_back(fmt::format(
@@ -33,7 +43,9 @@ Summary::Summary(
                     flow->get_id(), conn_id, sent.value(), delivered.value()));
             }
 
-            double overhead = delivered != SizeByte(0) ? (sent / delivered - 1) * 100 : std::nan("");
+            double overhead = delivered != SizeByte(0)
+                                  ? (sent / delivered - 1) * 100
+                                  : std::nan("");
 
             const TimeNs fct = flow->get_fct();
             SizeByte packet_size = flow->get_packet_size();
@@ -45,21 +57,31 @@ Summary::Summary(
             SizeByte retransmit_size = retransmit_count * packet_size;
 
             m_values[conn_id][flow->get_id()] =
-                FlowSummary{sent,         delivered,       packet_size,
-                            overhead,     retransmit_size, retransmit_count,
-                            sending_rate, throughput,      fct};
+                FlowSummary{added_from_conn,  sent,
+                            delivered,        packet_size,
+                            overhead,         retransmit_size,
+                            retransmit_count, sending_rate,
+                            throughput,       fct};
         }
         if (expt_data_delivery > real_data_delivery) {
             m_errors.emplace_back(fmt::format(
                 "For connection {} expected delivery {} but real is {}",
                 conn_id, expt_data_delivery.value(),
                 real_data_delivery.value()));
         }
-        if (expt_data_delivery == SizeByte(0) && real_data_delivery > SizeByte(0)) {
+        if (expt_data_delivery == SizeByte(0) &&
+            real_data_delivery > SizeByte(0)) {
             m_errors.emplace_back(fmt::format(
-                "For connection {} expected delivery 0 but real is {}",
-                conn_id, real_data_delivery.value()));
+                "For connection {} expected delivery 0 but real is {}", conn_id,
+                real_data_delivery.value()));
         }
+        // PS: case expt_data_delivery != real_data_delivery == 0 is is not
+        // always an error, because in our model, if there is less data to send
+        // than the size of the flow's packet, the flow still sends a full-sized
+        // packet.
+        // For example, if there is 1 byte to send and the packet size is 1500
+        // bytes, the flow will send a full 1500-byte packet.
+        // For this reason we do not consider this case an error.
     }
 }
 
@@ -69,26 +91,28 @@ void Summary::write_to_csv(std::filesystem::path& output_path) const {
     if (!out) {
         throw std::runtime_error("Failed to create file for summary");
     }
-    out << "Flow id, Packet size (bytes), Sent (bytes), Delivered (bytes), "
+    out << "Flow id, Packet size (bytes), Data from conn (bytes), Sent "
+           "(bytes), Delivered (bytes), "
            "Overhead (%), Retransmit size (bytes), Retransmit count, Sending "
            "rate (Gbps), Throughput (Gbps), FCT (ns)\n";
     for (const auto& [conn_id, flows] : m_values) {
         out << "\n";
         for (const auto& [flow_id, fs] : flows) {
             out << flow_id << ", " << fs.packet_size.value() << ", "
-                << fs.sent.value() << ", " << fs.delivered.value() << ", "
-                << fs.overhead << ", " << fs.retransmit_size.value() << ", "
-                << fs.retransmit_count << ", " << fs.sending_rate.value()
-                << ", " << fs.throughput.value() << ", " << fs.fct.value()
-                << "\n";
+                << fs.added_from_conn.value() << ", " << fs.sent.value() << ", "
+                << fs.delivered.value() << ", " << fs.overhead << ", "
+                << fs.retransmit_size.value() << ", " << fs.retransmit_count
+                << ", " << fs.sending_rate.value() << ", "
+                << fs.throughput.value() << ", " << fs.fct.value() << "\n";
         }
     }
     out.close();
 }
 
 void Summary::check() const {
     if (!m_warnings.empty()) {
-        LOG_WARN(fmt::format("Summary warnings: {}", fmt::join(m_warnings, "\n")));
+        LOG_WARN(
+            fmt::format("Summary warnings: {}", fmt::join(m_warnings, "\n")));
     }
     if (!m_errors.empty()) {
         throw std::runtime_error(

source/utils/summary.hpp

@@ -12,6 +12,7 @@ using ErrorMessage = std::string;
 using WarningMessage = std::string;
 
 struct FlowSummary {
+    SizeByte added_from_conn{0};
     SizeByte sent{0};
     SizeByte delivered{0};
     SizeByte packet_size{0};