Skip to content

621 configurable ingestion thread count #636

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fkengun wants to merge 13 commits into
base: develop
Choose a base branch
from
Open

621 configurable ingestion thread count #636

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
1dad12c
Add KEEPER_CONF.INGESTION_THREAD_COUNT to control the ingestion threa…
Apr 27, 2026
93c6003
Merge remote-tracking branch 'origin/develop' into 621-configurable-i…
May 1, 2026
88d96c7
Only log event string in trace log in Debug build
May 5, 2026
6fd1972
Only log event string in trace log in Debug build
May 5, 2026
f37d71a
Move IngestionThreadCount under KeeperRecordingService, and default to 4
May 5, 2026
d97f569
Add ingestion thread count config for Grapher
May 5, 2026
e926656
Add ingestion thread count config for Player
May 5, 2026
c9e3ba1
Fix clang-format
May 6, 2026
0ec24ab
Remove IngestionThreadCount config from deprecated conf manager
May 6, 2026
4d31af5
Remove debug loggin in record_event()
May 6, 2026
4daa7ff
Consistently initialize ingestion thread count in Keeper
May 6, 2026
c55d4b1
Fix IngestionQueue data race on concurrent RPC handlers
May 7, 2026
9b1d9cd
Hold IngestionQueue shared lock across ingestEvent
May 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions ChronoGrapher/include/ChronoGrapherConfiguration.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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";
Expand All @@ -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() +
Expand Down
21 changes: 12 additions & 9 deletions ChronoGrapher/src/ChronoGrapher.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Comment thread
fkengun marked this conversation as resolved.
recordingEngine = new tl::engine(margo_id);

std::stringstream s1;
Expand Down
13 changes: 12 additions & 1 deletion ChronoGrapher/src/ChronoGrapherConfiguration.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
15 changes: 14 additions & 1 deletion ChronoKeeper/include/ChronoKeeperConfiguration.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment thread
fkengun marked this conversation as resolved.
RPCProviderConf VISOR_REGISTRY_SERVICE_CONF;
RPCProviderConf KEEPER_GRAPHER_DRAIN_SERVICE_CONF;
Expand All @@ -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;
Expand Down
70 changes: 50 additions & 20 deletions ChronoKeeper/include/IngestionQueue.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
#include <deque>
#include <unordered_map>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <sstream>

Expand Down Expand Up @@ -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,
Expand All @@ -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={}",
Expand All @@ -57,36 +58,52 @@ 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 shared lock is held across handle->ingestEvent so the retirement
// path (KeeperDataStore::retireDecayedPipelines: removeIngestionHandle
// followed by `delete pipeline`) cannot tear down the handle mid-call.
// Concurrent ingest threads still proceed in parallel since shared locks
// don't block each other; only the rare retire path waits.
void ingestLogEvent(LogEvent const& event)
{
std::shared_lock<std::shared_mutex> lock(handleMapMutex);
#ifndef NDEBUG
std::stringstream ss;
ss << event;
LOG_DEBUG("[IngestionQueue] Received event for StoryID={}: Event Details={}, HandleMapSize={}",
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())
if(ingestionHandle_iter != storyIngestionHandles.end())
{
LOG_WARNING("[IngestionQueue] Orphan event for story {}. Storing for later processing.", event.storyId);
std::lock_guard<std::mutex> lock(ingestionQueueMutex);
orphanEventQueue.push_back(event);
}
else
{
//individual StoryIngestionHandle has its own mutex
(*ingestionHandle_iter).second->ingestEvent(event);
ingestionHandle_iter->second->ingestEvent(event);
return;
}
lock.unlock();

LOG_WARNING("[IngestionQueue] Orphan event for story {}. Storing for later processing.", event.storyId);
std::lock_guard<std::mutex> orphan_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);
Expand All @@ -105,17 +122,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.");
}
Expand All @@ -125,11 +151,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
Expand Down
4 changes: 3 additions & 1 deletion ChronoKeeper/include/KeeperRecordingService.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
13 changes: 12 additions & 1 deletion ChronoKeeper/src/ChronoKeeperConfiguration.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 "
Expand Down
20 changes: 11 additions & 9 deletions ChronoKeeper/src/ChronoKeeperInstance.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Comment thread
fkengun marked this conversation as resolved.
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;
Expand Down
18 changes: 12 additions & 6 deletions ChronoPlayer/include/ChronoPlayerConfiguration.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{};
Expand All @@ -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";
Expand All @@ -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() + "]";
Expand Down
Loading
Loading