Clarify functions and add additional utilities.

Some of the callback functions specified by the messager struct were
badly named; their names have been updated to reflect their actual
usage.

Additionally, the GLib implementations provide some utility functions
that make looping really easy. A complete event loop using poll(2) now
looks something like this:

    while (!curvecpr_client_messager_glib_is_finished(cmg)) {
        long long timeout =
            curvecpr_client_messager_glib_next_timeout(cmg) / 1000000;

        struct pollfd polls[1] = {
            { .fd = my_socket, .events = POLLIN }
        };
        if (poll(polls, 1, timeout) < 0)
            break;

        if (polls[0].revents) {
            unsigned char buf[1152];
            int r = recvfrom(my_socket, buf, 1152, 0, NULL, NULL);
            if (r <= 0)
                break;

            curvecpr_client_messager_glib_recv(cmg, buf, r);
        }

        curvecpr_client_messager_glib_process_sendq(cmg);
    }

Author: impl

libcurvecpr-glib/include/curvecpr_glib/client_messager_glib.h

@@ -54,8 +54,9 @@ void curvecpr_client_messager_glib_new (struct curvecpr_client_messager_glib *cm
 void curvecpr_client_messager_glib_dealloc (struct curvecpr_client_messager_glib *cmg);
 int curvecpr_client_messager_glib_connected (struct curvecpr_client_messager_glib *cmg);
 int curvecpr_client_messager_glib_send (struct curvecpr_client_messager_glib *cmg, const unsigned char *buf, size_t num);
-int curvecpr_client_messager_glib_close (struct curvecpr_client_messager_glib *cmg);
 int curvecpr_client_messager_glib_recv (struct curvecpr_client_messager_glib *cmg, const unsigned char *buf, size_t num);
+unsigned char curvecpr_client_messager_glib_is_finished (struct curvecpr_client_messager_glib *cmg);
+int curvecpr_client_messager_glib_finish (struct curvecpr_client_messager_glib *cmg);
 int curvecpr_client_messager_glib_process_sendq (struct curvecpr_client_messager_glib *cmg);
 long long curvecpr_client_messager_glib_next_timeout (struct curvecpr_client_messager_glib *cmg);
 

libcurvecpr-glib/include/curvecpr_glib/messager_glib.h

@@ -49,7 +49,8 @@ struct curvecpr_messager_glib {
 
 void curvecpr_messager_glib_new (struct curvecpr_messager_glib *mg, struct curvecpr_messager_glib_cf *cf, unsigned char client);
 void curvecpr_messager_glib_dealloc (struct curvecpr_messager_glib *mg);
-int curvecpr_messager_glib_close (struct curvecpr_messager_glib *mg);
+unsigned char curvecpr_messager_glib_is_finished (struct curvecpr_messager_glib *mg);
+int curvecpr_messager_glib_finish (struct curvecpr_messager_glib *mg);
 int curvecpr_messager_glib_send (struct curvecpr_messager_glib *mg, const unsigned char *buf, size_t num);
 int curvecpr_messager_glib_recv (struct curvecpr_messager_glib *mg, const unsigned char *buf, size_t num);
 int curvecpr_messager_glib_process_sendq (struct curvecpr_messager_glib *mg);

libcurvecpr-glib/lib/client_messager_glib.c

@@ -111,9 +111,14 @@ int curvecpr_client_messager_glib_send (struct curvecpr_client_messager_glib *cm
     return curvecpr_messager_glib_send(&cmg->mg, buf, num);
 }
 
-int curvecpr_client_messager_glib_close (struct curvecpr_client_messager_glib *cmg)
+unsigned char curvecpr_client_messager_glib_is_finished (struct curvecpr_client_messager_glib *cmg)
 {
-    return curvecpr_messager_glib_close(&cmg->mg);
+    return curvecpr_messager_glib_is_finished(&cmg->mg);
+}
+
+int curvecpr_client_messager_glib_finish (struct curvecpr_client_messager_glib *cmg)
+{
+    return curvecpr_messager_glib_finish(&cmg->mg);
 }
 
 int curvecpr_client_messager_glib_recv (struct curvecpr_client_messager_glib *cmg, const unsigned char *buf, size_t num)
@@ -123,10 +128,10 @@ int curvecpr_client_messager_glib_recv (struct curvecpr_client_messager_glib *cm
 
 int curvecpr_client_messager_glib_process_sendq (struct curvecpr_client_messager_glib *cmg)
 {
-    return curvecpr_messager_glib_process_sendq(&cmg->mg);
+    return cmg->client.negotiated == CURVECPR_CLIENT_PENDING ? 0 : curvecpr_messager_glib_process_sendq(&cmg->mg);
 }
 
 long long curvecpr_client_messager_glib_next_timeout (struct curvecpr_client_messager_glib *cmg)
 {
-    return curvecpr_messager_glib_next_timeout(&cmg->mg);
+    return cmg->client.negotiated == CURVECPR_CLIENT_PENDING ? -1 : curvecpr_messager_glib_next_timeout(&cmg->mg);
 }

libcurvecpr-glib/lib/messager_glib.c

@@ -84,6 +84,19 @@ static int _sendq_head (struct curvecpr_messager *messager, struct curvecpr_bloc
     return -1;
 }
 
+static unsigned char _sendq_is_empty (struct curvecpr_messager *messager)
+{
+    struct curvecpr_messager_glib *mg = messager->cf.priv;
+
+    return !mg->sendq_head_exists && /* We don't have a block actually waiting to be
+                                        written. */
+        mg->pending_used == 0 &&     /* We don't have any bytes that we could turn into a
+                                        block to be written. */
+        (!mg->pending_eof ||         /* The EOF flag isn't set. */
+            mg->messager.my_eof);    /* Even if our EOF flag is set, the messager must
+                                        not have sent the EOF message. */
+}
+
 static int _sendq_move_to_sendmarkq (struct curvecpr_messager *messager, const struct curvecpr_block *block, struct curvecpr_block **block_stored)
 {
     struct curvecpr_messager_glib *mg = messager->cf.priv;
@@ -251,7 +264,7 @@ static int _recvmarkq_put (struct curvecpr_messager *messager, const struct curv
     return 0;
 }
 
-static int _recvmarkq_get (struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored)
+static int _recvmarkq_get_nth_unacknowledged (struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored)
 {
     struct curvecpr_messager_glib *mg = messager->cf.priv;
 
@@ -280,7 +293,7 @@ static int _recvmarkq_get (struct curvecpr_messager *messager, unsigned int n, s
     return -1;
 }
 
-static int _recvmarkq_move_range_to_recvq (struct curvecpr_messager *messager, unsigned long long start, unsigned long long end)
+static int _recvmarkq_remove_range (struct curvecpr_messager *messager, unsigned long long start, unsigned long long end)
 {
     struct curvecpr_messager_glib *mg = messager->cf.priv;
 
@@ -317,17 +330,18 @@ void curvecpr_messager_glib_new (struct curvecpr_messager_glib *mg, struct curve
         .ops = {
             .sendq_head = _sendq_head,
             .sendq_move_to_sendmarkq = _sendq_move_to_sendmarkq,
-     
+            .sendq_is_empty = _sendq_is_empty,
+
             .sendmarkq_head = _sendmarkq_head,
             .sendmarkq_get = _sendmarkq_get,
             .sendmarkq_remove_range = _sendmarkq_remove_range,
             .sendmarkq_is_full = _sendmarkq_is_full,
-     
+
             .recvmarkq_put = _recvmarkq_put,
-            .recvmarkq_get = _recvmarkq_get,
+            .recvmarkq_get_nth_unacknowledged = _recvmarkq_get_nth_unacknowledged,
             .recvmarkq_is_full = _recvmarkq_is_full,
-            .recvmarkq_move_range_to_recvq = _recvmarkq_move_range_to_recvq,
-     
+            .recvmarkq_remove_range = _recvmarkq_remove_range,
+
             .send = _send
         },
         .priv = mg
@@ -362,7 +376,12 @@ void curvecpr_messager_glib_dealloc (struct curvecpr_messager_glib *mg)
     g_sequence_free(mg->recvmarkq);
 }
 
-int curvecpr_messager_glib_close (struct curvecpr_messager_glib *mg)
+unsigned char curvecpr_messager_glib_is_finished (struct curvecpr_messager_glib *mg)
+{
+    return mg->messager.my_final && mg->messager.their_final;
+}
+
+int curvecpr_messager_glib_finish (struct curvecpr_messager_glib *mg)
 {
     if (mg->pending_eof)
         return -EINVAL;

libcurvecpr/include/curvecpr/messager.h

@@ -13,6 +13,7 @@ struct curvecpr_messager;
 struct curvecpr_messager_ops {
     int (*sendq_head)(struct curvecpr_messager *messager, struct curvecpr_block **block_stored);
     int (*sendq_move_to_sendmarkq)(struct curvecpr_messager *messager, const struct curvecpr_block *block, struct curvecpr_block **block_stored);
+    unsigned char (*sendq_is_empty)(struct curvecpr_messager *messager);
 
     /* The sent-to-be-marked queue (sendmarkq) is a priority queue of blocks ordered by
        the time at which they were last sent. */
@@ -22,9 +23,9 @@ struct curvecpr_messager_ops {
     unsigned char (*sendmarkq_is_full)(struct curvecpr_messager *messager);
 
     int (*recvmarkq_put)(struct curvecpr_messager *messager, const struct curvecpr_block *block, struct curvecpr_block **block_stored);
-    int (*recvmarkq_get)(struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored);
+    int (*recvmarkq_get_nth_unacknowledged)(struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored);
     unsigned char (*recvmarkq_is_full)(struct curvecpr_messager *messager);
-    int (*recvmarkq_move_range_to_recvq)(struct curvecpr_messager *messager, unsigned long long start, unsigned long long end);
+    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);
 };

libcurvecpr/lib/messager.c

@@ -265,7 +265,7 @@ static int _send_block (struct curvecpr_messager *messager, struct curvecpr_bloc
         unsigned long long maximum_gap = UINT32_MAX;
 
         for (;;) {
-            if (cf->ops.recvmarkq_get(messager, block_num++, &received_block))
+            if (cf->ops.recvmarkq_get_nth_unacknowledged(messager, block_num++, &received_block))
                 break;
 
             acknowledgment_ranges[i].exists = 1;
@@ -401,7 +401,7 @@ static int _send_block (struct curvecpr_messager *messager, struct curvecpr_bloc
         int i;
 
         for (i = 0; i < 6 && acknowledgment_ranges[i].exists; ++i)
-            cf->ops.recvmarkq_move_range_to_recvq(messager, acknowledgment_ranges[i].start, acknowledgment_ranges[i].end);
+            cf->ops.recvmarkq_remove_range(messager, acknowledgment_ranges[i].start, acknowledgment_ranges[i].end);
 
     }
 
@@ -486,9 +486,7 @@ long long curvecpr_messager_next_timeout (struct curvecpr_messager *messager)
 
     if (!cf->ops.sendmarkq_is_full(messager)) {
         /* If we have pending data, we might write it. */
-        if (cf->ops.sendq_head(messager, &block)) {
-            /* Nothing to write. */
-        } else {
+        if (!cf->ops.sendq_is_empty(messager)) {
             /* Write at the write rate. */
             if (at > messager->my_sent_clock + chicago->wr_rate)
                 at = messager->my_sent_clock + chicago->wr_rate;

libcurvecpr/test/messager/test_send_with_1_failure_moves_message_from_sendq.c

@@ -16,7 +16,7 @@ static unsigned char t_q_is_full (struct curvecpr_messager *messager)
     return 0;
 }
 
-static int t_recvmarkq_get (struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored)
+static int t_recvmarkq_get_nth_unacknowledged (struct curvecpr_messager *messager, unsigned int n, struct curvecpr_block **block_stored)
 {
     return 1;
 }
@@ -64,7 +64,7 @@ START_TEST (test_send_with_1_failure_moves_message_from_sendq)
         .ops = {
             .sendmarkq_is_full = t_q_is_full,
             .recvmarkq_is_full = t_q_is_full,
-            .recvmarkq_get = t_recvmarkq_get,
+            .recvmarkq_get_nth_unacknowledged = t_recvmarkq_get_nth_unacknowledged,
             .sendmarkq_head = t_sendmarkq_head,
             .sendq_head = t_sendq_head,
             .send = t_send,