Merge branch 'bug/#6' into 'main'

#6 Add peer garbage collection

See merge request nicolaspope/beyond-protocol!17

include/ftl/protocol/self.hpp

@@ -114,6 +114,23 @@ class Self {
      */
     size_t numberOfNodes() const;
 
+    /**
+     * @brief Get the maximum allowed number of connections. Any attempt to connect more
+     * peers will result in them being rejected.
+     * 
+     * @return size_t 
+     */
+    size_t getMaxConnections() const;
+
+    /**
+     * @brief Set the maximum allowed connections. This should only be changed before
+     * there are any active connections, resizing with active connections could cause
+     * errors. The default number is 10.
+     * 
+     * @param m Number of allowed node connections 
+     */
+    void setMaxConnections(size_t m);
+
     /**
      * @brief Will block until all currently registered connnections have completed.
      * You should not use this, but rather use onConnect.

src/self.cpp

@@ -76,6 +76,14 @@ size_t Self::numberOfNodes() const {
     return universe_->numberOfPeers();
 }
 
+size_t Self::getMaxConnections() const {
+    return universe_->getMaxConnections();
+}
+
+void Self::setMaxConnections(size_t m) {
+    universe_->setMaxConnections(m);
+}
+
 int Self::waitConnections(int seconds) {
     return universe_->waitConnections(seconds);
 }

src/universe.cpp

@@ -47,6 +47,8 @@ using ftl::net::internal::SocketServer;
 using ftl::net::internal::Server_TCP;
 using std::chrono::milliseconds;
 
+constexpr int kDefaultMaxConnections = 10;
+
 namespace ftl {
 namespace net {
 
@@ -81,30 +83,12 @@ Universe::Universe() :
         active_(true),
         this_peer(ftl::protocol::id),
         impl_(new ftl::net::NetImplDetail()),
-        peers_(10),
+        peers_(kDefaultMaxConnections),
         phase_(0),
         periodic_time_(1.0),
         reconnect_attempts_(5),
         thread_(Universe::__start, this) {
     _installBindings();
-
-    // Add an idle timer job to garbage collect peer objects
-    // Note: Important to be a timer job to ensure no other timer jobs are
-    // using the object.
-    // FIXME: Replace use of timer.
-    /*garbage_timer_ = ftl::timer::add(ftl::timer::kTimerIdle10, [this](int64_t ts) {
-        if (garbage_.size() > 0) {
-            UNIQUE_LOCK(net_mutex_,lk);
-            if (ftl::pool.n_idle() == ftl::pool.size()) {
-                if (garbage_.size() > 0) LOG(1) << "Garbage collection";
-                while (garbage_.size() > 0) {
-                    delete garbage_.front();
-                    garbage_.pop_front();
-                }
-            }
-        }
-        return true;
-    });*/
 }
 
 Universe::~Universe() {
@@ -112,6 +96,11 @@ Universe::~Universe() {
     CHECK_EQ(peer_instances_, 0);
 }
 
+void Universe::setMaxConnections(size_t m) {
+    UNIQUE_LOCK(net_mutex_, lk);
+    peers_.resize(m);
+}
+
 size_t Universe::getSendBufferSize(ftl::URI::scheme_t s) {
     // TODO(Nick): Allow these to be configured again.
     switch (s) {
@@ -448,6 +437,19 @@ void Universe::_periodic() {
             i = reconnects_.erase(i);
         }*/
     }
+
+    // Garbage peers may not be needed any more
+    if (garbage_.size() > 0) {
+        UNIQUE_LOCK(net_mutex_, lk);
+        // Only do garbage if processing is idle.
+        if (ftl::pool.n_idle() == ftl::pool.size()) {
+            if (garbage_.size() > 0) DLOG(1) << "Garbage collection";
+            while (garbage_.size() > 0) {
+                garbage_.front().reset();
+                garbage_.pop_front();
+            }
+        }
+    }
 }
 
 void Universe::__start(Universe *u) {

src/universe.hpp

@@ -172,6 +172,9 @@ class Universe {
 
     static inline std::shared_ptr<Universe> getInstance() { return instance_; }
 
+    void setMaxConnections(size_t m);
+    size_t getMaxConnections() const { return peers_.size(); }
+
     // --- Test support -------------------------------------------------------
 
     PeerPtr injectFakePeer(std::unique_ptr<ftl::net::internal::SocketConnection> s);