Extend ROS2 support Step 2: Fine grained ServerSynchronization

CHANGELOG.md

@@ -8,6 +8,8 @@
  * Fixed geom::Rotation::RotateVector() rotation directions of pitch and roll
  * Prepare server for multistream support and ROS2 client calls
  * Improved V2X sensor capabilities: send complex custom user-defined data, support V2I sensors not attached to a vehicle
+ * Introduced fine grained ServerSynchronization mechanism: each client decides for its own if it requires synchronization or not and provides its own synchronization window.
+   Be aware: some existing code using master/slave sync mechanism might need rework. See also generate_traffic.py.
 
 ## CARLA 0.9.16
 

LibCarla/source/carla/client/World.cpp

@@ -75,7 +75,14 @@ namespace client {
         if (tics_correct >= 2)
           return id;
 
-        Tick(local_timeout);
+        if (settings.synchronous_mode) {
+          // tick if synchronous mode is active
+          Tick(local_timeout);
+        }
+        else {
+          WaitForTick(local_timeout);
+        } 
+
       }
 
       log_warning("World::ApplySettings: After", number_of_attemps, " attemps, the settings were not correctly set. Please check that everything is consistent.");

LibCarla/source/carla/rpc/RpcServerInterface.h

@@ -14,6 +14,7 @@
 #include "carla/rpc/MapInfo.h"
 #include "carla/rpc/MapLayer.h"
 #include "carla/rpc/Response.h"
+#include "carla/rpc/ServerSynchronizationTypes.h"
 #include "carla/rpc/Transform.h"
 #include "carla/rpc/VehicleTelemetryData.h"
 #include "carla/streaming/detail/Dispatcher.h"
@@ -80,6 +81,27 @@ class RpcServerInterface {
   /**
    * @}
    */
+
+  /**
+   * @brief synchronization calls
+   * @{
+   */
+  virtual Response<uint64_t> call_tick(
+      synchronization_client_id_type const &client_id,
+      synchronization_participant_id_type const &participant_id,
+      carla::rpc::SynchronizationTickMode synchronization_tick_mode) = 0;
+  virtual Response<synchronization_participant_id_type> call_register_synchronization_participant(
+      synchronization_client_id_type const &client_id,
+      synchronization_participant_id_type const &participant_id_hint = ALL_PARTICIPANTS) = 0;
+  virtual Response<bool> call_deregister_synchronization_participant(
+      synchronization_client_id_type const &client_id, synchronization_participant_id_type const &participant_id) = 0;
+  virtual Response<bool> call_update_synchronization_window(
+      synchronization_client_id_type const &client_id, synchronization_participant_id_type const &participant_id,
+      synchronization_target_game_time const &target_game_time = NO_SYNC_TARGET_GAME_TIME) = 0;
+  virtual carla::rpc::Response<std::pair< bool , std::vector<carla::rpc::synchronization_window_participant_state> > >  call_get_synchronization_window_status() = 0;
+  /**
+   * @}
+   */
 };
 
 }  // namespace rpc

LibCarla/source/carla/rpc/Server.h

@@ -15,6 +15,7 @@
 #include <boost/asio/post.hpp>
 
 #include <rpc/server.h>
+#include <rpc/this_session.h>
 
 #include <future>
 
@@ -65,6 +66,14 @@ namespace rpc {
       _server.stop();
     }
 
+    void BindOnClientConnected(::rpc::server::callback_type callback) {
+      _server.set_on_connection(callback);
+    }
+
+    void BindOnClientDisconnected(::rpc::server::callback_type callback) {
+      _server.set_on_disconnection(callback);
+    }
+
   private:
 
     boost::asio::io_context _sync_io_context;
