Merge pull request #998 from vinser52/svinogra_ipc_cache

Implement size limit for the cache of opened IPC handles

Author: bratpiorka

.cmake-format

@@ -26,7 +26,8 @@ with section("parse"):
       'kwargs': {
         'NAME': '*',
         'SRCS': '*',
-        'LIBS': '*'}},
+        'LIBS': '*',
+        'ENVS': '*'}},
     'add_umf_library': {
       "pargs": 0,
       "flags": [],

.github/workflows/reusable_compatibility.yml

@@ -94,9 +94,11 @@ jobs:
 
     - name: Run "tag" UMF tests with latest UMF libs (warnings enabled)
       working-directory: ${{github.workspace}}/tag_version/build
+      # GTEST_FILTER is used below to skip test that is not compatible
       run: >
         UMF_LOG="level:warning;flush:debug;output:stderr;pid:no"
         LD_LIBRARY_PATH=${{github.workspace}}/latest_version/build/lib/
+        GTEST_FILTER="-*umfIpcTest.GetPoolByOpenedHandle*"
         ctest --verbose
 
   windows:
@@ -181,6 +183,7 @@ jobs:
       working-directory: ${{github.workspace}}/tag_version/build
       run: |
         $env:UMF_LOG="level:warning;flush:debug;output:stderr;pid:no" 
+        $env:GTEST_FILTER="-*umfIpcTest.GetPoolByOpenedHandle*"
         cp ${{github.workspace}}/latest_version/build/bin/Debug/umf.dll ${{github.workspace}}/tag_version/build/bin/Debug/umf.dll
         ctest -C Debug --verbose
 
@@ -230,8 +233,10 @@ jobs:
 
     - name: Run "tag" UMF tests
       working-directory: ${{github.workspace}}/tag_version/build
-      run: |
-        LD_LIBRARY_PATH=${{github.workspace}}/tag_version/build/lib/ ctest --output-on-failure
+      run: >
+        LD_LIBRARY_PATH=${{github.workspace}}/tag_version/build/lib/
+        GTEST_FILTER="-*umfIpcTest.GetPoolByOpenedHandle*"
+        ctest --output-on-failure
 
     - name: Checkout latest UMF version
       uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
@@ -266,4 +271,5 @@ jobs:
       run: >
         UMF_LOG="level:warning;flush:debug;output:stderr;pid:no"
         LD_LIBRARY_PATH=${{github.workspace}}/latest_version/build/lib/
+        GTEST_FILTER="-*umfIpcTest.GetPoolByOpenedHandle*"
         ctest --verbose -E "not_impl"

docs/config/api.rst

@@ -168,6 +168,26 @@ IPC API allows retrieving IPC handles for the memory buffers allocated from
 UMF memory pools. The memory provider used by the pool should support IPC 
 operations for this API to work. Otherwise IPC APIs return an error.
 
+IPC caching
+------------------------------------------
+
+UMF employs IPC caching to avoid multiple IPC handles being created for the same 
+coarse-grain memory region allocated by the memory provider. UMF guarantees that 
+for each coarse-grain memory region allocated by the memory provider, only one 
+IPC handle is created when the :any:`umfGetIPCHandle` function is called. All 
+subsequent calls to the :any:`umfGetIPCHandle` function for the pointer to the 
+same memory region will return the entry from the cache.
+
+The same is true for the :any:`umfOpenIPCHandle` function. The actual mapping
+of the IPC handle to the virtual address space is created only once, and all
+subsequent calls to open the same IPC handle will return the entry from the cache.
+The size of the cache for opened IPC handles is controlled by the ``UMF_MAX_OPENED_IPC_HANDLES``
+environment variable. By default, the cache size is unlimited. However, if the environment 
+variable is set and the cache size exceeds the limit, old items will be evicted. UMF tracks 
+the ref count for each entry in the cache and can evict only items with the ref count equal to 0. 
+The ref count is increased when the :any:`umfOpenIPCHandle` function is called and decreased 
+when the :any:`umfCloseIPCHandle` function is called for the corresponding IPC handle.
+
 .. _ipc-api:
 
 IPC API

src/ipc.c

@@ -146,19 +146,15 @@ umf_result_t umfOpenIPCHandle(umf_ipc_handler_handle_t hIPCHandler,
 }
 
 umf_result_t umfCloseIPCHandle(void *ptr) {
-    umf_alloc_info_t allocInfo;
-    umf_result_t ret = umfMemoryTrackerGetAllocInfo(ptr, &allocInfo);
+    umf_ipc_info_t ipcInfo;
+    umf_result_t ret = umfMemoryTrackerGetIpcInfo(ptr, &ipcInfo);
     if (ret != UMF_RESULT_SUCCESS) {
-        LOG_ERR("cannot get alloc info for ptr = %p.", ptr);
+        LOG_ERR("cannot get IPC info for ptr = %p.", ptr);
         return ret;
     }
 
-    // We cannot use umfPoolGetMemoryProvider function because it returns
-    // upstream provider but we need tracking one
-    umf_memory_provider_handle_t hProvider = allocInfo.pool->provider;
-
-    return umfMemoryProviderCloseIPCHandle(hProvider, allocInfo.base,
-                                           allocInfo.baseSize);
+    return umfMemoryProviderCloseIPCHandle(ipcInfo.provider, ipcInfo.base,
+                                           ipcInfo.baseSize);
 }
 
 umf_result_t umfPoolGetIPCHandler(umf_memory_pool_handle_t hPool,

src/ipc_cache.c

@@ -54,6 +54,22 @@ typedef struct ipc_opened_cache_t {
 
 ipc_opened_cache_global_t *IPC_OPENED_CACHE_GLOBAL = NULL;
 
+// Returns value of the UMF_MAX_OPENED_IPC_HANDLES environment variable
+// or 0 if it is not set.
+static size_t umfIpcCacheGlobalInitMaxOpenedHandles(void) {
+    const char *max_size_str = getenv("UMF_MAX_OPENED_IPC_HANDLES");
+    if (max_size_str) {
+        char *endptr;
+        size_t max_size = strtoul(max_size_str, &endptr, 10);
+        if (*endptr == '\0') {
+            return max_size;
+        }
+        LOG_ERR("Invalid value of UMF_MAX_OPENED_IPC_HANDLES: %s",
+                max_size_str);
+    }
+    return 0;
+}
+
 umf_result_t umfIpcCacheGlobalInit(void) {
     umf_result_t ret = UMF_RESULT_SUCCESS;
     ipc_opened_cache_global_t *cache_global =
@@ -78,8 +94,7 @@ umf_result_t umfIpcCacheGlobalInit(void) {
         goto err_mutex_destroy;
     }
 
-    // TODO: make max_size configurable via environment variable
-    cache_global->max_size = 0;
+    cache_global->max_size = umfIpcCacheGlobalInitMaxOpenedHandles();
     cache_global->cur_size = 0;
     cache_global->lru_list = NULL;
 
@@ -191,7 +206,19 @@ umf_result_t umfIpcOpenedCacheGet(ipc_opened_cache_handle_t cache,
         if (entry == NULL && cache->global->max_size != 0 &&
             cache->global->cur_size >= cache->global->max_size) {
             // If max_size is set and the cache is full, evict the least recently used entry.
-            entry = cache->global->lru_list->prev;
+            // we need to search for the least recently used entry with ref_count == 0
+            // The utlist implementation of the doubly-linked list keeps a tail pointer in head->prev
+            ipc_opened_cache_entry_t *candidate = cache->global->lru_list->prev;
+            do {
+                uint64_t ref_count = 0;
+                utils_atomic_load_acquire_u64(&candidate->ref_count,
+                                              &ref_count);
+                if (ref_count == 0) {
+                    entry = candidate;
+                    break;
+                }
+                candidate = candidate->prev;
+            } while (candidate != cache->global->lru_list->prev);
         }
 
         if (entry) { // we have eviction candidate
@@ -244,3 +271,20 @@ umf_result_t umfIpcOpenedCacheGet(ipc_opened_cache_handle_t cache,
 
     return ret;
 }
+
+umf_result_t
+umfIpcHandleMappedCacheRelease(ipc_opened_cache_value_t *cacheValue) {
+    if (!cacheValue) {
+        LOG_ERR("cacheValue is NULL");
+        return UMF_RESULT_ERROR_INVALID_ARGUMENT;
+    }
+
+    // get pointer to the entry
+    ipc_opened_cache_entry_t *entry =
+        (ipc_opened_cache_entry_t *)((char *)cacheValue -
+                                     offsetof(ipc_opened_cache_entry_t, value));
+    // decrement the ref count
+    utils_atomic_decrement_u64(&entry->ref_count);
+
+    return UMF_RESULT_SUCCESS;
+}

src/ipc_cache.h

@@ -47,4 +47,6 @@ umf_result_t umfIpcOpenedCacheGet(ipc_opened_cache_handle_t cache,
                                   uint64_t handle_id,
                                   ipc_opened_cache_value_t **retEntry);
 
+umf_result_t
+umfIpcHandleMappedCacheRelease(ipc_opened_cache_value_t *cacheValue);
 #endif /* UMF_IPC_CACHE_H */