Replace all Packet with PacketPtr

Author: PaulRalnikov

source/network/connection/flow/packet.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <memory>
 #include <string>
 
 #include "four_tuple.hpp"
@@ -12,7 +13,9 @@ using PathHash = std::uint32_t;
 
 struct Packet;
 
-using OnPacketDeliveryCallback = std::function<void(const Packet&)>;
+using PacketPtr = std::shared_ptr<Packet>;
+
+using OnPacketDeliveryCallback = std::function<void(PacketPtr)>;
 
 struct Packet : FourTuple {
     Packet(SizeByte a_size = SizeByte(0ul), Id a_source_id = "",
@@ -33,7 +36,7 @@ struct Packet : FourTuple {
     // Note: callback takes packet itself because it might be changed while
     // travelling along the net
     OnPacketDeliveryCallback callback =
-        []([[maybe_unused]] const Packet& packet) {};
+        []([[maybe_unused]] PacketPtr packet) {};
     TimeNs generated_time;  // Note: ACK's generated time is the data packet
                             // generated time
     TimeNs sent_time;  // Note: ACK's sent time is the data packet sent time

source/network/connection/flow/packet_ack_info.hpp

@@ -8,7 +8,7 @@ namespace sim {
 struct PacketAckInfo {
     TimeNs rtt;
     const utils::Statistics<TimeNs>& rtt_stat;
-    const Packet& ack;
+    PacketPtr ack;
 };
 
 }  // namespace sim

source/network/connection/flow/tcp/tcp_flow.cpp

@@ -72,7 +72,7 @@ TcpFlow::TcpFlow(Id a_id, FlowFourTuple a_four_tuple, bool a_ecn_capable,
     }
 }
 
-std::shared_ptr<Packet> TcpFlow::create_data_packet(
+PacketPtr TcpFlow::create_data_packet(
     PacketInfo info, std::shared_ptr<IHost> sender,
     std::shared_ptr<IHost> receiver) {
     auto packet = std::make_shared<Packet>();
@@ -82,7 +82,7 @@ std::shared_ptr<Packet> TcpFlow::create_data_packet(
     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);
+    set_avg_rtt_if_present(packet);
 
     packet->data_id = std::move(info.id);
     packet->sender_id = sender->get_id();
@@ -94,7 +94,7 @@ std::shared_ptr<Packet> TcpFlow::create_data_packet(
     std::shared_ptr<TcpFlow> flow = shared_from_this();
 
     packet->callback = [flow, delivery_callback = info.callback](
-                           const Packet& delivered_packet) {
+                           PacketPtr delivered_packet) {
         flow->process_data_packet(delivered_packet, delivery_callback);
     };
     packet->generated_time = info.generated_time;
@@ -107,15 +107,15 @@ std::shared_ptr<Packet> TcpFlow::create_data_packet(
     return packet;
 }
 
-void TcpFlow::set_avg_rtt_if_present(Packet& packet) {
+void TcpFlow::set_avg_rtt_if_present(PacketPtr packet) {
     std::optional<TimeNs> avg_rtt = m_context.rtt_statistics.get_mean();
     if (avg_rtt.has_value()) {
-        set_avg_rtt_flag(packet.flags, avg_rtt.value())
+        set_avg_rtt_flag(packet->flags, avg_rtt.value())
             .log_err_if_not_present("Failed to set average RTT");
     }
 }
 
-void TcpFlow::send_data_packet(std::shared_ptr<Packet> data) {
+void TcpFlow::send_data_packet(PacketPtr data) {
     TimeNs now = Scheduler::get_instance().get_current_time();
 
     Scheduler::get_instance().add(
@@ -127,13 +127,13 @@ void TcpFlow::send_data_packet(std::shared_ptr<Packet> data) {
     m_context.sender->enqueue_packet(data);
 }
 
-void TcpFlow::process_data_packet(const Packet& data,
+void TcpFlow::process_data_packet(PacketPtr data,
                                   const PacketCallback& callback) {
-    m_packet_reordering.add_record(data.packet_num);
+    m_packet_reordering.add_record(data->packet_num);
     m_metrics.packet_reordering->add_record(
         Scheduler::get_instance().get_current_time(),
         m_packet_reordering.value());
-    auto ack = std::make_shared<Packet>(data);
+    auto ack = std::make_shared<Packet>(*data);
     ack->sender_id = IdWithHash(m_context.receiver->get_id());
     ack->sender_port = m_context.receiver_port;
     ack->receiver_id = IdWithHash(m_context.sender->get_id());
@@ -146,42 +146,42 @@ void TcpFlow::process_data_packet(const Packet& data,
             fmt::format("Flow {}: could not set type label to ack packet {}",
                         m_id, ack->to_string()));
     }
-    SizeByte data_packet_size = data.size;
+    SizeByte data_packet_size = data->size;
 
     std::shared_ptr<TcpFlow> flow = shared_from_this();
     ack->callback = [flow, callback,
-                     data_packet_size](const Packet& delivered_ack) {
+                     data_packet_size](PacketPtr delivered_ack) {
         flow->process_ack(delivered_ack, data_packet_size, callback);
     };
 
     m_context.receiver->enqueue_packet(std::move(ack));
 }
 
-void TcpFlow::process_ack(const Packet& ack, SizeByte data_packet_size,
+void TcpFlow::process_ack(PacketPtr ack, SizeByte data_packet_size,
                           PacketCallback callback) {
     TimeNs now = Scheduler::get_instance().get_current_time();
 
-    TimeNs rtt = now - ack.sent_time;
+    TimeNs rtt = now - ack->sent_time;
 
     m_context.rtt_statistics.add_record(rtt);
 
     m_metrics.rtt->add_record(now, rtt.value());
 
     update_rto_on_ack();
 
-    if (!m_ack_monitor.confirm_one(ack.packet_num)) {
+    if (!m_ack_monitor.confirm_one(ack->packet_num)) {
         LOG_WARN(
             fmt::format("Flow {} got ack {} that confirms nothing; ignored",
-                        m_id, ack.to_string()));
+                        m_id, ack->to_string()));
         return;
     }
     m_context.last_ack_receive_time = now;
 
     m_context.delivered_size += data_packet_size;
 
     SpeedGbps delivery_rate =
-        (m_context.delivered_size - ack.delivered_data_size_at_origin) /
-        (now - ack.generated_time);
+        (m_context.delivered_size - ack->delivered_data_size_at_origin) /
+        (now - ack->generated_time);
 
     m_metrics.delivery_rate->add_record(now, delivery_rate.value());
     m_context.delivery_rate_statistics.add_record(delivery_rate);
@@ -196,7 +196,7 @@ void TcpFlow::update_rto_on_ack() {
     m_rto.is_steady = true;
 }
 
-void TcpFlow::on_timeout(std::shared_ptr<Packet> data) {
+void TcpFlow::on_timeout(PacketPtr data) {
     if (m_ack_monitor.is_confirmed(data->packet_num)) {
         LOG_INFO(fmt::format(
             "Flow {}: packet {} is confirmed when timeout reached; no retransmit",
@@ -213,7 +213,7 @@ void TcpFlow::update_rto_on_timeout() {
     }
 }
 
-void TcpFlow::retransmit_packet(std::shared_ptr<Packet> data) {
+void TcpFlow::retransmit_packet(PacketPtr data) {
     m_context.retransmit_size += data->size;
     send_data_packet(data);
 }

source/network/connection/flow/tcp/tcp_flow.hpp

@@ -43,25 +43,25 @@ class TcpFlow : public IFlow, public std::enable_shared_from_this<TcpFlow> {
     TcpFlow(Id a_id, FlowFourTuple a_four_tuple, bool a_ecn_capable, RTO a_rto,
             TcpFlowMetricsFilters a_metrics_flags);
 
-    std::shared_ptr<Packet> create_data_packet(PacketInfo info,
+    PacketPtr create_data_packet(PacketInfo info,
                                                std::shared_ptr<IHost> sender,
                                                std::shared_ptr<IHost> receiver);
 
-    void set_avg_rtt_if_present(Packet& packet);
+    void set_avg_rtt_if_present(PacketPtr packet);
 
-    void send_data_packet(std::shared_ptr<Packet> data);
+    void send_data_packet(PacketPtr data);
 
-    void process_data_packet(const Packet& data_packet,
+    void process_data_packet(PacketPtr data_packet,
                              const PacketCallback& callback);
 
-    void process_ack(const Packet& ack, SizeByte data_packet_size,
+    void process_ack(PacketPtr ack, SizeByte data_packet_size,
                      PacketCallback callback);
 
     void update_rto_on_ack();
 
-    void on_timeout(std::shared_ptr<Packet> data);
+    void on_timeout(PacketPtr data);
     void update_rto_on_timeout();
-    void retransmit_packet(std::shared_ptr<Packet> data);
+    void retransmit_packet(PacketPtr data);
 
 private:
     // flag labels

source/network/connection/mplb/cc/tahoe/tcp_tahoe_cc.cpp

@@ -13,7 +13,7 @@ TcpTahoeCC::TcpTahoeCC(double a_start_cwnd, double a_sstresh)
 
 void TcpTahoeCC::on_ack(const PacketAckInfo& info) {
     m_last_avg_rtt = info.rtt_stat.get_mean().value();
-    if (info.ack.congestion_experienced) {
+    if (info.ack->congestion_experienced) {
         on_timeout();
     } else if (m_cwnd < m_ssthresh) {
         // Slow start

source/network/connection/rdma/rdma_connection.cpp

@@ -154,7 +154,7 @@ void RdmaConnection::send_next_data_packet() {
     schedule_data_send();
 }
 
-std::shared_ptr<Packet> RdmaConnection::create_data_packet(const Data& data) {
+PacketPtr RdmaConnection::create_data_packet(const Data& data) {
     auto packet = std::make_shared<Packet>();
     packet->packet_num = m_next_packet_num++;
 
@@ -167,7 +167,7 @@ std::shared_ptr<Packet> RdmaConnection::create_data_packet(const Data& data) {
 
     RdmaConnectionPtr conn = shared_from_this();
 
-    packet->callback = [conn](const Packet& delivered_packet) {
+    packet->callback = [conn](PacketPtr delivered_packet) {
         conn->process_data_packet(delivered_packet);
     };
     TimeNs now = Scheduler::get_instance().get_current_time();
@@ -178,27 +178,27 @@ std::shared_ptr<Packet> RdmaConnection::create_data_packet(const Data& data) {
     return packet;
 }
 
-void RdmaConnection::process_data_packet(const Packet& packet) {
+void RdmaConnection::process_data_packet(PacketPtr packet) {
     if (!m_receiver_started) {
         m_receiver_started = true;
         schedule_ack_timer();
     }
-    if (packet.packet_num < m_next_expected_packet_num) {
+    if (packet->packet_num < m_next_expected_packet_num) {
         LOG_ERROR(fmt::format(
             "RDMA {}: receiver got data packet {} with number smaller "
             "than expected {}; ignored",
-            m_id, packet.to_string(), m_next_expected_packet_num));
-    } else if (packet.packet_num == m_next_expected_packet_num) {
+            m_id, packet->to_string(), m_next_expected_packet_num));
+    } else if (packet->packet_num == m_next_expected_packet_num) {
         process_expected_data_packet();
     } else {
         LOG_ERROR(fmt::format(
             "RDMA {}: receiver got data packet {} with number greater "
             "than expected {}; could not put it to reorder buffer; ignored",
-            m_id, packet.to_string(), m_next_expected_packet_num));
+            m_id, packet->to_string(), m_next_expected_packet_num));
         send_nak();
         return;
     }
-    if (packet.congestion_experienced) {
+    if (packet->congestion_experienced) {
         send_cnp();
     }
 }
@@ -230,7 +230,7 @@ void RdmaConnection::send_nak() {
 
     RdmaConnectionPtr conn = shared_from_this();
 
-    nak->callback = [conn](const Packet& nak) { conn->process_nak(nak); };
+    nak->callback = [conn](PacketPtr nak) { conn->process_nak(nak); };
     TimeNs now = Scheduler::get_instance().get_current_time();
     nak->generated_time = now;
     nak->sent_time = now;
@@ -241,8 +241,8 @@ void RdmaConnection::send_nak() {
     m_receiver->enqueue_packet(std::move(nak));
 }
 
-void RdmaConnection::process_nak(const Packet& nak) {
-    while (m_last_acked_pcn < nak.packet_num) {
+void RdmaConnection::process_nak(PacketPtr nak) {
+    while (m_last_acked_pcn < nak->packet_num) {
         m_last_acked_pcn++;
         confirm_first_unconfirmed_packet();
     }
@@ -269,7 +269,7 @@ void RdmaConnection::send_ack() {
 
     RdmaConnectionPtr conn = shared_from_this();
 
-    ack->callback = [conn](const Packet& delivered_packet) {
+    ack->callback = [conn](PacketPtr delivered_packet) {
         conn->process_ack(delivered_packet);
     };
     TimeNs now = Scheduler::get_instance().get_current_time();
@@ -282,8 +282,8 @@ void RdmaConnection::send_ack() {
     m_receiver->enqueue_packet(std::move(ack));
 }
 
-void RdmaConnection::process_ack(const Packet& ack) {
-    PacketNum ack_num = ack.packet_num;
+void RdmaConnection::process_ack(PacketPtr ack) {
+    PacketNum ack_num = ack->packet_num;
     if (ack_num < m_last_acked_pcn) {
         LOG_ERROR(
             fmt::format("RDMA connection {}: sender got ack with number "
@@ -322,7 +322,7 @@ void RdmaConnection::send_cnp() {
 
     RdmaConnectionPtr conn = shared_from_this();
 
-    cnp->callback = [conn](const Packet& cnp) { conn->process_cnp(cnp); };
+    cnp->callback = [conn](PacketPtr cnp) { conn->process_cnp(cnp); };
     TimeNs now = Scheduler::get_instance().get_current_time();
     cnp->generated_time = now;
     cnp->sent_time = now;
@@ -333,7 +333,7 @@ void RdmaConnection::send_cnp() {
     m_receiver->enqueue_packet(std::move(cnp));
 }
 
-void RdmaConnection::process_cnp([[maybe_unused]] const Packet& cnp) {
+void RdmaConnection::process_cnp([[maybe_unused]] PacketPtr cnp) {
     m_dcqcn.on_cnp();
 }
 
@@ -346,15 +346,15 @@ void RdmaConnection::confirm_first_unconfirmed_packet() {
                         m_id));
         return;
     }
-    const Packet& confirmed = *m_send_queue.front();
+    PacketPtr confirmed = m_send_queue.front();
     utils::Defer defer([this]() {
         m_send_queue.pop_front();
         if (m_send_queue.empty()) {
             m_dcqcn.stop();
         }
     });
 
-    const DataId& data_id = confirmed.data_id;
+    const DataId& data_id = confirmed->data_id;
     auto it = m_data_context_table.find(data_id);
     if (it == m_data_context_table.end()) {
         LOG_ERROR(
@@ -386,7 +386,7 @@ void RdmaConnection::send_ack_request() {
     RdmaConnectionPtr conn = shared_from_this();
 
     ack_request->callback =
-        [conn]([[maybe_unused]] const Packet& delivered_packet) {
+        [conn]([[maybe_unused]] PacketPtr delivered_packet) {
             conn->process_ack_request();
         };
     TimeNs now = Scheduler::get_instance().get_current_time();

source/network/connection/rdma/rdma_connectrion.hpp

@@ -51,23 +51,23 @@ class RdmaConnection : public IConnection,
 
     void send_next_data_packet();
 
-    std::shared_ptr<Packet> create_data_packet(const Data& data);
+    PacketPtr create_data_packet(const Data& data);
 
-    void process_data_packet(const Packet& data);
+    void process_data_packet(PacketPtr data);
 
     void process_expected_data_packet();
 
     void send_nak();
 
-    void process_nak(const Packet& nak);
+    void process_nak(PacketPtr nak);
 
     void send_ack();
 
-    void process_ack(const Packet& ack);
+    void process_ack(PacketPtr ack);
 
     void send_cnp();
 
-    void process_cnp(const Packet& cnp);
+    void process_cnp(PacketPtr cnp);
 
     void confirm_first_unconfirmed_packet();
 
@@ -89,7 +89,7 @@ class RdmaConnection : public IConnection,
 
     // Invariant: packet_num of first packet in this queue is equal to
     // m_last_acked_pcn
-    std::deque<std::shared_ptr<Packet>> m_send_queue;
+    std::deque<PacketPtr> m_send_queue;
 
     uint32_t m_send_window;