Add support for timeout notification callbacks.

When building an application (such as a server) that uses multiple
messagers, managing calls to curvecpr_messager_next_timeout() can be
tedious. This change introduces an optional callback-based mechanism for
handling updates to timeouts.

Also, there's a CHANGELOG now.

Author: impl

CHANGELOG.md

@@ -0,0 +1,15 @@
+## v0.1.2
+
+* Add support for a custom callback to receive timeouts for the messager,
+  eliminating the need to repeatedly poll every messager instance
+  (particularly useful in servers).
+* Increment the major component of the shared library version due to ABI
+  compatibility break.
+
+## v0.1.1
+
+* Include `check_extras.h` in `EXTRA_DIST` so `make check` executes correctly.
+
+## v0.1.0
+
+* Initial release.

configure.ac

@@ -7,7 +7,7 @@ AC_CONFIG_HEADER([config.h])
 AC_CONFIG_MACRO_DIR([m4])
 
 # Library version.
-CURVECPR_LIBRARY_VERSION=1:0:0
+CURVECPR_LIBRARY_VERSION=2:0:0
 AC_SUBST(CURVECPR_LIBRARY_VERSION)
 
 # Checks for programs.

libcurvecpr/include/curvecpr/messager.h

@@ -28,6 +28,8 @@ struct curvecpr_messager_ops {
     int (*recvmarkq_remove_range)(struct curvecpr_messager *messager, unsigned long long start, unsigned long long end);
 
     int (*send)(struct curvecpr_messager *messager, const unsigned char *buf, size_t num);
+
+    void (*put_next_timeout)(struct curvecpr_messager *messager, const long long timeout_ns);
 };
 
 struct curvecpr_messager_cf {

libcurvecpr/lib/messager.c

@@ -61,6 +61,9 @@ void curvecpr_messager_new (struct curvecpr_messager *messager, const struct cur
     /* If we're in client mode, initiate packets have a maximum size of 512 bytes.
        Otherwise, we're in server mode, and we can start at 1024. */
     messager->my_maximum_send_bytes = client ? 512 : 1024;
+
+    /* Fire off initial timeout notification. */
+    curvecpr_messager_next_timeout(messager);
 }
 
 int curvecpr_messager_recv (struct curvecpr_messager *messager, const unsigned char *buf, size_t num)
@@ -197,6 +200,9 @@ int curvecpr_messager_recv (struct curvecpr_messager *messager, const unsigned c
         }
     }
 
+    /* Update timeout (if callback defined). */
+    curvecpr_messager_next_timeout(messager);
+
     /* Update acknowledgment information (but only if this isn't a pure
        acknowledgment). */
     if (id) {
@@ -418,6 +424,9 @@ static int _send_block (struct curvecpr_messager *messager, struct curvecpr_bloc
     /* Reset last received ID so we don't acknowledge an old message. */
     messager->their_sent_id = 0;
 
+    /* Update timeout (if callback defined). */
+    curvecpr_messager_next_timeout(messager);
+
     return 0;
 }
 
@@ -479,7 +488,7 @@ long long curvecpr_messager_next_timeout (struct curvecpr_messager *messager)
 
     struct curvecpr_block *block = NULL;
 
-    long long at;
+    long long at, timeout;
 
     curvecpr_chicago_refresh_clock(chicago);
 
@@ -505,7 +514,12 @@ long long curvecpr_messager_next_timeout (struct curvecpr_messager *messager)
     /* If the current time is after the next action time, the timeout is 0. However, we
        always have at least a 1 millisecond timeout to prevent the CPU from spinning. */
     if (chicago->clock > at)
-        return 1000000;
+        timeout = 1000000;
     else
-        return at - chicago->clock + 1000000;
+        timeout = at - chicago->clock + 1000000;
+
+    if (cf->ops.put_next_timeout)
+        cf->ops.put_next_timeout(messager, timeout);
+
+    return timeout;
 }

libcurvecpr/test/Makefile.am

@@ -15,6 +15,9 @@ messager_test_recv_requests_removal_from_sendmarkq_SOURCES = messager/test_recv_
 check_PROGRAMS += messager/test_send_with_1_failure_moves_message_from_sendq
 messager_test_send_with_1_failure_moves_message_from_sendq_SOURCES = messager/test_send_with_1_failure_moves_message_from_sendq.c
 
+check_PROGRAMS += messager/test_timeout_callback_fires
+messager_test_timeout_callback_fires_SOURCES = messager/test_timeout_callback_fires.c
+
 check_PROGRAMS += util/test_nanoseconds
 util_test_nanoseconds_SOURCES = util/test_nanoseconds.c
 

libcurvecpr/test/messager/.gitignore

@@ -1,3 +1,4 @@
 /test_new_configures_object
 /test_recv_requests_removal_from_sendmarkq
 /test_send_with_1_failure_moves_message_from_sendq
+/test_timeout_callback_fires

libcurvecpr/test/messager/test_new_configures_object.c

@@ -5,10 +5,32 @@
 
 static char static_priv[] = "Hello!";
 
