diff --git a/axiomatic_adapter/src/axiomatic_adapter.cpp b/axiomatic_adapter/src/axiomatic_adapter.cpp index ceaa88a..1172e58 100644 --- a/axiomatic_adapter/src/axiomatic_adapter.cpp +++ b/axiomatic_adapter/src/axiomatic_adapter.cpp @@ -16,6 +16,7 @@ #include #include +#include #include #include @@ -167,6 +168,15 @@ class AxiomaticAdapter::AxiomaticAdapterImpl tcp_receive_thread_ = std::thread([this]() { while (!stop_thread_requested_) { + // a failed send() flags the link as lost; recover it here (on this + // thread, so it never races the read below). reconnect() returns false + // only on shutdown. + if (connection_lost_) { + if (!reconnect()) { + break; + } + } + polymath::socketcan::CanFrame frame = polymath::socketcan::CanFrame(); std::optional error = receive(frame); @@ -198,6 +208,41 @@ class AxiomaticAdapter::AxiomaticAdapterImpl return false; } + // Tear down a broken socket and retry openSocket() every 100 ms until it + // reconnects or shutdown is requested. Runs ONLY on the reception thread, so + // it never races receive(). Returns false if a stop was requested first. + bool reconnect() + { + std::cerr << "[Axiomatic] Connection lost — reconnecting to " << ip_address_ << ":" << port_ << "..." << std::endl; + + // force any in-flight send() to unblock and error out so it releases + // write_mutex_; without this a send() blocked on a wedged socket would + // deadlock the teardown below + ::shutdown(tcp_socket_.native_handle(), SHUT_RDWR); + + while (!stop_thread_requested_) { + { + // serialize the teardown against a concurrent send() on the other thread + std::lock_guard guard(write_mutex_); + std::cerr << "[Axiomatic DBG] reconnect: closing socket..." << std::endl; + closeSocket(); + std::cerr << "[Axiomatic DBG] reconnect: closed" << std::endl; + } + + std::cerr << "[Axiomatic DBG] reconnect: attempting openSocket()..." << std::endl; + if (openSocket()) { + connection_lost_ = false; + std::cerr << "[Axiomatic] Reconnected to " << ip_address_ << ":" << port_ << std::endl; + return true; + } + + std::cerr << "[Axiomatic DBG] reconnect: openSocket() failed, retrying in " << RECONNECT_RETRY_INTERVAL_MS.count() + << " ms" << std::endl; + std::this_thread::sleep_for(RECONNECT_RETRY_INTERVAL_MS); + } + return false; + } + std::optional receive(polymath::socketcan::CanFrame & can_frame) { // A previous TCP read may have decoded several CAN frames out of a single @@ -411,9 +456,19 @@ class AxiomaticAdapter::AxiomaticAdapterImpl // insert the can frame data full_message.insert(full_message.end(), frame_data.begin(), frame_data.end()); + // don't touch the socket while the reception thread is reconnecting it + if (connection_lost_) { + return std::optional("TCP Send Failed: connection lost (reconnecting)"); + } + try { + // lock out the reconnect teardown for the duration of the write + std::lock_guard guard(write_mutex_); boost::asio::write(tcp_socket_, boost::asio::buffer(full_message.data(), full_message.size())); } catch (const std::exception & e) { + // flag the link lost; the reception thread owns the actual reconnect + connection_lost_ = true; + std::cerr << "[Axiomatic DBG] send: write threw: " << e.what() << std::endl; return std::optional(std::string("TCP Send Failed: ") + e.what()); } return std::nullopt; @@ -433,6 +488,9 @@ class AxiomaticAdapter::AxiomaticAdapterImpl static constexpr std::array AXIOMATIC_CAN_MESSAGE_HEADER = {'A', 'X', 'I', 'O', 0xBA, 0x36, 0x01}; static constexpr std::chrono::milliseconds TCP_IP_CONNECTION_TIMEOUT_MS{3000}; + // after a send() fails, retry openSocket() this often until reconnected + static constexpr std::chrono::milliseconds RECONNECT_RETRY_INTERVAL_MS{100}; + // receive buffer size for each async_receive call. Larger than the // protocol's per-message cap (256 bytes) by a wide margin static constexpr size_t RECEIVE_BUFFER_SIZE = 65536; @@ -517,6 +575,13 @@ class AxiomaticAdapter::AxiomaticAdapterImpl boost::asio::ip::tcp::socket tcp_socket_; TCPSocketState socket_state_{TCPSocketState::CLOSED}; + // set by send() when a write fails; the reception thread reads it, reconnects, + // and clears it. Atomic because send() runs on a different thread. + std::atomic connection_lost_{false}; + + // guards send()'s write against the reception thread's reconnect teardown + std::mutex write_mutex_; + std::thread tcp_receive_thread_; std::atomic thread_running_; std::atomic stop_thread_requested_;