Fixed race condition in ConcurrentQueue

Also, minor test cleanup

Test: fixed failing HalClientVectorTest test, verified others

Change-Id: I693cac8e31019347b1d3c4f34c0554fbc9b062d0
Fix: b/33305646

5 files changed, 17 insertions, 26 deletions

diff --git a/vehicle/2.0/default/tests/SubscriptionManager_test.cpp b/vehicle/2.0/default/tests/SubscriptionManager_test.cpp
index 19b11e6b7..171a692b5 100644
--- a/vehicle/2.0/default/tests/SubscriptionManager_test.cpp
+++ b/vehicle/2.0/default/tests/SubscriptionManager_test.cpp
@@ -19,9 +19,6 @@
 
 #include <gtest/gtest.h>
 
-#include <vehicle_hal_manager/VehiclePropConfigIndex.h>
-#include <VehicleHal.h>
-#include <vehicle_hal_manager/VehicleHalManager.h>
 #include "vehicle_hal_manager/SubscriptionManager.h"
 
 #include "VehicleHalTestUtils.h"
diff --git a/vehicle/2.0/default/tests/VehicleHalManager_test.cpp b/vehicle/2.0/default/tests/VehicleHalManager_test.cpp
index 6ef12052d..d06f6f4fa 100644
--- a/vehicle/2.0/default/tests/VehicleHalManager_test.cpp
+++ b/vehicle/2.0/default/tests/VehicleHalManager_test.cpp
@@ -17,13 +17,11 @@
 #include <unordered_map>
 #include <iostream>
 
+#include <utils/SystemClock.h>
+
 #include <gtest/gtest.h>
 
-#include <vehicle_hal_manager/VehiclePropConfigIndex.h>
-#include <VehicleHal.h>
-#include <vehicle_hal_manager/VehicleHalManager.h>
-#include <utils/SystemClock.h>
-#include "vehicle_hal_manager/SubscriptionManager.h"
+#include "vehicle_hal_manager/VehicleHalManager.h"
 
 #include "VehicleHalTestUtils.h"
 
@@ -429,11 +427,11 @@ TEST(HalClientVectorTest, basic) {
     clients.addOrUpdate(c2);
     ASSERT_EQ(2u, clients.size());
     ASSERT_FALSE(clients.isEmpty());
-    ASSERT_GE(0, clients.indexOf(c1));
-    ASSERT_GE(0, clients.remove(c1));
-    ASSERT_GE(0, clients.indexOf(c1));
-    ASSERT_GE(0, clients.remove(c1));
-    ASSERT_GE(0, clients.remove(c2));
+    ASSERT_LE(0, clients.indexOf(c1));
+    ASSERT_LE(0, clients.remove(c1));
+    ASSERT_GT(0, clients.indexOf(c1));  // c1 was already removed
+    ASSERT_GT(0, clients.remove(c1));   // attempt to remove c1 again
+    ASSERT_LE(0, clients.remove(c2));
 
     ASSERT_TRUE(clients.isEmpty());
 }
diff --git a/vehicle/2.0/default/tests/VehicleObjectPool_test.cpp b/vehicle/2.0/default/tests/VehicleObjectPool_test.cpp
index 88b1be05d..135f9fa27 100644
--- a/vehicle/2.0/default/tests/VehicleObjectPool_test.cpp
+++ b/vehicle/2.0/default/tests/VehicleObjectPool_test.cpp
@@ -18,9 +18,10 @@
 
 #include <gtest/gtest.h>
 
-#include <vehicle_hal_manager/VehicleObjectPool.h>
 #include <utils/SystemClock.h>
 
+#include "vehicle_hal_manager/VehicleObjectPool.h"
+
 namespace android {
 namespace hardware {
 namespace vehicle {
diff --git a/vehicle/2.0/default/tests/VehiclePropConfigIndex_test.cpp b/vehicle/2.0/default/tests/VehiclePropConfigIndex_test.cpp
index aae7e62a8..28cdcbb80 100644
--- a/vehicle/2.0/default/tests/VehiclePropConfigIndex_test.cpp
+++ b/vehicle/2.0/default/tests/VehiclePropConfigIndex_test.cpp
@@ -16,7 +16,7 @@
 
 #include <gtest/gtest.h>
 
-#include <vehicle_hal_manager/VehiclePropConfigIndex.h>
+#include "vehicle_hal_manager/VehiclePropConfigIndex.h"
 
 #include "VehicleHalTestUtils.h"
 
diff --git a/vehicle/2.0/default/vehicle_hal_manager/ConcurrentQueue.h b/vehicle/2.0/default/vehicle_hal_manager/ConcurrentQueue.h
index 485f3dc38..8f575dc8b 100644
--- a/vehicle/2.0/default/vehicle_hal_manager/ConcurrentQueue.h
+++ b/vehicle/2.0/default/vehicle_hal_manager/ConcurrentQueue.h
@@ -108,20 +108,17 @@ public:
         mQueue = queue;
         mBatchInterval = batchInterval;
 
-        std::thread(&BatchingConsumer<T>::runInternal, this, func).detach();
+        mWorkerThread = std::thread(
+            &BatchingConsumer<T>::runInternal, this, func);
     }
 
     void requestStop() {
-        if (mState.exchange(State::STOP_REQUESTED) != State::RUNNING) {
-            mState = State::STOPPED;
-            mCondStopped.notify_one();
-        }
+        mState = State::STOP_REQUESTED;
     }
 
     void waitStopped() {
-        std::unique_lock<std::mutex> g(mLock);
-        while (State::STOPPED != mState) {
-            mCondStopped.wait(g);
+        if (mWorkerThread.joinable()) {
+            mWorkerThread.join();
         }
     }
 
@@ -144,12 +141,10 @@ private:
         }
 
         mState = State::STOPPED;
-        mCondStopped.notify_one();
     }
 
 private:
-    std::mutex mLock;
-    std::condition_variable mCondStopped;
+    std::thread mWorkerThread;
 
     std::atomic<State> mState;
     std::chrono::nanoseconds mBatchInterval;