[C]: update time of last position change and last activity at publication creation. Reorder checks for efficiency. Issue #377.

Author: tmontgomery

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

@@ -1323,10 +1323,11 @@ int aeron_driver_subscribeable_add_position(
 
     if (ensure_capacity_result >= 0)
     {
-        aeron_position_t *entry = &subscribeable->array[subscribeable->length++];
+        aeron_position_t *entry = &subscribeable->array[subscribeable->length];
         entry->counter_id = counter_id;
         entry->value_addr = value_addr;
         subscribeable->add_position_hook_func(subscribeable->clientd, value_addr);
+        subscribeable->length++;
         result = 0;
     }
 

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

@@ -44,6 +44,7 @@ int aeron_ipc_publication_create(
     aeron_ipc_publication_t *_pub = NULL;
     const uint64_t usable_fs_space = context->usable_fs_space_func(context->aeron_dir);
     const uint64_t log_length = AERON_LOGBUFFER_COMPUTE_LOG_LENGTH(term_buffer_length);
+    const int64_t now_ns = context->nano_clock();
 
     *publication = NULL;
 
@@ -90,10 +91,11 @@ int aeron_ipc_publication_create(
     aeron_logbuffer_fill_default_header(
         _pub->mapped_raw_log.log_meta_data.addr, session_id, stream_id, initial_term_id);
 
+    _pub->nano_clock = context->nano_clock;
     _pub->conductor_fields.subscribeable.array = NULL;
     _pub->conductor_fields.subscribeable.length = 0;
     _pub->conductor_fields.subscribeable.capacity = 0;
-    _pub->conductor_fields.subscribeable.add_position_hook_func = aeron_driver_subscribeable_null_hook;
+    _pub->conductor_fields.subscribeable.add_position_hook_func = aeron_ipc_publication_add_subscriber_hook;
     _pub->conductor_fields.subscribeable.remove_position_hook_func = aeron_ipc_publication_remove_subscriber_hook;
     _pub->conductor_fields.subscribeable.clientd = _pub;
     _pub->conductor_fields.managed_resource.registration_id = registration_id;
@@ -105,7 +107,7 @@ int aeron_ipc_publication_create(
     _pub->conductor_fields.trip_limit = 0;
     _pub->conductor_fields.consumer_position = 0;
     _pub->conductor_fields.last_consumer_position = 0;
-    _pub->conductor_fields.time_of_last_consumer_position_change = 0;
+    _pub->conductor_fields.time_of_last_consumer_position_change = now_ns;
     _pub->conductor_fields.status = AERON_IPC_PUBLICATION_STATUS_ACTIVE;
     _pub->conductor_fields.refcnt = 1;
     _pub->session_id = session_id;
@@ -238,8 +240,8 @@ void aeron_ipc_publication_check_for_blocked_publisher(aeron_ipc_publication_t *
 {
     if (publication->conductor_fields.consumer_position == publication->conductor_fields.last_consumer_position)
     {
-        if (aeron_ipc_publication_producer_position(publication) > publication->conductor_fields.consumer_position &&
-            now_ns > (publication->conductor_fields.time_of_last_consumer_position_change + publication->unblock_timeout_ns))
+        if (now_ns > (publication->conductor_fields.time_of_last_consumer_position_change + publication->unblock_timeout_ns) &&
+            aeron_ipc_publication_producer_position(publication) > publication->conductor_fields.consumer_position)
         {
             if (aeron_logbuffer_unblocker_unblock(
                 publication->mapped_raw_log.term_buffers,
@@ -257,6 +259,7 @@ void aeron_ipc_publication_check_for_blocked_publisher(aeron_ipc_publication_t *
     }
 }
 
+extern void aeron_ipc_publication_add_subscriber_hook(void *clientd, int64_t *value_addr);
 extern void aeron_ipc_publication_remove_subscriber_hook(void *clientd, int64_t *value_addr);
 extern int64_t aeron_ipc_publication_producer_position(aeron_ipc_publication_t *publication);
 extern int64_t aeron_ipc_publication_joining_position(aeron_ipc_publication_t *publication);

aeron-driver/src/main/c/aeron_ipc_publication.h

@@ -54,6 +54,7 @@ typedef struct aeron_ipc_publication_stct
     aeron_mapped_raw_log_t mapped_raw_log;
     aeron_position_t pub_lmt_position;
     aeron_logbuffer_metadata_t *log_meta_data;
+    aeron_clock_func_t nano_clock;
 
     char *log_file_name;
     int64_t term_window_length;
@@ -98,6 +99,17 @@ void aeron_ipc_publication_decref(void *clientd);
 
 void aeron_ipc_publication_check_for_blocked_publisher(aeron_ipc_publication_t *publication, int64_t now_ns);
 
+inline void aeron_ipc_publication_add_subscriber_hook(void *clientd, int64_t *value_addr)
+{
+    aeron_ipc_publication_t *publication = (aeron_ipc_publication_t *)clientd;
+
+    /* 0 to 1 transition */
+    if (0 == publication->conductor_fields.subscribeable.length)
+    {
+        publication->conductor_fields.time_of_last_consumer_position_change = publication->nano_clock();
+    }
+}
+
 inline void aeron_ipc_publication_remove_subscriber_hook(void *clientd, int64_t *value_addr)
 {
     aeron_ipc_publication_t *publication = (aeron_ipc_publication_t *)clientd;

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

@@ -144,7 +144,7 @@ int aeron_network_publication_create(
     _pub->conductor_fields.clean_position = 0;
     _pub->conductor_fields.status = AERON_NETWORK_PUBLICATION_STATUS_ACTIVE;
     _pub->conductor_fields.refcnt = 1;
-    _pub->conductor_fields.time_of_last_activity_ns = 0;
+    _pub->conductor_fields.time_of_last_activity_ns = now_ns;
     _pub->conductor_fields.last_snd_pos = 0;
     _pub->session_id = session_id;
     _pub->stream_id = stream_id;
@@ -639,8 +639,8 @@ void aeron_network_publication_check_for_blocked_publisher(
 {
     if (snd_pos == publication->conductor_fields.last_snd_pos)
     {
-        if (aeron_network_publication_producer_position(publication) > snd_pos &&
-            now_ns > (publication->conductor_fields.time_of_last_activity_ns + publication->unblock_timeout_ns))
+        if (now_ns > (publication->conductor_fields.time_of_last_activity_ns + publication->unblock_timeout_ns) &&
+            aeron_network_publication_producer_position(publication) > snd_pos)
         {
             if (aeron_logbuffer_unblocker_unblock(
                 publication->mapped_raw_log.term_buffers,