streaming/detail/tcp/Client.cpp

Go to the documentation of this file.

// Copyright (c) 2026 Computer Vision Center (CVC) at the Universitat Autonoma
// de Barcelona (UAB).
//
// This work is licensed under the terms of the MIT license.
// For a copy, see <https://opensource.org/licenses/MIT>.
 
#include "carla/streaming/detail/tcp/Client.h"
 
#include "carla/BufferPool.h"
#include "carla/Debug.h"
#include "carla/Exception.h"
#include "carla/Logging.h"
#include "carla/Time.h"
 
#include <boost/asio/connect.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/read.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/bind_executor.hpp>
 
#include <exception>
 
namespace carla {
namespace streaming {
namespace detail {
namespace tcp {
 
  // ===========================================================================
  // -- IncomingMessage --------------------------------------------------------
  // ===========================================================================
 
  /// Helper for reading incoming TCP messages. Allocates the whole message in
  /// a single buffer.
  class IncomingMessage {
  public:
 
    explicit IncomingMessage(Buffer &&buffer) : _message(std::move(buffer)) {}
 
    boost::asio::mutable_buffer size_as_buffer() {
      return boost::asio::buffer(&_size, sizeof(_size));
    }
 
    boost::asio::mutable_buffer buffer() {
      DEBUG_ASSERT(_size > 0u);
      _message.reset(_size);
      return _message.buffer();
    }
 
    auto size() const {
      return _size;
    }
 
    auto pop() {
      return std::move(_message);
    }
 
  private:
 
    message_size_type _size = 0u;
 
    Buffer _message;
  };
 
  // ===========================================================================
  // -- Client -----------------------------------------------------------------
  // ===========================================================================
 
  Client::Client(
      boost::asio::io_context &io_context,
      const token_type &token,
      callback_function_type callback)
    : LIBCARLA_INITIALIZE_LIFETIME_PROFILER(
          std::string("tcp client ") + std::to_string(token.get_stream_id())),
      _token(token),
      _callback(std::move(callback)),
      _socket(io_context),
      _strand(io_context),
      _connection_timer(io_context),
      _buffer_pool(std::make_shared<BufferPool>()) {
    if (!_token.protocol_is_tcp()) {
      throw_exception(std::invalid_argument("invalid token, only TCP tokens supported"));
    }
  }
 
  Client::~Client() = default;
 
  void Client::Connect() {
    auto self = shared_from_this();
      if (_done) {
        return;
      }
 
      using boost::system::error_code;
 
      if (_socket.is_open()) {
        _socket.close();
      }
 
      DEBUG_ASSERT(_token.is_valid());
      DEBUG_ASSERT(_token.protocol_is_tcp());
      const auto ep = _token.to_tcp_endpoint();
 
      auto handle_connect = [this, self, ep](error_code ec) {
        if (!ec) {
          if (_done) {
            return;
          }
          // This forces not using Nagle's algorithm.
          // Improves the sync mode velocity on Linux by a factor of ~3.
          _socket.set_option(boost::asio::ip::tcp::no_delay(true));
          log_debug("streaming client: connected to", ep);
          // Send the stream id to subscribe to the stream.
          const auto &stream_id = _token.get_stream_id();
          log_debug("streaming client: sending stream id", stream_id);
          boost::asio::async_write(
              _socket,
              boost::asio::buffer(&stream_id, sizeof(stream_id)),
              boost::asio::bind_executor(_strand, [=](error_code ec, size_t DEBUG_ONLY(bytes)) {
                // Ensures to stop the execution once the connection has been stopped.
                if (_done) {
                  return;
                }
                if (!ec) {
                  DEBUG_ASSERT_EQ(bytes, sizeof(stream_id));
                  // If succeeded start reading data.
                  ReadData();
                } else {
                  // Else try again.
                  log_debug("streaming client: failed to send stream id:", ec.message());
                  Connect();
                }
              }));
        } else {
          log_info("streaming client: connection failed:", ec.message());
          Reconnect();
        }
      };
 
      log_debug("streaming client: connecting to", ep);
      _socket.async_connect(ep, boost::asio::bind_executor(_strand, handle_connect));
  }
 
