Fix ring-buffer map consumer pointer on multiple subscribe/unsubscrib…

…e. (#3002)

* fix initialization

* fix typo

* fix initial subscription

* nit

* nit

* fix multiple subscription cases

* wip

* wip

* wip

* wip

* fix init for catch2

libs/api/ebpf_api.cpp

@@ -3882,7 +3882,8 @@ ebpf_ring_buffer_map_subscribe(
         ebpf_operation_ring_buffer_map_async_query_request_t async_query_request{
             sizeof(async_query_request),
             ebpf_operation_id_t::EBPF_OPERATION_RING_BUFFER_MAP_ASYNC_QUERY,
-            local_subscription->ring_buffer_map_handle};
+            local_subscription->ring_buffer_map_handle,
+            query_buffer_reply.consumer_offset};
         result = win32_error_code_to_ebpf_result(invoke_ioctl(
             async_query_request,
             local_subscription->reply,

libs/execution_context/ebpf_core.c

@@ -1831,7 +1831,8 @@ _ebpf_core_protocol_ring_buffer_map_query_buffer(
         goto Exit;
     }
 
-    result = ebpf_ring_buffer_map_query_buffer(map, (uint8_t**)(uintptr_t*)&reply->buffer_address);
+    result =
+        ebpf_ring_buffer_map_query_buffer(map, (uint8_t**)(uintptr_t*)&reply->buffer_address, &reply->consumer_offset);
 
 Exit:
     EBPF_OBJECT_RELEASE_REFERENCE((ebpf_core_object_t*)map);

libs/execution_context/ebpf_maps.c

@@ -1867,8 +1867,10 @@ _ebpf_ring_buffer_map_cancel_async_query(_In_ _Frees_ptr_ void* cancel_context)
 }
 
 _Must_inspect_result_ ebpf_result_t
-ebpf_ring_buffer_map_query_buffer(_In_ const ebpf_map_t* map, _Outptr_ uint8_t** buffer)
+ebpf_ring_buffer_map_query_buffer(_In_ const ebpf_map_t* map, _Outptr_ uint8_t** buffer, _Out_ size_t* consumer_offset)
 {
+    size_t producer_offset;
+    ebpf_ring_buffer_query((ebpf_ring_buffer_t*)map->data, consumer_offset, &producer_offset);
     return ebpf_ring_buffer_map_buffer((ebpf_ring_buffer_t*)map->data, buffer);
 }
 

libs/execution_context/ebpf_maps.h

@@ -192,11 +192,13 @@ extern "C"
      *
      * @param[in] map Ring buffer map to query.
      * @param[out] buffer Pointer to ring buffer data.
+     * @param[out] consumer_offset Offset of consumer in ring buffer data.
      * @retval EPBF_SUCCESS Successfully mapped the ring buffer.
      * @retval EBPF_INVALID_ARGUMENT Unable to map the ring buffer.
      */
     _Must_inspect_result_ ebpf_result_t
-    ebpf_ring_buffer_map_query_buffer(_In_ const ebpf_map_t* map, _Outptr_ uint8_t** buffer);
+    ebpf_ring_buffer_map_query_buffer(
+        _In_ const ebpf_map_t* map, _Outptr_ uint8_t** buffer, _Out_ size_t* consumer_offset);
 
     /**
      * @brief Return consumed buffer back to the ring buffer map.

libs/execution_context/ebpf_protocol.h

@@ -377,6 +377,8 @@ typedef struct _ebpf_operation_ring_buffer_map_query_buffer_reply
     struct _ebpf_operation_header header;
     // Address to user-space read-only buffer for the ring-buffer records.
     uint64_t buffer_address;
+    // The current consumer offset, so that subsequent reads can start from here.
+    size_t consumer_offset;
 } ebpf_operation_ring_buffer_map_query_buffer_reply_t;
 
 typedef struct _ebpf_operation_ring_buffer_map_async_query_request

libs/execution_context/unit/execution_context_unit_test.cpp

@@ -896,11 +896,13 @@ TEST_CASE("ring_buffer_async_query", "[execution_context]")
     struct _completion
     {
         uint8_t* buffer = nullptr;
+        size_t consumer_offset = 0;
         ebpf_ring_buffer_map_async_query_result_t async_query_result = {};
         uint64_t value{};
     } completion;
 
-    REQUIRE(ebpf_ring_buffer_map_query_buffer(map.get(), &completion.buffer) == EBPF_SUCCESS);
+    REQUIRE(
+        ebpf_ring_buffer_map_query_buffer(map.get(), &completion.buffer, &completion.consumer_offset) == EBPF_SUCCESS);
 
     REQUIRE(
         ebpf_async_set_completion_callback(

tests/end_to_end/end_to_end.cpp

@@ -924,12 +924,17 @@ bindmonitor_ring_buffer_test(ebpf_execution_type_t execution_type)
     std::function<ebpf_result_t(void*, uint32_t*)> invoke =
         [&hook](_Inout_ void* context, _Out_ uint32_t* result) -> ebpf_result_t { return hook.fire(context, result); };
 
-    ring_buffer_api_test_helper(process_map_fd, fake_app_ids, [&](int i) {
-        // Emulate bind operation.
-        std::vector<char> fake_app_id = fake_app_ids[i];
-        fake_app_id.push_back('\0');
-        REQUIRE(emulate_bind(invoke, fake_pid + i, fake_app_id.data()) == BIND_PERMIT);
-    });
+    // Test multiple subscriptions to the same ring buffer map, to ensure that the ring buffer map will continue
+    // to provide notifications to the subscriber.
+    for (int i = 0; i < 3; i++) {
+
+        ring_buffer_api_test_helper(process_map_fd, fake_app_ids, [&](int i) {
+            // Emulate bind operation.
+            std::vector<char> fake_app_id = fake_app_ids[i];
+            fake_app_id.push_back('\0');
+            REQUIRE(emulate_bind(invoke, fake_pid + i, fake_app_id.data()) == BIND_PERMIT);
+        });
+    }
 
     hook.detach_and_close_link(&link);