+static unsigned char t_sendmarkq_is_full (struct curvecpr_messager *messager)
+{
+    return 0;
+}
+
+static unsigned char t_sendq_is_empty (struct curvecpr_messager *messager)
+{
+    return 1;
+}
+
+static int t_sendmarkq_head (struct curvecpr_messager *messager, struct curvecpr_block **block_stored)
+{
+    return 1;
+}
+
 START_TEST (test_new_configures_object)
 {
     struct curvecpr_messager messager;
-    struct curvecpr_messager_cf cf = { .priv = static_priv };
+    struct curvecpr_messager_cf cf = {
+        .ops = {
+            .sendmarkq_is_full = t_sendmarkq_is_full,
+            .sendq_is_empty = t_sendq_is_empty,
+            .sendmarkq_head = t_sendmarkq_head
+        },
+        .priv = static_priv
+    };
 
     curvecpr_messager_new(&messager, &cf, 1);
 

libcurvecpr/test/messager/test_recv_requests_removal_from_sendmarkq.c

@@ -5,6 +5,21 @@
 
 #include <curvecpr/bytes.h>
 
+static unsigned char t_sendmarkq_is_full (struct curvecpr_messager *messager)
+{
+    return 0;
+}
+
+static unsigned char t_sendq_is_empty (struct curvecpr_messager *messager)
+{
+    return 1;
+}
+
+static int t_sendmarkq_head (struct curvecpr_messager *messager, struct curvecpr_block **block_stored)
+{
+    return 1;
+}
+
 static int t_sendmarkq_remove_range (struct curvecpr_messager *messager, unsigned long long start, unsigned long long end)
 {
     fail_unless(start == 0);
@@ -18,6 +33,9 @@ START_TEST (test_recv_requests_removal_from_sendmarkq)
     struct curvecpr_messager messager;
     struct curvecpr_messager_cf cf = {
         .ops = {
+            .sendmarkq_is_full = t_sendmarkq_is_full,
+            .sendq_is_empty = t_sendq_is_empty,
+            .sendmarkq_head = t_sendmarkq_head,
             .sendmarkq_remove_range = t_sendmarkq_remove_range
         }
     };

libcurvecpr/test/messager/test_send_with_1_failure_moves_message_from_sendq.c

@@ -67,6 +67,7 @@ START_TEST (test_send_with_1_failure_moves_message_from_sendq)
     struct curvecpr_messager messager;
     struct curvecpr_messager_cf cf = {
         .ops = {
+            .sendq_is_empty = t_q_is_empty,
             .sendmarkq_is_full = t_q_is_full,
             .recvmarkq_is_empty = t_q_is_empty,
             .recvmarkq_get_nth_unacknowledged = t_recvmarkq_get_nth_unacknowledged,

libcurvecpr/test/messager/test_timeout_callback_fires.c

@@ -0,0 +1,85 @@
+#include <check.h>
+#include <check_extras.h>
+
+#include <curvecpr/messager.h>
+
+long long test_timeout = -1LL;
+
+static struct curvecpr_block static_block = {
+    .id = 0,
+    .clock = 0,
+    .eof = CURVECPR_BLOCK_STREAM,
+    .data_len = 7,
+    .data = "Hello!"
+};
+
+static unsigned char t_q_is_full (struct curvecpr_messager *messager)
+{
+    return 0;
+}
+
+static unsigned char t_q_is_empty (struct curvecpr_messager *messager)
+{
+    return 1;
+}
+
+static int t_recvmarkq_get_nth_unacknowledged (struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored)
+{
+    return 1;
+}
+
+static int t_sendmarkq_head (struct curvecpr_messager *messager, struct curvecpr_block **block_stored)
+{
+    return 1;
+}
+
+static int t_sendq_head (struct curvecpr_messager *messager, struct curvecpr_block **block_stored)
+{
+    *block_stored = &static_block;
+    return 0;
+}
+
+static int t_send (struct curvecpr_messager *messager, const unsigned char *buf, size_t num)
+{
+    return 0;
+}
+
+static int t_sendq_move_to_sendmarkq (struct curvecpr_messager *messager, const struct curvecpr_block *block, struct curvecpr_block **block_stored)
+{
+    return 0;
+}
+
+static void t_put_next_timeout (struct curvecpr_messager *messager, long long timeout)
+{
+    test_timeout = timeout;
+}
+
+START_TEST (test_timeout_callback_fires)
+{
+    struct curvecpr_messager messager;
+    struct curvecpr_messager_cf cf = {
+        .ops = {
+            .sendq_is_empty = t_q_is_empty,
+            .sendmarkq_is_full = t_q_is_full,
+            .recvmarkq_is_empty = t_q_is_empty,
+            .recvmarkq_get_nth_unacknowledged = t_recvmarkq_get_nth_unacknowledged,
+            .sendmarkq_head = t_sendmarkq_head,
+            .sendq_head = t_sendq_head,
+            .send = t_send,
+            .sendq_move_to_sendmarkq = t_sendq_move_to_sendmarkq,
+            .put_next_timeout = t_put_next_timeout
+        }
+    };
+
+    curvecpr_messager_new(&messager, &cf, 1);
+
+    fail_unless(test_timeout >= 0);
+    test_timeout = -1LL;
+
+    curvecpr_messager_process_sendq(&messager);
+
+    fail_unless(test_timeout >= 0);
+}
+END_TEST
+
+RUN_TEST (test_timeout_callback_fires)