Fix data race in the umfIpcOpenedCacheDestroy function

Author: vinser52

src/ipc_cache.c

@@ -144,6 +144,8 @@ umfIpcOpenedCacheCreate(ipc_opened_cache_eviction_cb_t eviction_cb) {
 
 void umfIpcOpenedCacheDestroy(ipc_opened_cache_handle_t cache) {
     ipc_opened_cache_entry_t *entry, *tmp;
+
+    utils_mutex_lock(&(cache->global->cache_lock));
     HASH_ITER(hh, cache->hash_table, entry, tmp) {
         DL_DELETE(cache->global->lru_list, entry);
         HASH_DEL(cache->hash_table, entry);
@@ -153,6 +155,7 @@ void umfIpcOpenedCacheDestroy(ipc_opened_cache_handle_t cache) {
         umf_ba_free(cache->global->cache_allocator, entry);
     }
     HASH_CLEAR(hh, cache->hash_table);
+    utils_mutex_unlock(&(cache->global->cache_lock));
 
     umf_ba_global_free(cache);
 }

test/ipcFixtures.hpp

@@ -593,4 +593,73 @@ TEST_P(umfIpcTest, ConcurrentOpenCloseHandles) {
     EXPECT_EQ(stat.openCount, stat.closeCount);
 }
 
+TEST_P(umfIpcTest, ConcurrentDestroyIpcHandlers) {
+    constexpr size_t SIZE = 100;
+    constexpr size_t NUM_ALLOCS = 100;
+    constexpr size_t NUM_POOLS = 10;
+    void *ptrs[NUM_ALLOCS];
+    void *openedPtrs[NUM_POOLS][NUM_ALLOCS];
+    std::vector<umf::pool_unique_handle_t> consumerPools;
+    umf::pool_unique_handle_t producerPool = makePool();
+    ASSERT_NE(producerPool.get(), nullptr);
+
+    for (size_t i = 0; i < NUM_POOLS; ++i) {
+        consumerPools.push_back(makePool());
+    }
+
+    for (size_t i = 0; i < NUM_ALLOCS; ++i) {
+        void *ptr = umfPoolMalloc(producerPool.get(), SIZE);
+        ASSERT_NE(ptr, nullptr);
+        ptrs[i] = ptr;
+    }
+
+    for (size_t i = 0; i < NUM_ALLOCS; ++i) {
+        umf_ipc_handle_t ipcHandle = nullptr;
+        size_t handleSize = 0;
+        umf_result_t ret = umfGetIPCHandle(ptrs[i], &ipcHandle, &handleSize);
+        ASSERT_EQ(ret, UMF_RESULT_SUCCESS);
+
+        for (size_t poolId = 0; poolId < NUM_POOLS; poolId++) {
+            void *ptr = nullptr;
+            umf_ipc_handler_handle_t ipcHandler = nullptr;
+            ret =
+                umfPoolGetIPCHandler(consumerPools[poolId].get(), &ipcHandler);
+            ASSERT_EQ(ret, UMF_RESULT_SUCCESS);
+            ASSERT_NE(ipcHandler, nullptr);
+
+            ret = umfOpenIPCHandle(ipcHandler, ipcHandle, &ptr);
+            ASSERT_EQ(ret, UMF_RESULT_SUCCESS);
+            openedPtrs[poolId][i] = ptr;
+        }
+
+        ret = umfPutIPCHandle(ipcHandle);
+        ASSERT_EQ(ret, UMF_RESULT_SUCCESS);
+    }
+
+    for (size_t poolId = 0; poolId < NUM_POOLS; poolId++) {
+        for (size_t i = 0; i < NUM_ALLOCS; ++i) {
+            umf_result_t ret = umfCloseIPCHandle(openedPtrs[poolId][i]);
+            EXPECT_EQ(ret, UMF_RESULT_SUCCESS);
+        }
+    }
+
+    for (size_t i = 0; i < NUM_ALLOCS; ++i) {
+        umf_result_t ret = umfFree(ptrs[i]);
+        EXPECT_EQ(ret, UMF_RESULT_SUCCESS);
+    }
+
+    // Destroy pools in parallel to cause IPC cache cleanup in parallel.
+    umf_test::syncthreads_barrier syncthreads(NUM_POOLS);
+    auto poolDestroyFn = [&consumerPools, &syncthreads](size_t tid) {
+        syncthreads();
+        consumerPools[tid].reset(nullptr);
+    };
+    umf_test::parallel_exec(NUM_POOLS, poolDestroyFn);
+
+    producerPool.reset(nullptr);
+
+    EXPECT_EQ(stat.putCount, stat.getCount);
+    EXPECT_EQ(stat.openCount, stat.closeCount);
+}
+
 #endif /* UMF_TEST_IPC_FIXTURES_HPP */