  void Client::Stop() {
    _connection_timer.cancel();
    auto self = shared_from_this();
    // Serialize the shutdown with the completion handlers: every read/connect
    // handler runs on _strand, so mutating _done and closing the socket from
    // the caller's thread (game thread via sensor destruction) races a
    // concurrent async_read initiation on an io_context worker -- asio sockets
    // are not thread-safe, and the race crashed the client inside ReadData
    // when switching sensors. self keeps the client alive until this runs.
    boost::asio::post(_strand, [this, self]() {
      _done = true;
      if (_socket.is_open()) {
        _socket.close();
      }
    });
  }
 
  void Client::Reconnect() {
    auto self = shared_from_this();
    _connection_timer.expires_after(time_duration::seconds(1u).to_chrono());
    _connection_timer.async_wait([this, self](boost::system::error_code ec) {
      if (!ec) {
        Connect();
      }
    });
  }
 
  void Client::ReadData() {
    auto self = shared_from_this();
      if (_done) {
        return;
      }
 
      // log_debug("streaming client: Client::ReadData");
 
      auto message = std::make_shared<IncomingMessage>(_buffer_pool->Pop());
 
      auto handle_read_data = [this, self, message](boost::system::error_code ec, size_t DEBUG_ONLY(bytes)) {
        DEBUG_ONLY(log_debug("streaming client: Client::ReadData.handle_read_data", bytes, "bytes"));
        // A read that completed while Stop() was being processed must not
        // deliver its payload: the subscriber that _callback points into is
        // already being torn down (sensor destruction), and calling into it
        // is a use-after-free (observed as a manual_control segfault when
        // switching sensors).
        if (_done) {
          return;
        }
        if (!ec) {
          DEBUG_ASSERT_EQ(bytes, message->size());
          DEBUG_ASSERT_NE(bytes, 0u);
          // Move the buffer to the callback function and start reading the next
          // piece of data.
          // log_debug("streaming client: success reading data, calling the callback");
          // An exception leaving this handler would escape io_context::run()
          // on a worker thread and call std::terminate, killing the client
          // process; log it and keep reading the stream instead.
#ifndef LIBCARLA_NO_EXCEPTIONS
          try {
#endif // LIBCARLA_NO_EXCEPTIONS
            _callback(message->pop());
#ifndef LIBCARLA_NO_EXCEPTIONS
          } catch (const std::exception &e) {
            log_error("streaming client: exception in stream callback:", e.what());
          }
#endif // LIBCARLA_NO_EXCEPTIONS
          ReadData();
        } else {
          // As usual, if anything fails start over from the very top.
          log_debug("streaming client: failed to read data:", ec.message());
          Connect();
        }
      };
 
      auto handle_read_header = [this, self, message, handle_read_data](
          boost::system::error_code ec,
          size_t DEBUG_ONLY(bytes)) {
        DEBUG_ONLY(log_debug("streaming client: Client::ReadData.handle_read_header", bytes, "bytes"));
        if (!ec && (message->size() > 0u)) {
          DEBUG_ASSERT_EQ(bytes, sizeof(message_size_type));
          if (_done) {
            return;
          }
          // Now that we know the size of the coming buffer, we can allocate our
          // buffer and start putting data into it.
          boost::asio::async_read(
              _socket,
              message->buffer(),
              boost::asio::bind_executor(_strand, handle_read_data));
        } else if (!_done) {
          log_debug("streaming client: failed to read header:", ec.message());
          DEBUG_ONLY(log_debug("size  = ", message->size()));
          DEBUG_ONLY(log_debug("bytes = ", bytes));
          Connect();
        }
      };
 
      // Read the size of the buffer that is coming.
      boost::asio::async_read(
          _socket,
          message->size_as_buffer(),
          boost::asio::bind_executor(_strand, handle_read_header));
  }
 
} // namespace tcp
} // namespace detail
} // namespace streaming
} // namespace carla