Skip to content

Move mutex and message bus initialization into t2_init #387

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
yogeswaransky wants to merge 8 commits into
base: support/1.9
Choose a base branch
from
Open

Move mutex and message bus initialization into t2_init #387

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b19c78a
Move mutex and message bus initialization into t2_init
yogeswaransky Jun 24, 2026
abbe305
RDKEMW-19134: Move mutex and message bus initialization into t2_init
yogeswaransky Jun 26, 2026
8601cdc
RDKB-65318: Enhancements
yogeswaransky Jun 26, 2026
4ded52c
RDKEMW-19134: Move mutex and message bus initialization into t2_init
yogeswaransky Jun 26, 2026
4064a86
Merge branch 'feature/RDKEMW-19134' of https://github.com/rdkcentral/…
yogeswaransky Jun 26, 2026
76a633f
RDKEMW-19134: Synchronizing t2_init with t2_event calls
yogeswaransky Jul 5, 2026
b08e01d
RDKEMW-19134: Synchronizing t2_init with t2_event calls
yogeswaransky Jul 6, 2026
49d8308
RDKEMW-19134: Synchronizing t2_init with t2_event calls
yogeswaransky Jul 6, 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
2 changes: 1 addition & 1 deletion source/commonlib/telemetry_busmessage_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ const char destCompPath[64] = "/com/cisco/spvtg/ccsp/pam";


#define EVENT_ERROR(format, ...) \
fprintf(stderr, "T2ERROR:%s %s:%d: ", __func__ , __FILE__, __LINE__ ); \
fprintf(stderr, "T2INFO:%s %s:%d: ", __func__ , __FILE__, __LINE__ ); \
fprintf(stderr, (format), ##__VA_ARGS__ ); \
fprintf(stderr, "\n" );
Comment on lines 29 to 32
Comment on lines 29 to 32
Comment on lines 29 to 32

Expand Down
97 changes: 93 additions & 4 deletions source/commonlib/telemetry_busmessage_sender.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ static bool isT2Ready = false;
static bool isRbusEnabled = false ;
static pthread_mutex_t initMtx = PTHREAD_MUTEX_INITIALIZER;
static bool isMutexInitialized = false ;
static volatile bool isT2InitComplete = false ;

static hash_map_t *eventMarkerMap = NULL;

Expand Down Expand Up @@ -204,7 +205,7 @@ static void rBusInterface_Uninit( )

static T2ERROR initMessageBus( )
{
// EVENT_DEBUG("%s ++in\n", __FUNCTION__);
EVENT_ERROR("%s ++in\n", __FUNCTION__);
T2ERROR status = T2ERROR_SUCCESS;
Comment on lines 206 to 209
Comment on lines 206 to 209
char* component_id = (char*)CCSP_FIXED_COMP_ID;
#if defined(CCSP_SUPPORT_ENABLED)
Expand Down Expand Up @@ -248,7 +249,7 @@ static T2ERROR initMessageBus( )
}
}
#endif // CCSP_SUPPORT_ENABLED
// EVENT_DEBUG("%s --out\n", __FUNCTION__);
EVENT_ERROR("%s --out\n", __FUNCTION__);
return status;
Comment on lines 250 to 253
}
Comment on lines +252 to 254

Expand Down Expand Up @@ -440,6 +441,7 @@ int filtered_event_send(const char* data, const char *markerName)
rbusError_t ret = RBUS_ERROR_SUCCESS;
int status = 0 ;
EVENT_DEBUG("%s ++in\n", __FUNCTION__);
EVENT_ERROR("%s ++in\n", __FUNCTION__);
if(!bus_handle)
Comment on lines 443 to 445
{
Comment on lines 443 to 446
EVENT_ERROR("bus_handle is null .. exiting !!! \n");
Expand Down Expand Up @@ -530,6 +532,7 @@ int filtered_event_send(const char* data, const char *markerName)
}
#endif // CCSP_SUPPORT_ENABLED
EVENT_DEBUG("%s --out with status %d \n", __FUNCTION__, status);
EVENT_ERROR("%s --out with status %d \n", __FUNCTION__, status);
return status;
}
Comment on lines 534 to 537
Comment on lines 534 to 537

