Merge remote-tracking branch 'origin/master'

.github/workflows/ci-low-cadence.yml

@@ -25,7 +25,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [ '8', '11', '17', '21-ea' ]
+        java: [ '8', '17', '21' ]
         os: [ 'ubuntu-22.04', 'windows-latest' ]
     steps:
       - name: Checkout code

.github/workflows/ci.yml

@@ -33,7 +33,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [ '8', '11', '17', '21-ea' ]
+        java: [ '8', '17', '21' ]
         os: [ 'ubuntu-22.04', 'windows-latest' ]
     steps:
       - name: Checkout code

aeron-client/src/main/c/aeron_client.c

@@ -217,7 +217,7 @@ void aeron_print_counters(aeron_t *client, void (*stream_out)(const char *))
         return;
     }
 
-    aeron_print_counters_stream_out_t out;
+    aeron_print_counters_stream_out_t out = { .stream_out = stream_out };
 
     aeron_counters_reader_foreach_counter(&client->conductor.counters_reader, aeron_print_counters_format, &out);
 }

aeron-client/src/main/c/aeronc.h

@@ -590,7 +590,8 @@ typedef struct aeron_counter_value_descriptor_stct
     int64_t counter_value;
     int64_t registration_id;
     int64_t owner_id;
-    uint8_t pad1[(2 * AERON_COUNTER_CACHE_LINE_LENGTH) - (3 * sizeof(int64_t))];
+    int64_t reference_id;
+    uint8_t pad1[(2 * AERON_COUNTER_CACHE_LINE_LENGTH) - (4 * sizeof(int64_t))];
 }
 aeron_counter_value_descriptor_t;
 
@@ -626,6 +627,7 @@ aeron_counter_metadata_descriptor_t;
 #define AERON_COUNTER_REGISTRATION_ID_DEFAULT INT64_C(0)
 #define AERON_COUNTER_NOT_FREE_TO_REUSE (INT64_MAX)
 #define AERON_COUNTER_OWNER_ID_DEFAULT INT64_C(0)
+#define AERON_COUNTER_REFERENCE_ID_DEFAULT INT64_C(0)
 
 #define AERON_NULL_COUNTER_ID (-1)
 
