secondary.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/multigpu/incomingMessage.h"
#include "carla/multigpu/secondary.h"
 
#include "carla/multigpu/commands.h"
#include "carla/multigpu/keepAlive.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/read.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/bind_executor.hpp>
 
#include <exception>
 
namespace carla {
namespace multigpu {
 
  Secondary::Secondary(
    boost::asio::ip::tcp::endpoint ep,
    SecondaryCommands::callback_type callback) :
      _pool(),
      _socket(_pool.io_context()),
      _endpoint(ep),
      _strand(_pool.io_context()),
      _connection_timer(_pool.io_context()),
      _buffer_pool(std::make_shared<BufferPool>()) {
 
      _commander.set_callback(callback);
    }
 
 
  Secondary::Secondary(
    std::string ip,
    uint16_t port,
    SecondaryCommands::callback_type callback) :
      _pool(),
      _socket(_pool.io_context()),
      _strand(_pool.io_context()),
      _connection_timer(_pool.io_context()),
      _buffer_pool(std::make_shared<BufferPool>()) {
 
    boost::asio::ip::address ip_address = boost::asio::ip::make_address(ip);
    _endpoint = boost::asio::ip::tcp::endpoint(ip_address, port);
    _commander.set_callback(callback);
  }
 
  Secondary::~Secondary() {
    _pool.Stop();
  }
 
  void Secondary::Connect() {
    std::call_once(_run_flag, [this]() { AsyncRun(2u); });
 
    _commander.set_secondary(shared_from_this());
 
    std::weak_ptr<Secondary> weak = shared_from_this();
    boost::asio::post(_strand, [weak]() {
      auto self = weak.lock();
      if (!self) return;
 
      if (self->_done) {
        return;
      }
 
      if (self->_socket.is_open()) {
        // As Disconnect(): what is queued belongs to the connection closed here.
        self->Disconnect();
      }
 
      auto handle_connect = [weak](boost::system::error_code ec) {
        auto self = weak.lock();
        if (!self) return;
        if (ec) {
          log_error("secondary server: connection failed:", ec.message());
          if (!self->_done)
            self->Reconnect();
          return;
        }
 
        if (self->_done) {
          return;
        }
 
        // This forces not using Nagle's algorithm.
        // Improves the sync mode velocity on Linux by a factor of ~3.
        self->_socket.set_option(boost::asio::ip::tcp::no_delay(true));
        EnableKeepAlive(self->_socket);
        self->_connected = true;
 
        log_info("secondary server: connected to ", self->_endpoint);
 
        self->ReadData();
      };
 
      self->_socket.async_connect(self->_endpoint, boost::asio::bind_executor(self->_strand, handle_connect));
    });
  }
 
  void Secondary::Stop() {
    _done = true;
    _connected = false;
    // Joins the io threads, so no command callback runs after Stop() returns;
    // the socket and timer are then only touched from this thread.
    _pool.Stop();
    _connection_timer.cancel();
    boost::system::error_code ec;
    _socket.close(ec);
  }
 
  void Secondary::Reconnect() {
    std::weak_ptr<Secondary> weak = shared_from_this();
    _connection_timer.expires_after(time_duration::seconds(1u).to_chrono());
    _connection_timer.async_wait([weak](boost::system::error_code ec) {
      auto self = weak.lock();
      if (!self) return;
      if (!ec) {
        self->Connect();
      }
    });
  }
 
  void Secondary::AsyncRun(size_t worker_threads) {
    _pool.AsyncRun(worker_threads);
  }
 
  void Secondary::Write(std::shared_ptr<const carla::streaming::detail::tcp::Message> message) {
    DEBUG_ASSERT(message != nullptr);
    DEBUG_ASSERT(!message->empty());
    std::weak_ptr<Secondary> weak = shared_from_this();
    boost::asio::post(_strand, [weak, message]() {
      auto self = weak.lock();
      if (!self) return;
      if (!self->_socket.is_open()) {
        return;
      }
      self->_write_queue.push_back(message);
      if (!self->_is_writing) {
        self->WriteNext();
      }
    });
  }
 
  void Secondary::WriteNext() {
    if (_write_queue.empty() || !_socket.is_open()) {
      _write_queue.clear();
      _is_writing = false;
      return;
    }
    _is_writing = true;
    const message_type message = _write_queue.front();
    const uint64_t generation = _write_generation;
    std::weak_ptr<Secondary> weak = shared_from_this();
    auto handle_sent = [weak, message, generation](const boost::system::error_code &ec, size_t) {
      auto self = weak.lock();
      if (!self) return;
      // Disconnect() dropped the queue this write was the front of: what the
      // queue holds now, if anything, came after.
      if (generation != self->_write_generation) {
        return;
      }
      if (ec) {
        log_error("error sending data: ", ec.message());
        self->_write_queue.clear();
        self->_is_writing = false;
        return;
      }
      self->_write_queue.pop_front();
      self->WriteNext();
    };
 
    boost::asio::async_write(
        _socket,
        message->GetBufferSequence(),
        boost::asio::bind_executor(_strand, handle_sent));
  }
 
  void Secondary::Write(Buffer buffer) {
    Write(Secondary::MakeMessage(carla::BufferView::CreateFrom(std::move(buffer))));
  }
 
  void Secondary::Write(std::string text) {
    // One message, size prefix included, through the same ordered queue.
    Write(Buffer(reinterpret_cast<const unsigned char *>(text.data()), text.size()));
  }
 
  void Secondary::Disconnect() {
    _connected = false;
    _write_queue.clear();
    _is_writing = false;
    ++_write_generation;
    boost::system::error_code ignored;
    _socket.close(ignored);
  }
 
  void Secondary::ReadData() {
    std::weak_ptr<Secondary> weak = shared_from_this();
    boost::asio::post(_strand, [weak]() {
      auto self = weak.lock();
      if (!self) return;
      if (self->_done) {
        return;
      }
 
      auto message = std::make_shared<IncomingMessage>(self->_buffer_pool->Pop());
 
      auto handle_read_data = [weak, message](boost::system::error_code ec, size_t DEBUG_ONLY(bytes)) {
        auto self = weak.lock();
        if (!self) 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.
          self->GetCommander().process_command(message->pop());
          self->ReadData();
        } else {
          // The primary is gone. Established sessions are not resumed: close
          // ours, so the primary's side sees it closed too.
          if (!self->_done) {
            log_error("secondary server: failed to read data: ", ec.message());
          }
          self->Disconnect();
        }
      };
 
      auto handle_read_header = [weak, message, handle_read_data](
        boost::system::error_code ec,
        size_t DEBUG_ONLY(bytes)) {
          auto self = weak.lock();
          if (!self) return;
          if (!ec && (message->size() > kMaxMultiGpuMessageBytes)) {
            log_error("secondary server: the primary announced a", message->size(),
                "byte message, larger than any valid one; disconnecting");
            self->Disconnect();
          } else if (!ec && (message->size() > 0u)) {
            DEBUG_ASSERT_EQ(bytes, sizeof(carla::streaming::detail::message_size_type));
            if (self->_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(
                self->_socket,
                message->buffer(),
                boost::asio::bind_executor(self->_strand, handle_read_data));
          } else {
            if (!self->_done) {
              log_error("secondary server: failed to read header: ", ec.message());
            }
            self->Disconnect();
          }
        };
 
      // Read the size of the buffer that is coming.
      boost::asio::async_read(
          self->_socket,
          message->size_as_buffer(),
          boost::asio::bind_executor(self->_strand, handle_read_header));
    });
  }
 
} // namespace multigpu
} // namespace carla