Extend ROS2 support Step 2: Fine grained ServerSynchronization (#9450)

* Extend ROS2 support Step 2: Fine grained ServerSynchronization

Introduced a fine grained ServerSynchronization mechanism, where each
synchonization participant is treated independently and interacts
with the synchronization of the carla-server individually. If a client
is disconnected (or dies) the synchronization state of all participants
registered via that client are dropped, i.e. the server will continue
running in case the participants of that client were the only ones
demanding synchronous mode.

The synchronization interface provides means of a time window, up to
which the server is allowed to run. Like this, every client can prevent
the carla-server to run too fast depending on their individual speed.
There is no sync-master anymore. Every client decides for its own if
it requires synchronization or not.

Drawback of this change: some existing code might have to be changed
(see removal of synchronous_master in generate_traffic.py).

* Fix Windows build and smoke tests

Update RPC lib version for windows build

Ensure tick calls are ignored if sync mode is not active.
And prevent client changes of fixed_delta_seconds triggering warning
message.

* Client needs to wait for next tick

on non synchronous mode

Author: berndgassmann

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;