rpc/Server.h

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>.
 
#pragma once
 
#include "carla/MoveHandler.h"
#include "carla/Time.h"
#include "carla/rpc/Metadata.h"
#include "carla/rpc/Response.h"
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
 
#include <rpc/server.h>
 
#include <chrono>
#include <atomic>
#include <future>
#include <stdexcept>
 
namespace carla {
namespace rpc {
 
  // ===========================================================================
  // -- Server -----------------------------------------------------------------
  // ===========================================================================
 
  /// An RPC server in which functions can be bind to run synchronously or
  /// asynchronously.
  ///
  /// Use `AsyncRun` to start the worker threads, and use `SyncRunFor` to
  /// run a slice of work in the caller's thread.
  ///
  /// Functions that are bind using `BindAsync` will run asynchronously in the
  /// worker threads. Functions that are bind using `BindSync` will run within
  /// `SyncRunFor` function.
  class Server {
  public:
 
    template <typename... Args>
    explicit Server(Args &&... args);
 
    template <typename FunctorT>
    void BindSync(const std::string &name, FunctorT &&functor);
 
    template <typename FunctorT>
    void BindAsync(const std::string &name, FunctorT &&functor);
 
    void AsyncRun(size_t worker_threads) {
      _server.async_run(worker_threads);
    }
 
    void SyncRunFor(time_duration duration) {
      #ifdef LIBCARLA_INCLUDED_FROM_UE4
      #include <util/enable-ue4-macros.h>
      TRACE_CPUPROFILER_EVENT_SCOPE_STR(__FUNCTION__);
      #include <util/disable-ue4-macros.h>
      #endif // LIBCARLA_INCLUDED_FROM_UE4
      _sync_io_context.restart();
      _sync_io_context.run_for(duration.to_chrono());
    }
 
    /// @warning does not stop the game thread.
    void Stop() {
      _shutdown_in_progress = true;
      _server.stop();
    }
 
  private:
 
    boost::asio::io_context _sync_io_context;
 
    ::rpc::server _server;
    std::atomic_bool _shutdown_in_progress{false};
  };
 
  // ===========================================================================
  // -- Server implementation --------------------------------------------------
  // ===========================================================================
 
namespace detail {
 
  template <typename T>
  struct FunctionWrapper : FunctionWrapper<decltype(&T::operator())> {};
 
  template <typename C, typename R, typename... Args>
  struct FunctionWrapper<R (C::*)(Args...)> : FunctionWrapper<R (*)(Args...)> {};
 
  template <typename C, typename R, typename... Args>
  struct FunctionWrapper<R (C::*)(Args...) const> : FunctionWrapper<R (*)(Args...)> {};
 
  template<class T>
  struct FunctionWrapper<T &> : public FunctionWrapper<T> {};
 
  template<class T>
  struct FunctionWrapper<T &&> : public FunctionWrapper<T> {};
 
  template <typename R, typename... Args>
  struct FunctionWrapper<R (*)(Args...)> {
 
    /// Wraps @a functor into a function type with equivalent signature. The
    /// wrap function returned. When called, posts @a functor into the
    /// io_context; if the client called this method synchronously, waits for
    /// the posted task to finish, otherwise returns immediately.
    ///
    /// This way, no matter from which thread the wrap function is called, the
    /// @a functor provided is always called from the context of the io_context.
    /// I.e., we can use the io_context to run tasks on a specific thread (e.g.
    /// game thread).
    template <typename FuncT>
    static auto WrapSyncCall(std::atomic_bool &shutdown_in_progress, boost::asio::io_context &io, FuncT &&functor) {
      return [&shutdown_in_progress, &io, functor=std::forward<FuncT>(functor)](Metadata metadata, Args... args) -> R {
        auto task = std::packaged_task<R()>([functor=std::move(functor), args...]() {
          return functor(args...);
        });
        if (metadata.IsResponseIgnored()) {
          // Post task and ignore result.
          boost::asio::post(io, MoveHandler(task));
          return R();
        } else {
          // Post task and wait for result.
          auto result = task.get_future();
          boost::asio::post(io, MoveHandler(task));
          std::future_status status;
          do {
            status = result.wait_for(std::chrono::milliseconds(100));
          } while (!shutdown_in_progress && (status != std::future_status::ready));
          if (status == std::future_status::ready) {
            return result.get();
          } else {
            throw std::runtime_error("RPC server is shutting down");
          }
        }
      };
    }
 
    /// Wraps @a functor into a function type with equivalent signature that
    /// handles the metadata sent by the client. If the client called this
    /// method asynchronously, the result is ignored.
    template <typename FuncT>
    static auto WrapAsyncCall(FuncT &&functor) {
      return [functor=std::forward<FuncT>(functor)](::carla::rpc::Metadata metadata, Args... args) -> R {
        if (metadata.IsResponseIgnored()) {
          functor(args...);
          return R();
        } else {
          return functor(args...);
        }
      };
    }
  };
 
} // namespace detail
 
  template <typename ... Args>
  inline Server::Server(Args && ... args)
    : _server(std::forward<Args>(args) ...) {
    _server.suppress_exceptions(true);
  }
 
  template <typename FunctorT>
  inline void Server::BindSync(const std::string &name, FunctorT &&functor) {
    using Wrapper = detail::FunctionWrapper<FunctorT>;
    _server.bind(
        name,
        Wrapper::WrapSyncCall(_shutdown_in_progress, _sync_io_context, std::forward<FunctorT>(functor)));
  }
 
  template <typename FunctorT>
  inline void Server::BindAsync(const std::string &name, FunctorT &&functor) {
    using Wrapper = detail::FunctionWrapper<FunctorT>;
    _server.bind(
        name,
        Wrapper::WrapAsyncCall(std::forward<FunctorT>(functor)));
  }
 
} // namespace rpc
} // namespace carla