Socket TCPClient: do reconnect for half-closed connections

2 files changed, 52 insertions, 9 deletions

diff --git a/lib/Socket.cpp b/lib/Socket.cpp
index 2df1559..938b573 100644
--- a/lib/Socket.cpp
+++ b/lib/Socket.cpp
@@ -478,7 +478,7 @@ TCPSocket::~TCPSocket()
 
 TCPSocket::TCPSocket(TCPSocket&& other) :
     m_sock(other.m_sock),
-    m_remote_address(move(other.m_remote_address))
+    m_remote_address(std::move(other.m_remote_address))
 {
     if (other.m_sock != -1) {
         other.m_sock = -1;
@@ -967,12 +967,22 @@ ssize_t TCPClient::recv(void *buffer, size_t length, int flags, int timeout_ms)
             reconnect();
         }
 
+        m_last_received_packet_ts = chrono::steady_clock::now();
+
         return ret;
     }
     catch (const TCPSocket::Interrupted&) {
         return -1;
     }
     catch (const TCPSocket::Timeout&) {
+        const auto timeout = chrono::milliseconds(timeout_ms * 5);
+        if (m_last_received_packet_ts.has_value() and
+            chrono::steady_clock::now() - *m_last_received_packet_ts > timeout)
+        {
+            // This is to catch half-closed TCP connections
+            reconnect();
+        }
+
         return 0;
     }
 
@@ -983,6 +993,7 @@ void TCPClient::reconnect()
 {
     TCPSocket newsock;
     m_sock = std::move(newsock);
+    m_last_received_packet_ts = nullopt;
     m_sock.connect(m_hostname, m_port, true);
 }
 
@@ -990,7 +1001,7 @@ TCPConnection::TCPConnection(TCPSocket&& sock) :
             queue(),
             m_running(true),
             m_sender_thread(),
-            m_sock(move(sock))
+            m_sock(std::move(sock))
 {
 #if MISSING_OWN_ADDR
     auto own_addr = m_sock.getOwnAddress();
@@ -1109,7 +1120,7 @@ void TCPDataDispatcher::process()
             auto sock = m_listener_socket.accept(timeout_ms);
             if (sock.valid()) {
                 auto lock = unique_lock<mutex>(m_mutex);
-                m_connections.emplace(m_connections.begin(), move(sock));
+                m_connections.emplace(m_connections.begin(), std::move(sock));
 
                 if (m_buffers_to_preroll > 0) {
                     for (const auto& buf : m_preroll_queue) {
@@ -1181,7 +1192,7 @@ void TCPReceiveServer::process()
                 }
                 else {
                     buf.resize(r);
-                    m_queue.push(make_shared<TCPReceiveMessageData>(move(buf)));
+                    m_queue.push(make_shared<TCPReceiveMessageData>(std::move(buf)));
                 }
             }
             catch (const TCPSocket::Interrupted&) {
@@ -1222,7 +1233,7 @@ TCPSendClient::~TCPSendClient()
     }
 }
 
-void TCPSendClient::sendall(const std::vector<uint8_t>& buffer)
+TCPSendClient::ErrorStats TCPSendClient::sendall(const std::vector<uint8_t>& buffer)
 {
     if (not m_running) {
         throw runtime_error(m_exception_data);
@@ -1234,6 +1245,17 @@ void TCPSendClient::sendall(const std::vector<uint8_t>& buffer)
         vector<uint8_t> discard;
         m_queue.try_pop(discard);
     }
+
+    TCPSendClient::ErrorStats es;
+    es.num_reconnects = m_num_reconnects.load();
+
+    es.has_seen_new_errors = es.num_reconnects != m_num_reconnects_prev;
+    m_num_reconnects_prev = es.num_reconnects;
+
+    auto lock = unique_lock<mutex>(m_error_mutex);
+    es.last_error = m_last_error;
+
+    return es;
 }
 
 void TCPSendClient::process()
@@ -1255,12 +1277,16 @@ void TCPSendClient::process()
             }
             else {
                 try {
+                    m_num_reconnects.fetch_add(1, std::memory_order_seq_cst);
                     m_sock.connect(m_hostname, m_port);
                     m_is_connected = true;
                 }
                 catch (const runtime_error& e) {
                     m_is_connected = false;
                     this_thread::sleep_for(chrono::seconds(1));
+
+                    auto lock = unique_lock<mutex>(m_error_mutex);
+                    m_last_error = e.what();
                 }
             }
         }
diff --git a/lib/Socket.h b/lib/Socket.h
index 1320a64..7709145 100644
--- a/lib/Socket.h
+++ b/lib/Socket.h
@@ -31,9 +31,11 @@
 #include "ThreadsafeQueue.h"
 #include <cstdlib>
 #include <atomic>
-#include <string>
+#include <chrono>
 #include <list>
 #include <memory>
+#include <optional>
+#include <string>
 #include <thread>
 #include <vector>
 
@@ -236,6 +238,8 @@ class TCPClient {
         TCPSocket m_sock;
         std::string m_hostname;
         int m_port;
+
+        std::optional<std::chrono::steady_clock::time_point> m_last_received_packet_ts;
 };
 
 /* Helper class for TCPDataDispatcher, contains a queue of pending data and
@@ -329,10 +333,18 @@ class TCPSendClient {
     public:
         TCPSendClient(const std::string& hostname, int port);
         ~TCPSendClient();
+        TCPSendClient(const TCPSendClient&) = delete;
+        TCPSendClient& operator=(const TCPSendClient&) = delete;
 
-        /* Throws a runtime_error on error
-         */
-        void sendall(const std::vector<uint8_t>& buffer);
+
+        struct ErrorStats {
+            std::string last_error = "";
+            size_t num_reconnects = 0;
+            bool has_seen_new_errors = false;
+        };
+
+        /* Throws a runtime_error when the process thread isn't running */
+        ErrorStats sendall(const std::vector<uint8_t>& buffer);
 
     private:
         void process();
@@ -349,6 +361,11 @@ class TCPSendClient {
         std::string m_exception_data;
         std::thread m_sender_thread;
         TCPSocket m_listener_socket;
+
+        std::atomic<size_t> m_num_reconnects = ATOMIC_VAR_INIT(0);
+        size_t m_num_reconnects_prev = 0;
+        std::mutex m_error_mutex;
+        std::string m_last_error = "";
 };
 
 }