@@ -718,6 +720,18 @@ int aeron_counters_reader_counter_registration_id(
 int aeron_counters_reader_counter_owner_id(
     aeron_counters_reader_t *counters_reader, int32_t counter_id, int64_t *owner_id);
 
+/**
+ * Get the reference id assigned to a counter which typically be the registration id of an associated Image,
+ * Subscription, Publication, etc.
+ *
+ * @param counters_reader representing the this pointer.
+ * @param counter_id      for which the registration id requested.
+ * @param reference_id    pointer for value to be set on success.
+ * @return -1 on failure, 0 on success.
+ */
+int aeron_counters_reader_counter_reference_id(
+    aeron_counters_reader_t *counters_reader, int32_t counter_id, int64_t *reference_id);
+
 /**
  * Get the state for a counter.
  *

aeron-client/src/main/c/concurrent/aeron_counters_manager.c

@@ -137,6 +137,15 @@ void aeron_counters_manager_counter_owner_id(
     value_descriptor->owner_id = owner_id;
 }
 
+void aeron_counters_manager_counter_reference_id(
+    aeron_counters_manager_t *manager, int32_t counter_id, int64_t reference_id)
+{
+    aeron_counter_value_descriptor_t *value_descriptor = (aeron_counter_value_descriptor_t *)(
+        manager->values + AERON_COUNTER_OFFSET(counter_id));
+
+    value_descriptor->reference_id = reference_id;
+}
+
 void aeron_counters_manager_update_label(
     aeron_counters_manager_t *manager, int32_t counter_id, size_t label_length, const char *label)
 {
@@ -195,6 +204,7 @@ int32_t aeron_counters_manager_next_counter_id(aeron_counters_manager_t *manager
                     (manager->values + (counter_id * AERON_COUNTERS_MANAGER_VALUE_LENGTH));
                 AERON_PUT_ORDERED(value->registration_id, AERON_COUNTER_REGISTRATION_ID_DEFAULT);
                 value->owner_id = AERON_COUNTER_OWNER_ID_DEFAULT;
+                value->reference_id = AERON_COUNTER_REFERENCE_ID_DEFAULT;
                 AERON_PUT_ORDERED(value->counter_value, INT64_C(0));
 
                 return counter_id;
@@ -363,6 +373,22 @@ int aeron_counters_reader_counter_owner_id(
     return 0;
 }
 
+int aeron_counters_reader_counter_reference_id(
+    aeron_counters_reader_t *counters_reader, int32_t counter_id, int64_t *reference_id)
+{
+    if (counter_id < 0 || counter_id > counters_reader->max_counter_id)
+    {
+        return -1;
+    }
+
+    aeron_counter_value_descriptor_t *value_descriptor = (aeron_counter_value_descriptor_t *)(
+        counters_reader->values + AERON_COUNTER_OFFSET(counter_id));
+
+    *reference_id = value_descriptor->reference_id;
+
+    return 0;
+}
+
 int aeron_counters_reader_counter_state(aeron_counters_reader_t *counters_reader, int32_t counter_id, int32_t *state)
 {
     if (counter_id < 0 || counter_id > counters_reader->max_counter_id)

aeron-client/src/main/c/concurrent/aeron_counters_manager.h

@@ -125,6 +125,9 @@ void aeron_counters_manager_counter_registration_id(
 void aeron_counters_manager_counter_owner_id(
     aeron_counters_manager_t *manager, int32_t counter_id, int64_t owner_id);
 
+void aeron_counters_manager_counter_reference_id(
+    aeron_counters_manager_t *manager, int32_t counter_id, int64_t reference_id);
+
 void aeron_counters_manager_update_label(
     aeron_counters_manager_t *manager, int32_t counter_id, size_t label_length, const char *label);
 

aeron-client/src/test/c/concurrent/aeron_counters_test.cpp

@@ -235,4 +235,47 @@ TEST_F(CountersTest, shouldResetValueAndOwnerIdIfReused)
     aeron_counters_manager_counter_owner_id(&m_manager, id_two, expected_owner_id_two);
     EXPECT_EQ(0, aeron_counters_reader_counter_owner_id(&m_reader, id_two, &owner_id_two));
     EXPECT_EQ(expected_owner_id_two, owner_id_two);
-}
+}
+
+TEST_F(CountersTest, shouldSetReferenceId)
+{
+    int32_t id = aeron_counters_manager_allocate(&m_manager, 0, nullptr, 0, nullptr, 0);
+
+    int64_t reference_id = 999;
+    EXPECT_EQ(0, aeron_counters_reader_counter_reference_id(&m_reader, id, &reference_id));
+    EXPECT_EQ(AERON_COUNTER_REFERENCE_ID_DEFAULT, reference_id);
+
+    int64_t expected_reference_id = 777;
+    aeron_counters_manager_counter_reference_id(&m_manager, id, expected_reference_id);
+    EXPECT_EQ(0, aeron_counters_reader_counter_reference_id(&m_reader, id, &reference_id));
+    EXPECT_EQ(expected_reference_id, reference_id);
+}
+
+TEST_F(CountersTest, shouldResetValueAndReferenceIdIfReused)
+{
+    int32_t id_one = aeron_counters_manager_allocate(&m_manager, 0, nullptr, 0, nullptr, 0);
+
+    int64_t reference_id_one = 999;
+    EXPECT_EQ(0, aeron_counters_reader_counter_reference_id(&m_reader, id_one, &reference_id_one));
+    EXPECT_EQ(AERON_COUNTER_REGISTRATION_ID_DEFAULT, reference_id_one);
+
+    int64_t expected_reference_id_one = 777;
+    aeron_counters_manager_counter_reference_id(&m_manager, id_one, expected_reference_id_one);
+    EXPECT_EQ(0, aeron_counters_reader_counter_reference_id(&m_reader, id_one, &reference_id_one));
+    EXPECT_EQ(expected_reference_id_one, reference_id_one);
+
+    m_manager.free_to_reuse_timeout_ms = 0;
+    aeron_counters_manager_free(&m_manager, id_one);
+
+    int32_t id_two = aeron_counters_manager_allocate(&m_manager, 0, nullptr, 0, nullptr, 0);
+
+    int64_t reference_id_two = 999;
+    EXPECT_EQ(0, aeron_counters_reader_counter_reference_id(&m_reader, id_one, &reference_id_two));
+    EXPECT_EQ(id_one, id_two);
+    EXPECT_EQ(AERON_COUNTER_REGISTRATION_ID_DEFAULT, reference_id_two);
+
+    int64_t expected_reference_id_two = 333;
+    aeron_counters_manager_counter_reference_id(&m_manager, id_two, expected_reference_id_two);
+    EXPECT_EQ(0, aeron_counters_reader_counter_reference_id(&m_reader, id_two, &reference_id_two));
+    EXPECT_EQ(expected_reference_id_two, reference_id_two);
+}

aeron-driver/src/main/c/aeron_driver_conductor.c

@@ -3207,7 +3207,10 @@ int aeron_driver_conductor_link_subscribable(
 
         if (counter_id >= 0)
         {
-            aeron_counters_manager_counter_owner_id(&conductor->counters_manager, counter_id, link->client_id);
+            aeron_counters_manager_counter_owner_id(
+                &conductor->counters_manager, counter_id, link->client_id);
+            aeron_counters_manager_counter_reference_id(
+                &conductor->counters_manager, counter_id, original_registration_id);
             int64_t *position_addr = aeron_counters_manager_addr(&conductor->counters_manager, counter_id);
 
             if (aeron_driver_subscribable_add_position(subscribable, link, counter_id, position_addr, now_ns) >= 0)

aeron-driver/src/main/c/agent/aeron_driver_agent.c

@@ -72,7 +72,7 @@ static aeron_thread_t log_reader_thread;
 static aeron_driver_agent_dynamic_dissector_entry_t *dynamic_dissector_entries = NULL;
 static size_t num_dynamic_dissector_entries = 0;
 static int64_t dynamic_dissector_index = 0;
-static struct aeron_driver_agent_log_event_stct log_events[AERON_DRIVER_EVENT_NUM_ELEMENTS] =
+static aeron_driver_agent_log_event_t log_events[] =
     {
         { AERON_DRIVER_AGENT_EVENT_UNKNOWN_NAME,  AERON_DRIVER_AGENT_EVENT_TYPE_UNKNOWN, false },
         { "FRAME_IN",                             AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
@@ -127,16 +127,17 @@ static struct aeron_driver_agent_log_event_stct log_events[AERON_DRIVER_EVENT_NU
         { "NAME_RESOLUTION_RESOLVE",              AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
         { "GENERIC_MESSAGE",                      AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
         { "NAME_RESOLUTION_LOOKUP",               AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
-        { "NAME_RESOLUTION_HOST_NAME",               AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
+        { "NAME_RESOLUTION_HOST_NAME",            AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
         { "ADD_DYNAMIC_DISSECTOR",                AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
         { "DYNAMIC_DISSECTOR_EVENT",              AERON_DRIVER_AGENT_EVENT_TYPE_OTHER,   false },
     };
 
-typedef struct aeron_driver_agent_name_resolver_state_stct
+#define AERON_DRIVER_EVENT_NUM_ELEMENTS (sizeof(log_events) / sizeof(aeron_driver_agent_log_event_t))
+
+size_t aeron_driver_agent_max_event_count(void)
 {
-    aeron_name_resolver_resolve_func_t delegate_resolve_func;
+    return AERON_DRIVER_EVENT_NUM_ELEMENTS;
 }
-aeron_driver_agent_name_resolver_state_t;
 
 aeron_mpsc_rb_t *aeron_driver_agent_mpsc_rb(void)
 {
@@ -189,7 +190,7 @@ void aeron_driver_agent_logging_ring_buffer_init(void)
 
     if (aeron_mpsc_rb_init(&logging_mpsc_rb, rb_buffer, rb_length) < 0)
     {
-        fprintf(stderr, "could not init logging mpwc_rb. exiting.\n");
+        fprintf(stderr, "could not init logging mpsc_rb. exiting.\n");
         exit(EXIT_FAILURE);
     }
 }
@@ -216,7 +217,7 @@ static aeron_driver_agent_event_t aeron_driver_agent_event_name_to_id(const char
         return AERON_DRIVER_EVENT_UNKNOWN_EVENT;
     }
 
-    for (int i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
+    for (size_t i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
     {
         const char *name = log_events[i].name;
         if (0 == strncmp(name, event_name, strlen(name) + 1))
@@ -230,7 +231,7 @@ static aeron_driver_agent_event_t aeron_driver_agent_event_name_to_id(const char
 
 static inline bool is_valid_event_id(const int id)
 {
-    return id >= 0 && id < AERON_DRIVER_EVENT_NUM_ELEMENTS;
+    return id >= 0 && id < (int)AERON_DRIVER_EVENT_NUM_ELEMENTS;
 }
 
 const char *aeron_driver_agent_event_name(const aeron_driver_agent_event_t id)
@@ -250,7 +251,7 @@ bool aeron_driver_agent_is_event_enabled(const aeron_driver_agent_event_t id)
 
 static void aeron_driver_agent_set_enabled_all_events(const bool is_enabled)
 {
-    for (int i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
+    for (size_t i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
     {
         const char *event_name = log_events[i].name;
         if (!aeron_driver_agent_is_unknown_event(event_name))
@@ -262,7 +263,7 @@ static void aeron_driver_agent_set_enabled_all_events(const bool is_enabled)
 
 static void aeron_driver_agent_set_enabled_admin_events(const bool is_enabled)
 {
-    for (int i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
+    for (size_t i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
     {
         if (AERON_DRIVER_EVENT_FRAME_IN != i &&
             AERON_DRIVER_EVENT_FRAME_OUT != i &&
@@ -280,7 +281,7 @@ static void aeron_driver_agent_set_enabled_admin_events(const bool is_enabled)
 
 static void aeron_driver_agent_set_enabled_specific_events(const uint8_t type, const bool is_enabled)
 {
-    for (int i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
+    for (size_t i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
     {
         if (type == log_events[i].type)
         {
@@ -291,7 +292,7 @@ static void aeron_driver_agent_set_enabled_specific_events(const uint8_t type, c
 
 static bool any_event_enabled(const uint8_t type)
 {
-    for (int i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
+    for (size_t i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
     {
         if (type == log_events[i].type && log_events[i].enabled)
         {
@@ -474,7 +475,7 @@ bool aeron_driver_agent_logging_events_init(const char *event_log, const char *e
 
 void aeron_driver_agent_logging_events_free(void)
 {
-    for (int i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
+    for (size_t i = 0; i < AERON_DRIVER_EVENT_NUM_ELEMENTS; i++)
     {
         log_events[i].enabled = false;
     }

aeron-driver/src/main/c/agent/aeron_driver_agent.h

@@ -33,6 +33,7 @@
 
 typedef enum aeron_driver_agent_event_enum
 {
+    AERON_DRIVER_EVENT_UNKNOWN_EVENT = -1,
     AERON_DRIVER_EVENT_FRAME_IN = 1,
     AERON_DRIVER_EVENT_FRAME_OUT = 2,
     AERON_DRIVER_EVENT_CMD_IN_ADD_PUBLICATION = 3,
@@ -77,11 +78,8 @@ typedef enum aeron_driver_agent_event_enum
     AERON_DRIVER_EVENT_NAME_RESOLUTION_HOST_NAME = 53,
 
     // C-specific events. Note: event IDs are dynamic to avoid gaps in the sparse arrays.
-    AERON_DRIVER_EVENT_ADD_DYNAMIC_DISSECTOR,
-    AERON_DRIVER_EVENT_DYNAMIC_DISSECTOR_EVENT,
-
-    AERON_DRIVER_EVENT_NUM_ELEMENTS, // number of elements in this enum (including gaps)
-    AERON_DRIVER_EVENT_UNKNOWN_EVENT = -1
+    AERON_DRIVER_EVENT_ADD_DYNAMIC_DISSECTOR = 54,
+    AERON_DRIVER_EVENT_DYNAMIC_DISSECTOR_EVENT = 55
 }
 aeron_driver_agent_event_t;
 
@@ -209,6 +207,8 @@ typedef int (*aeron_driver_context_init_t)(aeron_driver_context_t **);
 
 int aeron_driver_agent_context_init(aeron_driver_context_t *context);
 
+size_t aeron_driver_agent_max_event_count(void);
+
 const char *aeron_driver_agent_dissect_log_header(
     int64_t time_ns,
     aeron_driver_agent_event_t event_id,