From 9aa011b4b704429d1c3910e00c1450650687ded5 Mon Sep 17 00:00:00 2001 From: David Tarazi Date: Wed, 29 Jul 2026 17:59:25 -0700 Subject: [PATCH 1/2] reduce to basic check --- axiomatic_adapter/src/axiomatic_adapter.cpp | 73 +++++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/axiomatic_adapter/src/axiomatic_adapter.cpp b/axiomatic_adapter/src/axiomatic_adapter.cpp index ceaa88a..b28bb25 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,11 +168,23 @@ 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); if (!error) { + std::cerr << "[Axiomatic DBG] forwarding frame 0x" << std::hex << frame.get_id() << std::dec + << " to socketcan..." << std::endl; receive_callback_(std::make_unique(frame)); + std::cerr << "[Axiomatic DBG] ...forwarded to socketcan" << std::endl; } else { error_callback_(*error); } @@ -198,6 +211,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 @@ -242,6 +290,9 @@ class AxiomaticAdapter::AxiomaticAdapterImpl tcp_io_context_.restart(); tcp_io_context_.run(); + std::cerr << "[Axiomatic DBG] receive: run() done ec=\"" << error_code.message() << "\" bytes=" << bytes_received + << std::endl; + // check for timeout or other errors if (error_code == boost::asio::error::timed_out) { return std::optional("Receive operation timed out"); @@ -411,9 +462,21 @@ 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_); + std::cerr << "[Axiomatic DBG] send: writing " << full_message.size() << " bytes to tcp..." << std::endl; boost::asio::write(tcp_socket_, boost::asio::buffer(full_message.data(), full_message.size())); + std::cerr << "[Axiomatic DBG] send: write done" << std::endl; } 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 +496,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 +583,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_; From 78724645a7e9245a7f9ac20e0b717a12a3bef6c6 Mon Sep 17 00:00:00 2001 From: David Tarazi Date: Wed, 29 Jul 2026 18:08:37 -0700 Subject: [PATCH 2/2] reduce logs --- axiomatic_adapter/src/axiomatic_adapter.cpp | 8 -------- 1 file changed, 8 deletions(-) diff --git a/axiomatic_adapter/src/axiomatic_adapter.cpp b/axiomatic_adapter/src/axiomatic_adapter.cpp index b28bb25..1172e58 100644 --- a/axiomatic_adapter/src/axiomatic_adapter.cpp +++ b/axiomatic_adapter/src/axiomatic_adapter.cpp @@ -181,10 +181,7 @@ class AxiomaticAdapter::AxiomaticAdapterImpl std::optional error = receive(frame); if (!error) { - std::cerr << "[Axiomatic DBG] forwarding frame 0x" << std::hex << frame.get_id() << std::dec - << " to socketcan..." << std::endl; receive_callback_(std::make_unique(frame)); - std::cerr << "[Axiomatic DBG] ...forwarded to socketcan" << std::endl; } else { error_callback_(*error); } @@ -290,9 +287,6 @@ class AxiomaticAdapter::AxiomaticAdapterImpl tcp_io_context_.restart(); tcp_io_context_.run(); - std::cerr << "[Axiomatic DBG] receive: run() done ec=\"" << error_code.message() << "\" bytes=" << bytes_received - << std::endl; - // check for timeout or other errors if (error_code == boost::asio::error::timed_out) { return std::optional("Receive operation timed out"); @@ -470,9 +464,7 @@ class AxiomaticAdapter::AxiomaticAdapterImpl try { // lock out the reconnect teardown for the duration of the write std::lock_guard guard(write_mutex_); - std::cerr << "[Axiomatic DBG] send: writing " << full_message.size() << " bytes to tcp..." << std::endl; boost::asio::write(tcp_socket_, boost::asio::buffer(full_message.data(), full_message.size())); - std::cerr << "[Axiomatic DBG] send: write done" << std::endl; } catch (const std::exception & e) { // flag the link lost; the reception thread owns the actual reconnect connection_lost_ = true;