621 configurable ingestion thread count

ChronoGrapher/include/ChronoGrapherConfiguration.h

@@ -11,10 +11,16 @@
 namespace chronolog
 {
 
+struct KeeperGrapherDrainServiceConf
+{
+    RPCProviderConf RPC_CONF;
+    uint32_t INGESTION_THREAD_COUNT = 1;
+};
+
 struct GrapherConfiguration
 {
     uint32_t RECORDING_GROUP{};
-    RPCProviderConf KEEPER_GRAPHER_DRAIN_SERVICE_CONF;
+    KeeperGrapherDrainServiceConf KEEPER_GRAPHER_DRAIN_SERVICE_CONF;
     RPCProviderConf DATA_STORE_ADMIN_SERVICE_CONF;
     RPCProviderConf VISOR_REGISTRY_SERVICE_CONF;
     LogConf LOG_CONF;
@@ -24,10 +30,10 @@ struct GrapherConfiguration
     GrapherConfiguration()
     {
         RECORDING_GROUP = 0;
-        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.PROTO_CONF = "ofi+sockets";
-        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.IP = "127.0.0.1";
-        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.BASE_PORT = 9999;
-        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.SERVICE_PROVIDER_ID = 99;
+        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.PROTO_CONF = "ofi+sockets";
+        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.IP = "127.0.0.1";
+        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.BASE_PORT = 9999;
+        KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID = 99;
 
         DATA_STORE_ADMIN_SERVICE_CONF.PROTO_CONF = "ofi+sockets";
         DATA_STORE_ADMIN_SERVICE_CONF.IP = "127.0.0.1";
@@ -51,7 +57,7 @@ struct GrapherConfiguration
     [[nodiscard]] std::string to_String() const
     {
         return "[CHRONO_GRAPHER_CONFIGURATION: RECORDING_GROUP: " + std::to_string(RECORDING_GROUP) +
-               ", KEEPER_GRAPHER_DRAIN_SERVICE_CONF: " + KEEPER_GRAPHER_DRAIN_SERVICE_CONF.to_String() +
+               ", KEEPER_GRAPHER_DRAIN_SERVICE_CONF: " + KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.to_String() +
                ", DATA_STORE_ADMIN_SERVICE_CONF: " + DATA_STORE_ADMIN_SERVICE_CONF.to_String() +
                ", VISOR_REGISTRY_SERVICE_CONF: " + VISOR_REGISTRY_SERVICE_CONF.to_String() +
                ", LOG_CONF: " + LOG_CONF.to_String() + ", DATA_STORE_CONF: " + DATA_STORE_CONF.to_String() +

ChronoGrapher/src/ChronoGrapher.cpp

@@ -129,15 +129,16 @@ int main(int argc, char** argv)
 
     // Instantiate GrapherRecordingService
     chronolog::RecordingGroupId recording_group_id = GRAPHER_CONF.RECORDING_GROUP;
-    std::string RECORDING_SERVICE_PROTOCOL = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.PROTO_CONF;
-    std::string RECORDING_SERVICE_IP = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.IP;
-    uint16_t RECORDING_SERVICE_PORT = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.BASE_PORT;
-    uint16_t recording_service_provider_id = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.SERVICE_PROVIDER_ID;
+    std::string RECORDING_SERVICE_PROTOCOL = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.PROTO_CONF;
+    std::string RECORDING_SERVICE_IP = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.IP;
+    uint16_t RECORDING_SERVICE_PORT = GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.BASE_PORT;
+    uint16_t recording_service_provider_id =
+            GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID;
 
-    chl::ServiceId recordingServiceId(GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.PROTO_CONF,
-                                      GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.IP,
-                                      GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.BASE_PORT,
-                                      GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.SERVICE_PROVIDER_ID);
+    chl::ServiceId recordingServiceId(GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.PROTO_CONF,
+                                      GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.IP,
+                                      GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.BASE_PORT,
+                                      GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID);
 
 
     std::string RECORDING_SERVICE_NA_STRING;
@@ -213,7 +214,9 @@ int main(int argc, char** argv)
 
     try
     {
-        margo_instance_id margo_id = margo_init(RECORDING_SERVICE_NA_STRING.c_str(), MARGO_SERVER_MODE, 1, 1);
+        int rpc_thread_count = static_cast<int>(GRAPHER_CONF.KEEPER_GRAPHER_DRAIN_SERVICE_CONF.INGESTION_THREAD_COUNT);
+        margo_instance_id margo_id =
+                margo_init(RECORDING_SERVICE_NA_STRING.c_str(), MARGO_SERVER_MODE, 1, rpc_thread_count);
         recordingEngine = new tl::engine(margo_id);
 
         std::stringstream s1;

ChronoGrapher/src/ChronoGrapherConfiguration.cpp

@@ -37,9 +37,20 @@ int chronolog::GrapherConfiguration::parseJsonConf(json_object* json_conf)
             {
                 if(strcmp(key, "rpc") == 0)
                 {
-                    if(KEEPER_GRAPHER_DRAIN_SERVICE_CONF.parseJsonConf(val) != chl::CL_SUCCESS)
+                    if(KEEPER_GRAPHER_DRAIN_SERVICE_CONF.RPC_CONF.parseJsonConf(val) != chl::CL_SUCCESS)
                         return chl::CL_ERR_INVALID_CONF;
                 }
+                else if(strcmp(key, "IngestionThreadCount") == 0)
+                {
+                    if(!json_object_is_type(val, json_type_int))
+                    {
+                        std::cerr << "[GrapherConfiguration] Invalid 'IngestionThreadCount': expected integer"
+                                  << std::endl;
+                        return chl::CL_ERR_INVALID_CONF;
+                    }
+                    int value = json_object_get_int(val);
+                    KEEPER_GRAPHER_DRAIN_SERVICE_CONF.INGESTION_THREAD_COUNT = (value >= 1 ? value : 1);
+                }
                 else
                 {
                     std::cerr << "[GrapherConfiguration] Unknown KeeperGrapherDrainService configuration: " << key

ChronoKeeper/include/ChronoKeeperConfiguration.h

@@ -14,10 +14,16 @@
 
 namespace chronolog
 {
+struct KeeperRecordingServiceConf
+{
+    RPCProviderConf RPC_CONF;
+    uint32_t INGESTION_THREAD_COUNT = 1;
+};
+
 struct KeeperConfiguration
 {
     uint32_t RECORDING_GROUP;
-    RPCProviderConf KEEPER_RECORDING_SERVICE_CONF;
+    KeeperRecordingServiceConf KEEPER_RECORDING_SERVICE_CONF;
     RPCProviderConf DATA_STORE_ADMIN_SERVICE_CONF;
     RPCProviderConf VISOR_REGISTRY_SERVICE_CONF;
     RPCProviderConf KEEPER_GRAPHER_DRAIN_SERVICE_CONF;
@@ -28,6 +34,13 @@ struct KeeperConfiguration
     KeeperConfiguration()
     {
         RECORDING_GROUP = 0;
+
+        KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.PROTO_CONF = "ofi+sockets";
+        KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.IP = "127.0.0.1";
+        KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.BASE_PORT = 6666;
+        KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID = 66;
+        KEEPER_RECORDING_SERVICE_CONF.INGESTION_THREAD_COUNT = 4;
+
         KEEPER_GRAPHER_DRAIN_SERVICE_CONF.PROTO_CONF = "ofi+sockets";
         KEEPER_GRAPHER_DRAIN_SERVICE_CONF.IP = "127.0.0.1";
         KEEPER_GRAPHER_DRAIN_SERVICE_CONF.BASE_PORT = 9999;

ChronoKeeper/include/IngestionQueue.h

@@ -5,6 +5,7 @@
 #include <deque>
 #include <unordered_map>
 #include <mutex>
+#include <shared_mutex>
 #include <string>
 #include <sstream>
 
@@ -32,8 +33,8 @@ class IngestionQueue
 
     void addStoryIngestionHandle(StoryId const& story_id, StoryIngestionHandle* ingestion_handle)
     {
-        std::lock_guard<std::mutex> lock(ingestionQueueMutex);
-        storyIngestionHandles.emplace(std::pair<StoryId, StoryIngestionHandle*>(story_id, ingestion_handle));
+        std::unique_lock<std::shared_mutex> lock(handleMapMutex);
+        storyIngestionHandles.emplace(story_id, ingestion_handle);
         LOG_DEBUG("[IngestionQueue] Added handle for StoryID={}: HandleAddress={}, StoryIngestionHandles={}, "
                   "HandleMapSize={}",
                   story_id,
@@ -44,7 +45,7 @@ class IngestionQueue
 
     void removeIngestionHandle(StoryId const& story_id)
     {
-        std::lock_guard<std::mutex> lock(ingestionQueueMutex);
+        std::unique_lock<std::shared_mutex> lock(handleMapMutex);
         if(storyIngestionHandles.erase(story_id))
         {
             LOG_DEBUG("[IngestionQueue] Removed handle for StoryID={}. Current handle MapSize={}",
@@ -57,36 +58,55 @@ class IngestionQueue
         }
     }
 
+    // Hot path: called concurrently by the keeper's RPC handler threads
+    // (rpc_thread_count = INGESTION_THREAD_COUNT). The map lookup is guarded
+    // by a shared lock so multiple ingest threads can resolve handles in
+    // parallel; only add/removeIngestionHandle take an exclusive lock.
+    // The handle itself is invoked outside the map lock so handler threads
+    // don't serialize on it -- StoryIngestionHandle::ingestEvent has its own
+    // per-story mutex.
     void ingestLogEvent(LogEvent const& event)
     {
-        std::stringstream ss;
-        ss << event;
-        LOG_DEBUG("[IngestionQueue] Received event for StoryID={}: Event Details={}, HandleMapSize={}",
-                  event.storyId,
-                  ss.str(),
-                  storyIngestionHandles.size());
-        auto ingestionHandle_iter = storyIngestionHandles.find(event.storyId);
-        if(ingestionHandle_iter == storyIngestionHandles.end())
+        StoryIngestionHandle* handle = nullptr;
         {
-            LOG_WARNING("[IngestionQueue] Orphan event for story {}. Storing for later processing.", event.storyId);
-            std::lock_guard<std::mutex> lock(ingestionQueueMutex);
-            orphanEventQueue.push_back(event);
+            std::shared_lock<std::shared_mutex> lock(handleMapMutex);
+#ifndef NDEBUG
+            std::stringstream ss;
+            ss << event;
+            LOG_TRACE("[IngestionQueue] Received event for StoryID={}: Event Details={}, HandleMapSize={}",
+                      event.storyId,
+                      ss.str(),
+                      storyIngestionHandles.size());
+#endif
+            auto ingestionHandle_iter = storyIngestionHandles.find(event.storyId);
+            if(ingestionHandle_iter != storyIngestionHandles.end())
+                handle = ingestionHandle_iter->second;
+        }
+
+        if(handle != nullptr)
+        {
+            handle->ingestEvent(event);
         }
         else
         {
-            //individual StoryIngestionHandle has its own mutex
-            (*ingestionHandle_iter).second->ingestEvent(event);
+            LOG_WARNING("[IngestionQueue] Orphan event for story {}. Storing for later processing.", event.storyId);
+            std::lock_guard<std::mutex> lock(orphanQueueMutex);
+            orphanEventQueue.push_back(event);
         }
     }
 
+    // Lock ordering when both mutexes are needed: orphanQueueMutex first,
+    // then handleMapMutex. ingestLogEvent never holds them simultaneously, so
+    // there is no inverse-order acquisition path.
     void drainOrphanEvents()
     {
+        std::lock_guard<std::mutex> orphanLock(orphanQueueMutex);
         if(orphanEventQueue.empty())
         {
             LOG_DEBUG("[IngestionQueue] Orphan event queue is empty. No actions taken.");
             return;
         }
-        std::lock_guard<std::mutex> lock(ingestionQueueMutex);
+        std::shared_lock<std::shared_mutex> mapLock(handleMapMutex);
         for(EventDeque::iterator iter = orphanEventQueue.begin(); iter != orphanEventQueue.end();)
         {
             auto ingestionHandle_iter = storyIngestionHandles.find((*iter).storyId);
@@ -105,17 +125,26 @@ class IngestionQueue
         LOG_DEBUG("[IngestionQueue] Drained {} orphan events into known handles.", orphanEventQueue.size());
     }
 
-    bool is_empty() const { return (orphanEventQueue.empty() && storyIngestionHandles.empty()); }
+    bool is_empty() const
+    {
+        std::lock_guard<std::mutex> orphanLock(orphanQueueMutex);
+        std::shared_lock<std::shared_mutex> mapLock(handleMapMutex);
+        return (orphanEventQueue.empty() && storyIngestionHandles.empty());
+    }
 
     void shutDown()
     {
-        LOG_INFO("[IngestionQueue] Initiating shutdown. HandleMapSize={}, Orphan EventQueueSize={}",
-                 storyIngestionHandles.size(),
-                 orphanEventQueue.size());
+        {
+            std::lock_guard<std::mutex> orphanLock(orphanQueueMutex);
+            std::shared_lock<std::shared_mutex> mapLock(handleMapMutex);
+            LOG_INFO("[IngestionQueue] Initiating shutdown. HandleMapSize={}, Orphan EventQueueSize={}",
+                     storyIngestionHandles.size(),
+                     orphanEventQueue.size());
+        }
         // last attempt to drain orphanEventQueue into known ingestionHandles
         drainOrphanEvents();
         // disengage all handles
-        std::lock_guard<std::mutex> lock(ingestionQueueMutex);
+        std::unique_lock<std::shared_mutex> lock(handleMapMutex);
         storyIngestionHandles.clear();
         LOG_INFO("[IngestionQueue] Shutdown completed. All handles disengaged.");
     }
@@ -125,11 +154,15 @@ class IngestionQueue
 
     IngestionQueue& operator=(IngestionQueue const&) = delete;
 
-    std::mutex ingestionQueueMutex;
+    // handleMapMutex is shared/exclusive: ingestLogEvent and drainOrphanEvents
+    // take shared locks (concurrent map lookups); add/remove take exclusive.
+    // mutable so const observers (is_empty) can take a shared lock.
+    mutable std::shared_mutex handleMapMutex;
     std::unordered_map<StoryId, StoryIngestionHandle*> storyIngestionHandles;
 
     // events for unknown stories or late events for closed stories will end up
     // in orphanEventQueue that we'll periodically try to drain into the DataStore
+    mutable std::mutex orphanQueueMutex;
     std::deque<LogEvent> orphanEventQueue;
 
     //Timer to triger periodic attempt to drain orphanEventQueue and collect/log statistics

ChronoKeeper/include/KeeperRecordingService.h

@@ -39,9 +39,11 @@ class KeeperRecordingService: public tl::provider<KeeperRecordingService>
     {
         //  ClientId teller_id,  StoryId story_id,
         //  ChronoTick const& chrono_tick, std::string const& record)
+#ifndef NDEBUG
         std::stringstream ss;
         ss << log_event;
-        LOG_DEBUG("[KeeperRecordingService] Recording event: {}", ss.str());
+        LOG_TRACE("[KeeperRecordingService] Recording event: {}", ss.str());
+#endif
         theIngestionQueue.ingestLogEvent(log_event);
         request.respond(chronolog::CL_SUCCESS);
     }

ChronoKeeper/src/ChronoKeeperConfiguration.cpp

@@ -35,9 +35,20 @@ int chronolog::KeeperConfiguration::parseJsonConf(json_object* json_conf)
             {
                 if(strcmp(key, "rpc") == 0)
                 {
-                    if(KEEPER_RECORDING_SERVICE_CONF.parseJsonConf(val) != chl::CL_SUCCESS)
+                    if(KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.parseJsonConf(val) != chl::CL_SUCCESS)
                         return chl::CL_ERR_INVALID_CONF;
                 }
+                else if(strcmp(key, "IngestionThreadCount") == 0)
+                {
+                    if(!json_object_is_type(val, json_type_int))
+                    {
+                        std::cerr << "[KeeperConfiguration] Invalid 'IngestionThreadCount': expected integer"
+                                  << std::endl;
+                        return chl::CL_ERR_INVALID_CONF;
+                    }
+                    int value = json_object_get_int(val);
+                    KEEPER_RECORDING_SERVICE_CONF.INGESTION_THREAD_COUNT = (value >= 1 ? value : 1);
+                }
                 else
                 {
                     std::cerr << "[KeeperConfiguration] Unknown KeeperRecordingService "

ChronoKeeper/src/ChronoKeeperInstance.cpp

@@ -127,10 +127,10 @@ int main(int argc, char** argv)
 
     /// KeeperRecordingService setup ___________________________________________________________________________________
     // Instantiate KeeperRecordingService
-    std::string KEEPER_RECORDING_SERVICE_PROTOCOL = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.PROTO_CONF;
-    std::string KEEPER_RECORDING_SERVICE_IP = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.IP;
-    uint16_t KEEPER_RECORDING_SERVICE_PORT = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.BASE_PORT;
-    uint16_t recording_service_provider_id = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.SERVICE_PROVIDER_ID;
+    std::string KEEPER_RECORDING_SERVICE_PROTOCOL = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.PROTO_CONF;
+    std::string KEEPER_RECORDING_SERVICE_IP = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.IP;
+    uint16_t KEEPER_RECORDING_SERVICE_PORT = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.BASE_PORT;
+    uint16_t recording_service_provider_id = KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID;
 
 
     // validate ip address, instantiate Recording Service and create KeeperIdCard
@@ -149,10 +149,10 @@ int main(int argc, char** argv)
     chronolog::RecordingGroupId keeper_group_id = KEEPER_CONF.RECORDING_GROUP;
     chronolog::KeeperIdCard keeperIdCard(
             keeper_group_id,
-            chronolog::ServiceId(KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.PROTO_CONF,
-                                 KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.IP,
-                                 KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.BASE_PORT,
-                                 KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.SERVICE_PROVIDER_ID));
+            chronolog::ServiceId(KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.PROTO_CONF,
+                                 KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.IP,
+                                 KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.BASE_PORT,
+                                 KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID));
 
     std::string KEEPER_RECORDING_SERVICE_NA_STRING;
     keeperIdCard.getRecordingServiceId().get_service_as_string(KEEPER_RECORDING_SERVICE_NA_STRING);
@@ -242,7 +242,9 @@ int main(int argc, char** argv)
 
     try
     {
-        margo_instance_id margo_id = margo_init(KEEPER_RECORDING_SERVICE_NA_STRING.c_str(), MARGO_SERVER_MODE, 1, 1);
+        int rpc_thread_count = static_cast<int>(KEEPER_CONF.KEEPER_RECORDING_SERVICE_CONF.INGESTION_THREAD_COUNT);
+        margo_instance_id margo_id =
+                margo_init(KEEPER_RECORDING_SERVICE_NA_STRING.c_str(), MARGO_SERVER_MODE, 1, rpc_thread_count);
         recordingEngine = new tl::engine(margo_id);
 
         std::stringstream s1;

ChronoPlayer/include/ChronoPlayerConfiguration.h

@@ -11,11 +11,17 @@
 namespace chronolog
 {
 
+struct PlaybackServiceConf
+{
+    RPCProviderConf RPC_CONF;
+    uint32_t INGESTION_THREAD_COUNT = 1;
+};
+
 struct PlayerConfiguration
 {
     uint32_t RECORDING_GROUP;
     RPCProviderConf DATA_STORE_ADMIN_SERVICE_CONF;
-    RPCProviderConf PLAYBACK_SERVICE_CONF;
+    PlaybackServiceConf PLAYBACK_SERVICE_CONF;
     RPCProviderConf VISOR_REGISTRY_SERVICE_CONF;
     LogConf LOG_CONF;
     DataStoreConf DATA_STORE_CONF{};
@@ -29,10 +35,10 @@ struct PlayerConfiguration
         DATA_STORE_ADMIN_SERVICE_CONF.BASE_PORT = 2222;
         DATA_STORE_ADMIN_SERVICE_CONF.SERVICE_PROVIDER_ID = 22;
 
-        PLAYBACK_SERVICE_CONF.PROTO_CONF = "ofi+sockets";
-        PLAYBACK_SERVICE_CONF.IP = "127.0.0.1";
-        PLAYBACK_SERVICE_CONF.BASE_PORT = 2225;
-        PLAYBACK_SERVICE_CONF.SERVICE_PROVIDER_ID = 25;
+        PLAYBACK_SERVICE_CONF.RPC_CONF.PROTO_CONF = "ofi+sockets";
+        PLAYBACK_SERVICE_CONF.RPC_CONF.IP = "127.0.0.1";
+        PLAYBACK_SERVICE_CONF.RPC_CONF.BASE_PORT = 2225;
+        PLAYBACK_SERVICE_CONF.RPC_CONF.SERVICE_PROVIDER_ID = 25;
 
         VISOR_REGISTRY_SERVICE_CONF.PROTO_CONF = "ofi+sockets";
         VISOR_REGISTRY_SERVICE_CONF.IP = "127.0.0.1";
@@ -52,7 +58,7 @@ struct PlayerConfiguration
     {
         return "[CHRONO_PLAYER_CONFIGURATION: RECORDING_GROUP: " + std::to_string(RECORDING_GROUP) +
                ", DATA_STORE_ADMIN_SERVICE_CONF: " + DATA_STORE_ADMIN_SERVICE_CONF.to_String() +
-               ", PLAYBACK_SERVICE_CONF: " + PLAYBACK_SERVICE_CONF.to_String() +
+               ", PLAYBACK_SERVICE_CONF: " + PLAYBACK_SERVICE_CONF.RPC_CONF.to_String() +
                ", VISOR_REGISTRY_SERVICE_CONF: " + VISOR_REGISTRY_SERVICE_CONF.to_String() +
                ", LOG_CONF: " + LOG_CONF.to_String() + ", DATA_STORE_CONF: " + DATA_STORE_CONF.to_String() +
                ", READER_CONF: " + READER_CONF.to_String() + "]";