RDK-60291 [telemetry] RDK Coverity Defect Resolution for Device Management

Author: mtirum011

source/bulkdata/datamodel.c

@@ -50,14 +50,24 @@ static void *process_rp_thread(void *data)
 {
     (void) data;//To fix compiler warning
     cJSON *reportProfiles = NULL;
+     bool shouldContinue = true;
 
     T2Debug("%s ++in\n", __FUNCTION__);
 
-    while(!stopProcessing)
+    while(shouldContinue)
     {
-        pthread_mutex_lock(&rpMutex);
-        T2Info("%s: Waiting for event from tr-181 \n", __FUNCTION__);
-        pthread_cond_wait(&rpCond, &rpMutex);
+        pthread_mutex_lock(&rpMutex); 
+	shouldContinue = !stopProcessing;
+        if(!shouldContinue)
+        {
+            pthread_mutex_unlock(&rpMutex);
+            break;
+        }
+	T2Info("%s: Waiting for event from tr-181 \n", __FUNCTION__);
+        while(t2_queue_count(rpQueue) == 0 && shouldContinue)
+        {
+            pthread_cond_wait(&rpCond, &rpMutex);
+        }
 
         T2Debug("%s: Received wake up signal \n", __FUNCTION__);
         if(t2_queue_count(rpQueue) > 0)
@@ -67,10 +77,10 @@ static void *process_rp_thread(void *data)
             {
                 ReportProfiles_ProcessReportProfilesBlob(reportProfiles, T2_RP);
                 cJSON_Delete(reportProfiles);
-                // Unused value reportProfiles
-            }
+                // Unused value reportProfiles          
+	    }
         }
-        pthread_mutex_unlock(&rpMutex);
+	pthread_mutex_unlock(&rpMutex);
     }
     T2Debug("%s --out\n", __FUNCTION__);
     return NULL;