@@ -108,7 +117,9 @@ namespace detail {
     template <typename FuncT>
     static auto WrapSyncCall(boost::asio::io_context &io, FuncT &&functor) {
       return [&io, functor=std::forward<FuncT>(functor)](Metadata metadata, Args... args) -> R {
-        auto task = std::packaged_task<R()>([functor=std::move(functor), args...]() {
+        auto const session_id = ::rpc::this_session().id();
+        auto task = std::packaged_task<R()>([session_id, functor=std::move(functor), args...]() {
+          ::rpc::this_session().set_id(session_id);
           return functor(args...);
         });
         if (metadata.IsResponseIgnored()) {

LibCarla/source/carla/rpc/ServerSynchronizationTypes.h

@@ -0,0 +1,36 @@
+// Copyright (c) 2024 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 <locale>
+
+namespace carla {
+namespace rpc {
+
+using synchronization_client_id_type = std::string;
+static const carla::rpc::synchronization_client_id_type ALL_CLIENTS{};
+
+using synchronization_participant_id_type = uint32_t;
+static constexpr carla::rpc::synchronization_participant_id_type ALL_PARTICIPANTS{0};
+
+using synchronization_target_game_time = double;
+static constexpr synchronization_target_game_time NO_SYNC_TARGET_GAME_TIME{0.};
+static constexpr synchronization_target_game_time BLOCKING_TARGET_GAME_TIME{1e-6};
+
+struct synchronization_window_participant_state {
+    synchronization_client_id_type client_id;
+    synchronization_participant_id_type participant_id;
+    synchronization_target_game_time target_game_time;
+};
+
+enum class SynchronizationTickMode {
+    FORCE_ENABLE_SYNC,
+    TICK_ONLY_IF_SYNC_ENABLED
+};
+
+}  // namespace rpc
+}  // namespace carla

PythonAPI/examples/generate_traffic.py

@@ -148,9 +148,9 @@ def main():
     all_id = []
     client = carla.Client(args.host, args.port)
     client.set_timeout(10.0)
-    synchronous_master = False
     random.seed(args.seed if args.seed is not None else int(time.time()))
 
+    original_world_settings = None
     try:
         world = client.get_world()
 
@@ -164,23 +164,24 @@ def main():
         if args.seed is not None:
             traffic_manager.set_random_device_seed(args.seed)
 
-        settings = world.get_settings()
+        original_world_settings = world.get_settings()
+        print("current_world_settings {}".format(original_world_settings))
+        settings = original_world_settings
         if not args.asynch:
             traffic_manager.set_synchronous_mode(True)
             if not settings.synchronous_mode:
-                synchronous_master = True
                 settings.synchronous_mode = True
                 settings.fixed_delta_seconds = 0.05
-            else:
-                synchronous_master = False
         else:
             print("You are currently in asynchronous mode. If this is a traffic simulation, \
             you could experience some issues. If it's not working correctly, switch to synchronous \
             mode by using traffic_manager.set_synchronous_mode(True)")
 
         if args.no_rendering:
             settings.no_rendering_mode = True
+        print("apply_world_settings {}".format(settings))
         world.apply_settings(settings)
+        print("settings applied")
 
         blueprints = get_actor_blueprints(world, args.filterv, args.generationv)
         if not blueprints:
@@ -229,7 +230,7 @@ def main():
             batch.append(SpawnActor(blueprint, transform)
                 .then(SetAutopilot(FutureActor, True, traffic_manager.get_port())))
 
-        for response in client.apply_batch_sync(batch, synchronous_master):
+        for response in client.apply_batch_sync(batch, do_tick=True):
             if response.error:
                 logging.error(response.error)
             else:
@@ -281,7 +282,7 @@ def main():
                 print("Walker has no speed")
                 walker_speed.append(0.0)
             batch.append(SpawnActor(walker_bp, spawn_point))
-        results = client.apply_batch_sync(batch, True)
+        results = client.apply_batch_sync(batch, do_tick=True)
         walker_speed2 = []
         for i in range(len(results)):
             if results[i].error:
@@ -295,7 +296,7 @@ def main():
         walker_controller_bp = world.get_blueprint_library().find('controller.ai.walker')
         for i in range(len(walkers_list)):
             batch.append(SpawnActor(walker_controller_bp, carla.Transform(), walkers_list[i]["id"]))
-        results = client.apply_batch_sync(batch, True)
+        results = client.apply_batch_sync(batch, do_tick=True)
         for i in range(len(results)):
             if results[i].error:
                 logging.error(results[i].error)
@@ -308,7 +309,7 @@ def main():
         all_actors = world.get_actors(all_id)
 
         # wait for a tick to ensure client receives the last transform of the walkers we have just created
-        if args.asynch or not synchronous_master:
+        if args.asynch:
             world.wait_for_tick()
         else:
             world.tick()
@@ -330,18 +331,22 @@ def main():
         traffic_manager.global_percentage_speed_difference(30.0)
 
         while True:
-            if not args.asynch and synchronous_master:
+            if not args.asynch:
                 world.tick()
             else:
                 world.wait_for_tick()
 
     finally:
 
-        if not args.asynch and synchronous_master:
-            settings = world.get_settings()
-            settings.synchronous_mode = False
-            settings.no_rendering_mode = False
-            settings.fixed_delta_seconds = None
+        if not args.asynch:
+            if original_world_settings:
+                settings= original_world_settings
+            else:
+                settings = world.get_settings()
+                settings.synchronous_mode = False
+                settings.no_rendering_mode = False
+                settings.fixed_delta_seconds = None
+            print("restore world_settings {}".format(settings))
             world.apply_settings(settings)
 
         print('\ndestroying %d vehicles' % len(vehicles_list))

PythonAPI/test/smoke/__init__.py

@@ -48,6 +48,8 @@ def setUp(self):
 
     def tearDown(self):
         self.world.apply_settings(self.settings)
-        self.world.tick()
+        if self.settings.synchronous_mode:
+            # tick if synchronous mode is active
+            self.world.tick()
         self.settings = None
         super(SyncSmokeTest, self).tearDown()

Unreal/CarlaUE4/Plugins/Carla/Source/Carla/Game/CarlaEngine.cpp

@@ -202,7 +202,7 @@ void FCarlaEngine::NotifyInitGame(const UCarlaSettings &Settings)
       Secondary = std::make_shared<carla::multigpu::Secondary>(PrimaryIP, PrimaryPort, CommandExecutor);
       Secondary->Connect();
       // set this server in synchronous mode
-      bSynchronousMode = true;
+      Server.EnableSynchronousMode();
     }
     else
     {
@@ -281,21 +281,24 @@ void FCarlaEngine::OnPreTick(UWorld *, ELevelTick TickType, float DeltaSeconds)
 
     if (bIsPrimaryServer)
     {
-      if (CurrentEpisode && !bSynchronousMode && SecondaryServer->HasClientsConnected())
-      {
-        // set synchronous mode
-        CurrentSettings.bSynchronousMode = true;
-        CurrentSettings.FixedDeltaSeconds = 1 / 20.0f;
-        OnEpisodeSettingsChanged(CurrentSettings);
-        CurrentEpisode->ApplySettings(CurrentSettings);
-      }
-
       // process RPC commands
       do
       {
         Server.RunSome(1u);
       }
-      while (bSynchronousMode && !Server.TickCueReceived());
+      while (Server.IsSynchronousModeActive() && !Server.TickCueReceived());
+
+      if ( (CurrentEpisode && !Server.IsSynchronousModeActive() && SecondaryServer->HasClientsConnected())
+          || ( Server.IsSynchronousModeActive() && (!CurrentSettings.FixedDeltaSeconds || !CurrentSettings.bSynchronousMode) ) )
+      {
+        // ensure the delta seconds are also considered in this run
+        DeltaSeconds = Server.GetTickDeltaSeconds();
+
+        CurrentSettings.bSynchronousMode = true;
+        CurrentSettings.FixedDeltaSeconds = DeltaSeconds;
+        OnEpisodeSettingsChanged(CurrentSettings);
+        CurrentEpisode->ApplySettings(CurrentSettings);
+      }
     }
     else
     {
@@ -382,7 +385,12 @@ void FCarlaEngine::OnEpisodeSettingsChanged(const FEpisodeSettings &Settings)
 {
   CurrentSettings = FEpisodeSettings(Settings);
 
-  bSynchronousMode = Settings.bSynchronousMode;
+  if (Settings.bSynchronousMode && !Server.IsSynchronousModeActive()) {
+    Server.EnableSynchronousMode();
+  }
+  else if (!Settings.bSynchronousMode && Server.IsSynchronousModeActive()) {
+    Server.DisableSynchronousMode();
+  }
 
   if (GEngine && GEngine->GameViewport)
   {

Unreal/CarlaUE4/Plugins/Carla/Source/Carla/Game/CarlaEngine.h

@@ -105,8 +105,6 @@ class FCarlaEngine : private NonCopyable
 
   bool bIsRunning = false;
 
-  bool bSynchronousMode = false;
-
   bool bMapChanged = false;
 
   FCarlaServer Server;