Expand Down Expand Up @@ -560,6 +563,7 @@ static T2ERROR doPopulateEventMarkerList( )

snprintf(deNameSpace[0], 124, "%s%s%s", T2_ROOT_PARAMETER, componentName, T2_EVENT_LIST_PARAM_SUFFIX);
EVENT_DEBUG("rbus mode : Query marker list with data element = %s \n", deNameSpace[0]);
EVENT_ERROR("rbus mode : Query marker list with data element = %s \n", deNameSpace[0]);

Comment on lines 564 to 567
pthread_mutex_lock(&markerListMutex);
Comment on lines 564 to 568
EVENT_DEBUG("Lock markerListMutex & Clean up eventMarkerMap \n");
Expand Down Expand Up @@ -596,6 +600,7 @@ static T2ERROR doPopulateEventMarkerList( )
eventMarkerMap = hash_map_create();
rbusProperty_t rbusPropertyList = rbusObject_GetProperties(objectValue);
EVENT_DEBUG("\t rbus mode : Update event map for component %s with below events : \n", componentName);
EVENT_ERROR("\t rbus mode : Update event map for component %s with below events : \n", componentName);
while(NULL != rbusPropertyList)
Comment on lines 601 to 604
{
Comment on lines 600 to 605
const char* eventname = rbusProperty_GetName(rbusPropertyList);
Expand Down Expand Up @@ -668,6 +673,7 @@ static bool isCachingRequired( )
rbusError_t retVal = RBUS_ERROR_SUCCESS;

retVal = rbus_getUint(bus_handle, T2_OPERATIONAL_STATUS, &t2ReadyStatus);
EVENT_ERROR("Retrieved T2 Operational status: %d\n", t2ReadyStatus);

if(retVal != RBUS_ERROR_SUCCESS)
{
Expand Down Expand Up @@ -741,7 +747,7 @@ static int report_or_cache_data(char* telemetry_data, const char* markerName)

if(isT2Ready)
{
// EVENT_DEBUG("T2: Sending event : %s\n", telemetry_data);
EVENT_ERROR("T2: Sending event : %s\n", telemetry_data);
ret = filtered_event_send(telemetry_data, markerName);
Comment on lines 748 to 751
Comment on lines +750 to 751
Comment on lines 748 to 751
if(0 != ret)
{
Expand All @@ -751,12 +757,43 @@ static int report_or_cache_data(char* telemetry_data, const char* markerName)
return ret;
}

static void *t2_init_worker(void *arg)
{
(void)arg;
pthread_detach(pthread_self());
EVENT_ERROR("%s ++in\n", __FUNCTION__);

initMutex();

if(initMessageBus() != T2ERROR_SUCCESS)
{
EVENT_ERROR("%s:%d, T2:initMessageBus failed in t2_init, will retry during event send\n", __func__, __LINE__);
}
else
{
isRFCT2Enable = true;
}

isT2InitComplete = true;
EVENT_ERROR("%s --out\n", __FUNCTION__);
return NULL;
}

/**
* Initialize the component name with unique name
*/
void t2_init(char *component)
{
pthread_t initThread;
EVENT_ERROR("%s ++in\n", __FUNCTION__);
componentName = strdup(component);

if(pthread_create(&initThread, NULL, t2_init_worker, NULL) != 0)
{
Comment on lines 789 to +792
EVENT_ERROR("%s:%d, T2:Failed to create init worker thread, falling back to sync init\n", __func__, __LINE__);
t2_init_worker(NULL);
}
Comment on lines +791 to +795
EVENT_ERROR("%s --out\n", __FUNCTION__);
}

void t2_uninit(void)
Expand All @@ -782,12 +819,28 @@ T2ERROR t2_event_s(const char* marker, const char* value)
T2ERROR retStatus = T2ERROR_FAILURE ;
EVENT_DEBUG("%s ++in\n", __FUNCTION__);
char* strvalue = NULL;
if(!isT2InitComplete)
{
EVENT_ERROR("%s:%d, T2:t2_init is not complete, caching event %s\n", __func__, __LINE__, marker ? marker : "NULL");
if(marker && value && strlen(value) > 0 && strcmp(value, "0") != 0)
{
int eventDataLen = strlen(marker) + strlen(value) + strlen(MESSAGE_DELIMITER) + 1;
char* buffer = (char*) malloc(eventDataLen * sizeof(char));
if(buffer)
{
pthread_t tid;
snprintf(buffer, eventDataLen, "%s%s%s", marker, MESSAGE_DELIMITER, value);
initMutex();
pthread_create(&tid, NULL, cacheEventToFile, (void *)buffer);
}
}
return T2ERROR_SUCCESS;
}
if(componentName == NULL)
{
EVENT_DEBUG("%s:%d, T2:component with pid = %d is trying to send event %s with value %s without component name \n", __func__, __LINE__, (int) getpid(), marker, value);
return T2ERROR_COMPONENT_NULL;
}
initMutex();
pthread_mutex_lock(&sMutex);
if ( NULL == marker || NULL == value)
{
Expand Down Expand Up @@ -830,6 +883,24 @@ T2ERROR t2_event_f(const char* marker, double value)
T2ERROR retStatus = T2ERROR_FAILURE ;
EVENT_DEBUG("%s ++in\n", __FUNCTION__);

if(!isT2InitComplete)
{
EVENT_ERROR("%s:%d, T2:t2_init is not complete, caching event %s\n", __func__, __LINE__, marker ? marker : "NULL");
if(marker)
{
char *buffer = (char*) malloc(MAX_DATA_LEN * sizeof(char));
if(buffer)
{
pthread_t tid;
char valStr[64] = { '\0' };
snprintf(valStr, sizeof(valStr), "%f", value);
snprintf(buffer, MAX_DATA_LEN, "%s%s%s", marker, MESSAGE_DELIMITER, valStr);
initMutex();
pthread_create(&tid, NULL, cacheEventToFile, (void *)buffer);
}
}
return T2ERROR_SUCCESS;
}
Comment on lines +886 to +903
if(componentName == NULL)
{
EVENT_DEBUG("%s:%d, T2:component with pid = %d is trying to send event %s with value %lf without component name \n", __func__, __LINE__, (int) getpid(), marker, value);
Expand Down Expand Up @@ -874,6 +945,24 @@ T2ERROR t2_event_d(const char* marker, int value)
T2ERROR retStatus = T2ERROR_FAILURE ;
EVENT_DEBUG("%s ++in\n", __FUNCTION__);

if(!isT2InitComplete)
{
EVENT_ERROR("%s:%d, T2:t2_init is not complete, caching event %s\n", __func__, __LINE__, marker ? marker : "NULL");
if(marker && value != 0)
{
char *buffer = (char*) malloc(MAX_DATA_LEN * sizeof(char));
if(buffer)
{
pthread_t tid;
char valStr[64] = { '\0' };
snprintf(valStr, sizeof(valStr), "%d", value);
snprintf(buffer, MAX_DATA_LEN, "%s%s%s", marker, MESSAGE_DELIMITER, valStr);
initMutex();
pthread_create(&tid, NULL, cacheEventToFile, (void *)buffer);
}
}
return T2ERROR_SUCCESS;
}
Comment on lines +948 to +965
if(componentName == NULL)
{
EVENT_DEBUG("%s:%d, T2:component with pid = %d is trying to send event %s with value %d without component name \n", __func__, __LINE__, (int) getpid(), marker, value);
Expand Down