libfxlink: fix race condition leading to lost messages

Basically if the calculator sends two messages in a row, it is possible
for a single libusb_handle_events() to get both. And the comm structure
wasn't designed for that, because it could buffer only one message at a
time, which the user needed to read after event handling.

The comm structure now has a 16-message buffer, which should be more
than enough for any single event handling loop. On the user level this
has implications in that fxlink_device_finish_bulk_IN() must be called
*in a loop* after each event handling cycle.

Reported in Lephenixnoir/gint#27

fxlink/tui/tui-interactive.c

@@ -549,8 +549,8 @@ int main_tui_interactive(libusb_context *ctx)
         /* Check for devices with finished transfers */
         for(int i = 0; i < TUI.devices.count; i++) {
             struct fxlink_device *fdev = &TUI.devices.devices[i];
-            struct fxlink_message *msg = fxlink_device_finish_bulk_IN(fdev);
-            if(msg) {
+            struct fxlink_message *msg;
+            while((msg = fxlink_device_finish_bulk_IN(fdev))) {
                 fxlink_interactive_handle_message(msg);
                 fxlink_message_free(msg, true);
                 fxlink_device_start_bulk_IN(fdev);

libfxlink/defs.c

@@ -11,6 +11,7 @@
 #include <time.h>
 #include <errno.h>
 #include <unistd.h>
+#include <ctype.h>
 
 char const *fmt_to_ANSI(int format)
 {
@@ -146,3 +147,26 @@ bool delay_cycle(delay_t *delay)
     if(*delay > 0) (*delay)--;
     return false;
 }
+
+void fxlink_hexdump(char const *data, int size, FILE *fp)
+{
+    for(int read = 0; read < size; read += 16) {
+        for(int i = 0; i < 16; i++) {
+            if(read + i < size) {
+                int byte = data[read + i] & 0xff;
+                fprintf(fp, " %02x", byte);
+            }
+            else {
+                fprintf(fp, "   ");
+            }
+        }
+
+        fprintf(fp, " | ");
+        for(int i = 0; i < 16 && read + i < size; i++) {
+            int byte = data[read + i] & 0xff;
+            fprintf(fp, "%c", isprint(byte) ? byte : '.');
+        }
+
+        fprintf(fp, "\n");
+    }
+}

libfxlink/devices.c

@@ -10,6 +10,8 @@
 #include <string.h>
 #include <assert.h>
 
+static void fxlink_device_queue_finished_IN(struct fxlink_device *fdev);
+
 char const *fxlink_device_id(struct fxlink_device const *fdev)
 {
     static char str[32];
@@ -401,17 +403,18 @@ static void bulk_IN_callback(struct libusb_transfer *transfer)
 
     switch(transfer->status) {
     case LIBUSB_TRANSFER_COMPLETED:
-        /* Start or continue an fxlink transfer. When finished, don't resubmit,
-           instead have the user pick up the message before continuing. */
+        // log_("bulk_IN_callback: got %d bytes (comm->ftransfer_IN=%p)\n",
+        //     data_size, comm->ftransfer_IN);
+        // fxlink_hexdump(data, data_size, stderr);
+        /* Start or continue an fxlink transfer. */
         resubmit = true;
-        if(comm->ftransfer_IN) {
+        if(comm->ftransfer_IN)
             fxlink_transfer_receive(comm->ftransfer_IN, data, data_size);
-            if(fxlink_transfer_complete(comm->ftransfer_IN))
-                resubmit = false;
-        }
-        else {
+        else
             comm->ftransfer_IN = fxlink_transfer_make_IN(data, data_size);
-        }
+
+        if(fxlink_transfer_complete(comm->ftransfer_IN))
+            fxlink_device_queue_finished_IN(fdev);
         break;
 
     /* Error: drop data and don't resubmit */
@@ -483,18 +486,16 @@ bool fxlink_device_start_bulk_IN(struct fxlink_device *fdev)
     return true;
 }
 
-struct fxlink_message *fxlink_device_finish_bulk_IN(struct fxlink_device *fdev)
+static void fxlink_device_queue_finished_IN(struct fxlink_device *fdev)
 {
     struct fxlink_comm *comm = fdev->comm;
 
     if(!comm || !comm->ftransfer_IN)
-        return NULL;
-    if(!fxlink_transfer_complete(comm->ftransfer_IN))
-        return NULL;
+        return;
 
     struct fxlink_message *msg = fxlink_transfer_finish_IN(comm->ftransfer_IN);
     if(!msg)
-        return NULL;
+        return;
 
     int version_major = (msg->version >> 8) & 0xff;
     int version_minor = msg->version & 0xff;
@@ -505,6 +506,28 @@ struct fxlink_message *fxlink_device_finish_bulk_IN(struct fxlink_device *fdev)
         msg->application, msg->type, fxlink_size_string(msg->size));
 
     comm->ftransfer_IN = NULL;
+
+    if(comm->queue_IN.size >= FXLINK_DEVICE_IN_QUEUE_SIZE) {
+        elog("device queue full (how?!), dropping above message");
+        fxlink_message_free(msg, true);
+        return;
+    }
+
+    comm->queue_IN.messages[comm->queue_IN.size++] = msg;
+}
+
+struct fxlink_message *fxlink_device_finish_bulk_IN(struct fxlink_device *fdev)
+{
+    struct fxlink_comm *comm = fdev->comm;
+    if(!comm || comm->queue_IN.size <= 0)
+        return NULL;
+
+    struct fxlink_message *msg = comm->queue_IN.messages[0];
+    for(int i = 1; i < comm->queue_IN.size; i++)
+        comm->queue_IN.messages[i-1] = comm->queue_IN.messages[i];
+    comm->queue_IN.size--;
+    comm->queue_IN.messages[comm->queue_IN.size] = 0;
+
     return msg;
 }
 

libfxlink/include/fxlink/defs.h

@@ -11,6 +11,7 @@
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdarg.h>
+#include <stdio.h>
 #include <poll.h>
 
 static inline int min(int x, int y)
@@ -90,6 +91,9 @@ int fxlink_multipoll(int timeout, struct pollfd *fds1, int count1, ...);
    pointer to a statically-allocated string. */
 char const *fxlink_size_string(int bytes);
 
+/* Hexdump a data buffer to FILE. */
+void fxlink_hexdump(char const *data, int size, FILE *fp);
+
 //---
 // Delay
 //---

libfxlink/include/fxlink/devices.h

@@ -167,6 +167,10 @@ enum {
 /* Return a string representation of the calc determined system. */
 char const *fxlink_device_system_string(struct fxlink_device const *fdev);
 
+/* Size of the queue where received messages are stored between USB handling
+   and user code reading them. It almost never has more then one element. */
+#define FXLINK_DEVICE_IN_QUEUE_SIZE 16
+
 /* Device state tracked for communication targets. */
 struct fxlink_comm {
     /* Whether the fxlink interface could be claimed */
@@ -184,6 +188,11 @@ struct fxlink_comm {
     struct fxlink_transfer *ftransfer_IN;
     /* Cancellation flag */
     bool cancelled_IN;
+    /* Completed input message queue */
+    struct {
+      struct fxlink_message *messages[FXLINK_DEVICE_IN_QUEUE_SIZE];
+      int size;
+   } queue_IN;
 
     /* Current OUT transfer */
     struct libusb_transfer *tr_bulk_OUT;
@@ -211,10 +220,12 @@ bool fxlink_device_claim_fxlink(struct fxlink_device *fdev);
    device structure is always ready to receive data from the calculator. */
 bool fxlink_device_start_bulk_IN(struct fxlink_device *fdev);
 
-/* Finish an IN transfer and obtain the completed message. This function should
-   be checked every frame as it will return a non-NULL pointer as soon as the
-   message is completed and the device will only start a new bulk IN transfer
-   until the message is moved out by this function. */
+/* Get a pointer to a completed IN message. This function should be checked *in
+   a loop* at every frame as completed messages queue up in a small buffer and
+   the device device will only start a new bulk IN transfer until the message
+   is moved out by this function. In practice, the buffer will only ever hold
+   more than one message if two messages complete within a single libusb event
+   handling call. But that happens. */
 struct fxlink_message *fxlink_device_finish_bulk_IN(
     struct fxlink_device *fdev);