Merge branch 'feature/net-statistics' into 'main'

Add tx and rx bps statistics

See merge request beyondaka/beyond-protocol!76

include/ftl/protocol/self.hpp

@@ -188,6 +188,11 @@ class Self {
     // Used for testing
     ftl::net::Universe *getUniverse() const { return universe_.get(); }
 
+    // === Statistics methods ===
+
+    float getKBitsPerSecondTX() const;
+    float getKBitsPerSecondRX() const;
+
     // === The RPC methods ===
 
     /**

src/peer.cpp

@@ -338,6 +338,8 @@ void Peer::data() {
         return;
     }
 
+    net_->rxBytes_ += rc;
+
     // May possibly need locking
     recv_buf_.buffer_consumed(rc);
 
@@ -576,6 +578,7 @@ int Peer::_send() {
         _close(reconnect_on_socket_error_);
     }
 
+    net_->txBytes_ += c;
     return c;
 }
 

src/self.cpp

@@ -81,6 +81,14 @@ bool Self::isConnected(const std::string &s) {
     return universe_->isConnected(s);
 }
 
+float Self::getKBitsPerSecondTX() const {
+    return universe_->getKBitsPerSecondTX();
+}
+
+float Self::getKBitsPerSecondRX() const {
+    return universe_->getKBitsPerSecondRX();
+}
+
 size_t Self::numberOfNodes() const {
     return universe_->numberOfPeers();
 }

src/universe.cpp

@@ -16,6 +16,8 @@
 #define LOGURU_REPLACE_GLOG 1
 #include <ftl/lib/loguru.hpp>
 
+#include <ftl/time.hpp>
+
 #include <nlohmann/json.hpp>
 
 #include "protocol/connection.hpp"
@@ -425,6 +427,15 @@ std::list<PeerPtr> Universe::getPeers() const {
 }
 
 void Universe::_periodic() {
+    // Update stats
+    int64_t now = ftl::time::get_time();
+    float seconds = static_cast<float>(now - stats_lastTS_) / 1000.0f;
+    stats_rxkbps_ = static_cast<float>(rxBytes_) / seconds / 1024.0f;
+    rxBytes_ = 0;
+    stats_txkbps_ = static_cast<float>(txBytes_) / seconds / 1024.0f;
+    txBytes_ = 0;
+    stats_lastTS_ = now;
+
     auto i = reconnects_.begin();
     while (i != reconnects_.end()) {
         std::string addr = i->peer->getURI();

src/universe.hpp

@@ -170,6 +170,9 @@ class Universe {
     void setSendBufferSize(ftl::URI::scheme_t s, size_t size);
     void setRecvBufferSize(ftl::URI::scheme_t s, size_t size);
 
+    float getKBitsPerSecondTX() const { return stats_txkbps_ * 8.0f; }
+    float getKBitsPerSecondRX() const { return stats_rxkbps_ * 8.0f; }
+
     static inline std::shared_ptr<Universe> getInstance() { return instance_; }
 
     void setMaxConnections(size_t m);
@@ -203,6 +206,13 @@ class Universe {
 
     std::unique_ptr<NetImplDetail> impl_;
 
+    // Statistics data.
+    float stats_txkbps_ = 0.0f;
+    float stats_rxkbps_ = 0.0f;
+    std::atomic_size_t txBytes_ = 0;
+    std::atomic_size_t rxBytes_ = 0;
+    int64_t stats_lastTS_ = 0;
+
     std::vector<std::unique_ptr<ftl::net::internal::SocketServer>> listeners_;
     std::vector<ftl::net::PeerPtr> peers_;
     std::unordered_map<std::string, size_t> peer_by_uri_;