diff options
| -rw-r--r-- | lib/Socket.cpp | 36 | ||||
| -rw-r--r-- | lib/Socket.h | 25 | ||||
| -rw-r--r-- | lib/edioutput/Transport.cpp | 10 | ||||
| -rw-r--r-- | lib/edioutput/Transport.h | 2 |
4 files changed, 61 insertions, 12 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 = ""; }; } diff --git a/lib/edioutput/Transport.cpp b/lib/edioutput/Transport.cpp index 8ebb9fc..4979e93 100644 --- a/lib/edioutput/Transport.cpp +++ b/lib/edioutput/Transport.cpp @@ -193,7 +193,15 @@ void Sender::write(const AFPacket& af_packet) tcp_dispatchers.at(tcp_dest.get())->write(af_packet); } else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(dest)) { - tcp_senders.at(tcp_dest.get())->sendall(af_packet); + const auto error_stats = tcp_senders.at(tcp_dest.get())->sendall(af_packet); + + if (m_conf.verbose and error_stats.has_seen_new_errors) { + fprintf(stderr, "TCP output %s:%d has %zu reconnects: most recent error: %s\n", + tcp_dest->dest_addr.c_str(), + tcp_dest->dest_port, + error_stats.num_reconnects, + error_stats.last_error.c_str()); + } } else { throw logic_error("EDI destination not implemented"); diff --git a/lib/edioutput/Transport.h b/lib/edioutput/Transport.h index 6a3f229..c62545c 100644 --- a/lib/edioutput/Transport.h +++ b/lib/edioutput/Transport.h @@ -31,11 +31,9 @@ #include "AFPacket.h" #include "PFT.h" #include "Socket.h" -#include <vector> #include <chrono> #include <map> #include <unordered_map> -#include <stdexcept> #include <fstream> #include <cstdint> #include <thread> |