Implement failure detection to detect and recover from network failures.

fx2y · Oct 5, 2023 · 69c883c · 69c883c
1 parent 9757a14
commit 69c883c
Show file tree

Hide file tree

Showing 3 changed files with 39 additions and 1 deletion.
diff --git a/src/connection_pool.cpp b/src/connection_pool.cpp
@@ -1,6 +1,7 @@
 #include "connection_pool.h"
 
 #include <utility>
+#include <iostream>
 
 /**
  * @brief Constructor for ConnectionPool class.
@@ -68,4 +69,22 @@ void ConnectionPool::return_connection(std::unique_ptr<tcp::socket> socket) {
     }
     // Use the emplace_back method to move the socket to the back of the vector
     connections_.emplace_back(std::move(socket));
-}
+}
+
+void ConnectionPool::detect_failures() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    for (auto &socket: connections_) {
+        boost::system::error_code ec;
+        socket->non_blocking(true);
+        socket->send(boost::asio::buffer("ping"), 0, ec);
+        if (ec) {
+            std::cerr << "Connection failed: " << ec.message() << std::endl;
+            socket->close(ec);
+            socket->connect(endpoint_, ec);
+            if (ec) {
+                std::cerr << "Reconnection failed: " << ec.message() << std::endl;
+            }
+        }
+        socket->non_blocking(false);
+    }
+}
diff --git a/src/connection_pool.h b/src/connection_pool.h
@@ -15,6 +15,8 @@ class ConnectionPool {
 
     void return_connection(std::unique_ptr<tcp::socket> socket);
 
+    void detect_failures();
+
 private:
     boost::asio::io_context &io_context_;
     tcp::endpoint endpoint_;

diff --git a/src/main.cpp b/src/main.cpp
@@ -4,11 +4,28 @@
 
 using boost::asio::ip::tcp;
 
+void
+timer_handler(const boost::system::error_code &ec, boost::asio::steady_timer &timer, ConnectionPool &connection_pool) {
+    if (!ec) {
+        connection_pool.detect_failures();
+        timer.expires_at(timer.expiry() + boost::asio::chrono::seconds(5));
+        timer.async_wait([&](const boost::system::error_code &ec) {
+            timer_handler(ec, timer, connection_pool);
+        });
+    }
+}
+
 int main() {
     boost::asio::io_context io_context;
     tcp::endpoint endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 1234);
     ConnectionPool connection_pool(io_context, endpoint, 10);
 
+    // Start a timer to call detect_failures every 5 seconds
+    boost::asio::steady_timer timer(io_context, boost::asio::chrono::seconds(5));
+    timer.async_wait([&](const boost::system::error_code &ec) {
+        timer_handler(ec, timer, connection_pool);
+    });
+
     try {
         tcp::endpoint local_endpoint(boost::asio::ip::address::from_string("127.0.0.1"), 0);
         std::unique_ptr<tcp::socket> socket_ptr = std::make_unique<tcp::socket>(io_context);