@@ -80,14 +90,26 @@ static void *process_tmprp_thread(void *data)
 {
     (void) data;//To fix compiler warning
     cJSON *tmpReportProfiles = NULL;
+    bool shouldContinue = true;
 
     T2Debug("%s ++in\n", __FUNCTION__);
 
-    while(!stopProcessing)
+    while(shouldContinue)
     {
         pthread_mutex_lock(&tmpRpMutex);
+	pthread_mutex_lock(&rpMutex);
+        shouldContinue = !stopProcessing;
+        pthread_mutex_unlock(&rpMutex);
+        if(!shouldContinue)
+        {
+            pthread_mutex_unlock(&tmpRpMutex);
+            break;
+        }
         T2Info("%s: Waiting for event from tr-181 \n", __FUNCTION__);
-        pthread_cond_wait(&tmpRpCond, &tmpRpMutex);
+        while(t2_queue_count(tmpRpQueue) == 0 && shouldContinue)
+        {
+            pthread_cond_wait(&tmpRpCond, &tmpRpMutex);
+        }
 
         T2Debug("%s: Received wake up signal \n", __FUNCTION__);
         if(t2_queue_count(tmpRpQueue) > 0)
@@ -110,10 +132,23 @@ static void *process_msg_thread(void *data)
 {
     (void) data;//To fix compiler warning
     struct __msgpack__ *msgpack;
-    while(!stopProcessing)
+    bool shouldContinue = true;
+
+    while(shouldContinue)
     {
         pthread_mutex_lock(&rpMsgMutex);
-        pthread_cond_wait(&msg_Cond, &rpMsgMutex);
+	pthread_mutex_lock(&rpMutex);
+        shouldContinue = !stopProcessing;
+        pthread_mutex_unlock(&rpMutex);
+        if(!shouldContinue)
+        {
+            pthread_mutex_unlock(&rpMsgMutex);
+            break;
+        }
+        while(t2_queue_count(rpMsgPkgQueue) == 0 && shouldContinue)
+        {
+            pthread_cond_wait(&msg_Cond, &rpMsgMutex);
+        }
         if(t2_queue_count(rpMsgPkgQueue) > 0)
         {
             msgpack = (struct __msgpack__ *)t2_queue_pop(rpMsgPkgQueue);
@@ -275,14 +310,17 @@ T2ERROR datamodel_MsgpackProcessProfile(char *str, int strSize)
     msgpack->msgpack_blob = str;
     msgpack->msgpack_blob_size = strSize;
     pthread_mutex_lock(&rpMsgMutex);
+    pthread_mutex_lock(&rpMutex);
     if (!stopProcessing)
     {
-        t2_queue_push(rpMsgPkgQueue, (void *)msgpack);
+        pthread_mutex_unlock(&rpMutex);
+	t2_queue_push(rpMsgPkgQueue, (void *)msgpack);
         pthread_cond_signal(&msg_Cond);
     }
     else
     {
-        free(msgpack->msgpack_blob);
+        pthread_mutex_unlock(&rpMutex);
+	free(msgpack->msgpack_blob);
         free(msgpack);
         T2Error("Datamodel not initialized, dropping request \n");
     }

source/bulkdata/profile.c

@@ -331,10 +331,13 @@ static void* CollectAndReport(void* data)
     do
     {
         T2Info("%s while Loop -- START \n", __FUNCTION__);
+        // Release reuseThreadMutex before acquiring reportInProgressMutex to maintain lock order
+        pthread_mutex_unlock(&profile->reuseThreadMutex);
         pthread_mutex_lock(&profile->reportInProgressMutex);
         profile->reportInProgress = true;
         pthread_cond_signal(&profile->reportInProgressCond);
         pthread_mutex_unlock(&profile->reportInProgressMutex);
+        pthread_mutex_lock(&profile->reuseThreadMutex);
 
         int count = profile->grepSeekProfile->execCounter;
 
@@ -549,8 +552,18 @@ static void* CollectAndReport(void* data)
                     pthread_cond_init(&profile->reportcond, NULL);
                     clock_gettime(CLOCK_REALTIME, &profile->currentTime);
                     profile->maxlatencyTime.tv_sec = profile->currentTime.tv_sec;
-                    srand(time(0)); // Initialise the random number generator
-                    maxuploadinmilliSec = rand() % (profile->maxUploadLatency - 1);
+                    unsigned int random_value = 0;
+		    FILE *urandom = fopen("/dev/urandom", "r");
+		    if(urandom != NULL && fread(&random_value, sizeof(random_value), 1, urandom) == 1)
+		    {
+			maxuploadinmilliSec = random_value % (profile->maxUploadLatency - 1);
+			fclose(urandom);
+		    }
+		    else
+		    {
+			if(urandom != NULL) fclose(urandom);
+		        maxuploadinmilliSec = (unsigned int)(time(0) % (profile->maxUploadLatency - 1));
+		    }
                     maxuploadinSec =  (maxuploadinmilliSec + 1) / 1000;
                 }
                 if( strcmp(profile->protocol, "HTTP") == 0 || strcmp(profile->protocol, "RBUS_METHOD") == 0 )
@@ -564,12 +577,18 @@ static void* CollectAndReport(void* data)
                             pthread_mutex_lock(&profile->reportMutex);
                             T2Info("waiting for %ld sec of macUploadLatency\n", (long) maxuploadinSec);
                             profile->maxlatencyTime.tv_sec += maxuploadinSec;
-                            n = pthread_cond_timedwait(&profile->reportcond, &profile->reportMutex, &profile->maxlatencyTime);
+                            do {
+                                n = pthread_cond_timedwait(&profile->reportcond, &profile->reportMutex, &profile->maxlatencyTime);
+                            } while (n == EINTR);
                             if(n == ETIMEDOUT)
                             {
                                 T2Info("TIMEOUT for maxUploadLatency of profile %s\n", profile->name);
                                 ret = sendReportOverHTTP(httpUrl, jsonReport, NULL);
                             }
+			    else if(n == 0)
+                            {
+                                T2Info("Profile : %s signaled before timeout, skipping report upload\n", profile->name);
+                            }
                             else
                             {
                                 T2Error("Profile : %s pthread_cond_timedwait ERROR!!!\n", profile->name);
@@ -615,12 +634,18 @@ static void* CollectAndReport(void* data)
                             pthread_mutex_lock(&profile->reportMutex);
                             T2Info("waiting for %ld sec of macUploadLatency\n", (long) maxuploadinSec);
                             profile->maxlatencyTime.tv_sec += maxuploadinSec;
-                            n = pthread_cond_timedwait(&profile->reportcond, &profile->reportMutex, &profile->maxlatencyTime);
+                            do {
+                                n = pthread_cond_timedwait(&profile->reportcond, &profile->reportMutex, &profile->maxlatencyTime);
+                            } while (n == EINTR);
                             if(n == ETIMEDOUT)
                             {
                                 T2Info("TIMEOUT for maxUploadLatency of profile %s\n", profile->name);
                                 ret = sendReportsOverRBUSMethod(profile->t2RBUSDest->rbusMethodName, profile->t2RBUSDest->rbusMethodParamList, jsonReport);
                             }
+			    else if(n == 0)
+                            {
+                                T2Info("Profile : %s signaled before timeout, skipping report upload\n", profile->name);
+                            }
                             else
                             {
                                 T2Error("Profile : %s pthread_cond_timedwait ERROR!!!\n", profile->name);
@@ -782,9 +807,11 @@ reportThreadEnd :
     }
     while(profile->enable);
     T2Info("%s --out Exiting collect and report Thread\n", __FUNCTION__);
+    pthread_mutex_unlock(&profile->reuseThreadMutex);
     pthread_mutex_lock(&profile->reportInProgressMutex);
     profile->reportInProgress = false;
     pthread_mutex_unlock(&profile->reportInProgressMutex);
+    pthread_mutex_lock(&profile->reuseThreadMutex);
     profile->threadExists = false;
     pthread_mutex_unlock(&profile->reuseThreadMutex);
     pthread_mutex_destroy(&profile->reuseThreadMutex);

source/bulkdata/reportprofiles.c

@@ -946,7 +946,17 @@ void ReportProfiles_ProcessReportProfilesBlob(cJSON *profiles_root, bool rprofil
 
     // Delete profiles not present in the new profile list
     char *profileNameKey = NULL;
-    int count = hash_map_count(profileHashMap) - 1;
+    uint32_t hashmap_count = hash_map_count(profileHashMap);
+    int count;
+    if (hashmap_count == 0 || hashmap_count == UINT32_MAX)
+    {
+        count = -1;
+    }
+    else
+    {
+        count = (int)(hashmap_count - 1);
+    }
+
     const char *DirPath = NULL;
 
     if (rprofiletypes == T2_RP)
@@ -1309,6 +1319,7 @@ int __ReportProfiles_ProcessReportProfilesMsgPackBlob(void *msgpack, bool checkP
         return T2ERROR_PROFILE_NOT_FOUND;
     }
     hash_map_t *profileHashMap;
+    uint32_t hashmap_count;
     int count;
     char *profileNameKey = NULL;
     int profileIndex;
@@ -1325,7 +1336,15 @@ int __ReportProfiles_ProcessReportProfilesMsgPackBlob(void *msgpack, bool checkP
     }
 
     /* Delete profiles not present in the new profile list */
-    count = hash_map_count(profileHashMap) - 1;
+    hashmap_count = hash_map_count(profileHashMap);
+    if (hashmap_count == 0 || hashmap_count == UINT32_MAX)
+    {
+        count = -1;
+    }
+    else
+    {
+        count = (int)(hashmap_count - 1);
+    }
     while(count >= 0)
     {
         profile_found_flag = false;

source/bulkdata/t2eventreceiver.c

@@ -204,7 +204,14 @@ void T2ER_Push(char* eventName, char* eventValue)
                     event->value = strdup(eventValue);
                     T2Debug("Adding eventName : %s eventValue : %s to t2event queue\n", event->name, event->value);
                     t2_queue_push(eQueue, (void *) event);
-                    if(!stopDispatchThread)
+                    
+		    bool shouldSignal = false;
+		    if(pthread_mutex_lock(&sTDMutex) == 0)
+                    {
+                        shouldSignal = !stopDispatchThread;
+                        pthread_mutex_unlock(&sTDMutex);
+                    }
+		    if(shouldSignal)
                     {
                         ret = pthread_cond_signal(&erCond);
                         if(ret != 0) // pthread_cond _signal failed so after unlocking the mutex it will return
@@ -235,15 +242,27 @@ void* T2ER_EventDispatchThread(void *arg)
     (void) arg; // To avoid compiler warning
     T2Debug("%s ++in\n", __FUNCTION__);
     Vector *profileList = NULL;
-    while(!stopDispatchThread)
+    bool shouldContinue = true;
+    while(shouldContinue)
     {
         if(pthread_mutex_lock(&erMutex) != 0) // mutex lock failed, without locking eQueue shouldn't be accessed
         {
             T2Error("%s pthread_mutex_lock for erMutex failed\n", __FUNCTION__);
             return NULL;
         }
         T2Debug("Checking for events in event queue , event count = %d\n", t2_queue_count(eQueue));
-        if(t2_queue_count(eQueue) > 0)
+        while(t2_queue_count(eQueue) == 0 && shouldContinue)
+        {
+            T2Debug("Event Queue size is 0, Waiting events from T2ER_Push\n");
+            int ret = pthread_cond_wait(&erCond, &erMutex);
+            if(ret != 0) // pthread cond wait failed return after unlock
+            {
+                T2Error("%s pthread_cond_wait failed with error code: %d\n", __FUNCTION__, ret);
+            }
+            T2Debug("Received signal from T2ER_Push\n");
+        }
+
+	if(t2_queue_count(eQueue) > 0)
         {
             T2Event *event = (T2Event *)t2_queue_pop(eQueue);
             if(event == NULL)
@@ -280,18 +299,21 @@ void* T2ER_EventDispatchThread(void *arg)
         }
         else
         {
-            T2Debug("Event Queue size is 0, Waiting events from T2ER_Push\n");
-            int ret = pthread_cond_wait(&erCond, &erMutex);
-            if(ret != 0) // pthread cond wait failed return after unlock
-            {
-                T2Error("%s pthread_cond_wait failed with error code: %d\n", __FUNCTION__, ret);
-            }
             if(pthread_mutex_unlock(&erMutex) != 0)
             {
                 T2Error("%s pthread_mutex_unlock for erMutex failed\n", __FUNCTION__);
                 return NULL;
             }
-            T2Debug("Received signal from T2ER_Push\n");
+        }
+	// Check if thread should continue
+        if(pthread_mutex_lock(&sTDMutex) == 0)
+        {
+            shouldContinue = !stopDispatchThread;
+            pthread_mutex_unlock(&sTDMutex);
+        }
+        else
+        {
+            shouldContinue = false; // Exit on lock failure
         }
     }
     T2Debug("%s --out\n", __FUNCTION__);
@@ -345,7 +367,10 @@ T2ERROR T2ER_Init()
         registerForTelemetryEvents(T2ER_PushDataWithDelim);
     }
 
-    system("touch /tmp/.t2ReadyToReceiveEvents");
+    if (system("touch /tmp/.t2ReadyToReceiveEvents") != 0)
+    {
+	T2Error("Failed to create /tmp/.t2ReadyToReceiveEvents flag file \n");
+    }
     setT2EventReceiveState(T2_STATE_COMPONENT_READY);
     T2Info("T2 is now Ready to Recieve Events\n");
 
@@ -500,14 +525,16 @@ void T2ER_Uninit()
     }
     EREnabled = false;
 
+    pthread_t threadToJoin;
+    if(pthread_mutex_lock(&sTDMutex) != 0) // mutex lock failed so return from T2ER_Uninit
+    {
+	T2Error("%s pthread_mutex_lock for sTDMutex failed\n", __FUNCTION__);
+	return;
+    }
     if(!stopDispatchThread)
     {
-        if(pthread_mutex_lock(&sTDMutex) != 0) // mutex lock failed so return from T2ER_Uninit
-        {
-            T2Error("%s pthread_mutex_lock for sTDMutex failed\n", __FUNCTION__);
-            return;
-        }
         stopDispatchThread = true;
+	threadToJoin = erThread; // Save thread handle while holding lock
         if(pthread_mutex_unlock(&sTDMutex) != 0) //mutex unlock failed so return from T2ER_Uninit
         {
             T2Error("%s pthread_mutex_unlock for sTDMutex failed\n", __FUNCTION__);
@@ -530,7 +557,7 @@ void T2ER_Uninit()
             return;
         }
 
-        if(pthread_join(erThread, NULL) != 0)
+        if(pthread_join(threadToJoin, NULL) != 0)
         {
             T2Error("%s erThread join failed\n", __FUNCTION__);
         }
@@ -551,6 +578,14 @@ void T2ER_Uninit()
             return;
         }
     }
+    else
+    {
+	if(pthread_mutex_unlock(&sTDMutex) != 0)
+        {
+            T2Error("%s pthread_mutex_unlock for sTDMutex failed\n", __FUNCTION__);
+            return;
+        }
+    }
     T2Debug("T2ER Event Dispatch Thread successfully terminated\n");
     t2_queue_destroy(eQueue, freeT2Event);
     eQueue = NULL;