Add Perf counters tx_bytes and rx_bytes

Author: cstlee

include/Homa/Perf.h

@@ -34,6 +34,12 @@ struct Stats {
 
     /// CPU time spent actively processing Homa messages in cycles.
     uint64_t active_cycles;
+
+    /// Number of bytes sent by the transport.
+    uint64_t tx_bytes;
+
+    /// Number of bytes received by the transport.
+    uint64_t rx_bytes;
 };
 
 /**

src/ControlPacket.h

@@ -18,6 +18,8 @@
 
 #include <Homa/Driver.h>
 
+#include "Perf.h"
+
 namespace Homa {
 namespace Core {
 namespace ControlPacket {
@@ -42,6 +44,7 @@ send(Driver* driver, Driver::Address address, Args&&... args)
     packet->length = sizeof(PacketHeaderType);
     packet->address = address;
     packet->priority = driver->getHighestPacketPriority();
+    Perf::counters.tx_bytes.add(packet->length);
     driver->sendPacket(packet);
     driver->releasePackets(&packet, 1);
 }

src/Perf.h

@@ -71,6 +71,8 @@ struct Counters {
      */
     Counters()
         : active_cycles(0)
+        , tx_bytes(0)
+        , rx_bytes(0)
     {}
 
     /**
@@ -84,6 +86,8 @@ struct Counters {
     void add(const Counters* other)
     {
         active_cycles.add(other->active_cycles);
+        tx_bytes.add(other->tx_bytes);
+        rx_bytes.add(other->rx_bytes);
     }
 
     /**
@@ -92,10 +96,18 @@ struct Counters {
     void dumpStats(Stats* stats)
     {
         stats->active_cycles = active_cycles.get();
+        stats->tx_bytes = tx_bytes.get();
+        stats->rx_bytes = rx_bytes.get();
     }
 
     /// CPU time spent actively processing Homa messages in cycles.
     Stat<uint64_t> active_cycles;
+
+    /// Number of bytes sent by the transport.
+    Stat<uint64_t> tx_bytes;
+
+    /// Number of bytes received by the transport.
+    Stat<uint64_t> rx_bytes;
 };
 
 /**

src/Sender.cc

@@ -231,6 +231,7 @@ Sender::handleResendPacket(Driver::Packet* packet, Driver* driver)
             Driver::Packet* packet = info->packets->getPacket(i);
             packet->priority = resendPriority;
             // Packets will be sent at the priority their original priority.
+            Perf::counters.tx_bytes.add(packet->length);
             driver->sendPacket(packet);
         }
     }
@@ -375,6 +376,7 @@ Sender::handleUnknownPacket(Driver::Packet* packet, Driver* driver)
             Driver::Packet* dataPacket = message->getPacket(0);
             assert(dataPacket != nullptr);
             dataPacket->priority = policy.priority;
+            Perf::counters.tx_bytes.add(dataPacket->length);
             driver->sendPacket(dataPacket);
             message->state.store(OutMessage::Status::SENT);
         } else {
@@ -777,6 +779,7 @@ Sender::sendMessage(Sender::Message* message, Driver::Address destination)
         Driver::Packet* packet = message->getPacket(0);
         assert(packet != nullptr);
         packet->priority = policy.priority;
+        Perf::counters.tx_bytes.add(packet->length);
         driver->sendPacket(packet);
         message->state.store(OutMessage::Status::SENT);
     } else {
@@ -985,6 +988,7 @@ Sender::trySend()
             }
             // ... if not, send away!
             packet->priority = info->priority;
+            Perf::counters.tx_bytes.add(packet->length);
             driver->sendPacket(packet);
             int packetDataBytes =
                 packet->length - info->packets->TRANSPORT_HEADER_LENGTH;

src/TransportImpl.cc

@@ -103,6 +103,7 @@ TransportImpl::processPackets()
         Driver::Packet* packet = packets[i];
         assert(packet->length >=
                Util::downCast<int>(sizeof(Protocol::Packet::CommonHeader)));
+        Perf::counters.rx_bytes.add(packet->length);
         Protocol::Packet::CommonHeader* header =
             static_cast<Protocol::Packet::CommonHeader*>(packet->payload);
         switch (header->opcode) {