Improve code and UT

common/asyncdbupdater.cpp

@@ -63,15 +63,13 @@ void AsyncDBUpdater::dbUpdateThread()
 
     while (m_runThread)
     {
-        m_dbUpdateDataNotifyCv.wait(cvLock);
-
         size_t count;
+        count = queueSize();
+        if (count == 0)
         {
-            count = queueSize();
-            if (!count)
-            {
-                continue;
-            }
+            // when queue is empty, wait notification, when data come, continue to check queue size again
+            m_dbUpdateDataNotifyCv.wait(cvLock);
+            continue;
         }
 
         for (size_t ie = 0; ie < count; ie++)
@@ -101,6 +99,8 @@ void AsyncDBUpdater::dbUpdateThread()
             }
         }
     }
+
+    SWSS_LOG_DEBUG("AsyncDBUpdater dbUpdateThread end: %s", m_tableName.c_str());
 }
 
 size_t AsyncDBUpdater::queueSize()

tests/zmq_state_ut.cpp

@@ -109,12 +109,21 @@ static void producerWorker(string tableName, string endpoint, bool dbPersistence
         p.del(keys);
     }
 
-    // wait all data been received by consumer and all persist data write to redis 
-    while (!allDataReceived && (p.dbUpdaterQueueSize() > 0))
+    // wait all data been received by consumer
+    while (!allDataReceived)
     {
         sleep(1);
     }
 
+    if (dbPersistence)
+    {
+        // wait all persist data write to redis
+        while (p.dbUpdaterQueueSize() > 0)
+        {
+            sleep(1);
+        }
+    }
+
     cout << "Producer thread ended: " << tableName << endl;
 }
 
@@ -180,13 +189,17 @@ static void consumerWorker(string tableName, string endpoint, bool dbPersistence
         }
     }
 
-    // wait all persist data write to redis 
-    while (c.dbUpdaterQueueSize() > 0)
+    allDataReceived = true;
+
+    if (dbPersistence)
     {
-        sleep(1);
+        // wait all persist data write to redis
+        while (c.dbUpdaterQueueSize() > 0)
+        {
+            sleep(1);
+        }
     }
 
-    allDataReceived = true;
     cout << "Consumer thread ended: " << tableName << endl;
 }
 
@@ -203,6 +216,7 @@ static void testMethod(bool producerPersistence)
     delCount = 0;
     batchSetCount = 0;
     batchDelCount = 0;
+    allDataReceived = false;
 
     // start consumer first, SHM can only have 1 consumer per table.
     thread *consumerThread = new thread(consumerWorker, testTableName, pullEndpoint, !producerPersistence);