Use asynchronous command queue

When using libusb, process commands asynchronously to reduce the effect
of serial traffic on audio.

Author: v3rm0n

Makefile

@@ -1,8 +1,8 @@
 #Set all your object files (the object files of all the .c files in your project, e.g. main.o my_sub_functions.o )
-OBJ = src/main.o src/serial.o src/slip.o src/command.o src/render.o src/ini.o src/config.o src/input.o src/gamecontrollers.o src/fx_cube.o src/usb.o src/audio.o src/usb_audio.o src/ringbuffer.o src/inprint2.o
+OBJ = src/main.o src/serial.o src/slip.o src/command.o src/render.o src/ini.o src/config.o src/input.o src/gamecontrollers.o src/fx_cube.o src/usb.o src/audio.o src/usb_audio.o src/ringbuffer.o src/inprint2.o src/command_queue.o
 
 #Set any dependant header files so that if they are edited they cause a complete re-compile (e.g. main.h some_subfunctions.h some_definitions_file.h ), or leave blank
-DEPS = src/serial.h src/slip.h src/command.h src/render.h src/ini.h src/config.h src/input.h src/gamecontrollers.h src/fx_cube.h src/audio.h src/ringbuffer.h src/inline_font.h
+DEPS = src/serial.h src/slip.h src/command.h src/render.h src/ini.h src/config.h src/input.h src/gamecontrollers.h src/fx_cube.h src/audio.h src/ringbuffer.h src/inline_font.h src/command_queue.h
 
 #Any special libraries you are using in your project (e.g. -lbcm2835 -lrt `pkg-config --libs gtk+-3.0` ), or leave blank
 INCLUDES = $(shell pkg-config --libs sdl2 libserialport | sed 's/-mwindows//')

src/command_queue.c

@@ -0,0 +1,39 @@
+#include <SDL.h>
+
+const int command_size = 4096;
+int write_cursor = 0;
+int read_cursor = 0;
+uint8_t **command_buffer;
+uint32_t *command_sizes;
+
+void command_queue_init() {
+  command_buffer = SDL_malloc(command_size * sizeof(uint8_t *));
+  command_sizes = SDL_malloc(command_size * sizeof(uint32_t));
+}
+
+void command_queue_destroy() {
+  SDL_free(command_buffer);
+  SDL_free(command_sizes);
+}
+
+int command_queue_push(uint8_t *data, uint32_t size) {
+  uint8_t *recv_buf = SDL_malloc(size * sizeof(uint8_t));
+  SDL_memcpy(recv_buf, data, size);
+
+  SDL_free(command_buffer[write_cursor % command_size]);
+  command_buffer[write_cursor % command_size] = recv_buf;
+  command_sizes[write_cursor % command_size] = size;
+  write_cursor++;
+  return 1;
+}
+
+int command_queue_pop(uint8_t **data, uint32_t *size) {
+  int compare = write_cursor - read_cursor;
+  if (compare == 0) {
+    return 0;
+  }
+  *data = command_buffer[read_cursor % command_size];
+  *size = command_sizes[read_cursor % command_size];
+  read_cursor++;
+  return compare;
+}

src/command_queue.h

@@ -0,0 +1,9 @@
+#ifndef COMMAND_QUEUE_H_
+#define COMMAND_QUEUE_H_
+#include <stdint.h>
+
+void command_queue_init();
+int command_queue_push(uint8_t *data, uint32_t size);
+int command_queue_pop(uint8_t **data, uint32_t *size);
+void command_queue_destroy();
+#endif

src/main.c

@@ -11,12 +11,14 @@
 #include "SDL2_inprint.h"
 #include "audio.h"
 #include "command.h"
+#include "command_queue.h"
 #include "config.h"
-#include "input.h"
 #include "gamecontrollers.h"
+#include "input.h"
 #include "render.h"
 #include "serial.h"
 #include "slip.h"
+#include "usb.h"
 
 enum state { QUIT, WAIT_FOR_DEVICE, RUN };
 
@@ -40,6 +42,8 @@ int main(const int argc, char *argv[]) {
     SDL_Log("Using preferred device %s.\n", preferred_device);
   }
 
+  command_queue_init();
+
   // Initialize the config to defaults read in the params from the
   // configfile if present
   config_params_s conf = init_config();
@@ -58,15 +62,14 @@ int main(const int argc, char *argv[]) {
   static const slip_descriptor_s slip_descriptor = {
       .buf = slip_buffer,
       .buf_size = sizeof(slip_buffer),
-      .recv_message = process_command, // the function where complete slip
-                                       // packets are processed further
+      .recv_message = command_queue_push, // the function where complete slip
+                                          // packets are processed further
   };
 
   static slip_handler_s slip;
 
   uint8_t prev_input = 0;
   uint8_t prev_note = 0;
-  uint16_t zerobyte_packets = 0; // used to detect device disconnection
 
   signal(SIGINT, intHandler);
   signal(SIGTERM, intHandler);
@@ -108,6 +111,9 @@ int main(const int argc, char *argv[]) {
         reset_display();
       }
       run = RUN;
+#ifdef USE_LIBUSB
+      async_read(serial_buf, serial_read_size, &slip);
+#endif
     } else {
       SDL_LogCritical(SDL_LOG_CATEGORY_ERROR, "Device not detected on begin loop.");
       if (conf.wait_for_device == 1) {
@@ -158,6 +164,9 @@ int main(const int argc, char *argv[]) {
               run = RUN;
               port_inited = 1;
               screensaver_destroy();
+#ifdef USE_LIBUSB
+              async_read(serial_buf, serial_read_size, &slip);
+#endif
             } else {
               SDL_LogCritical(SDL_LOG_CATEGORY_ERROR, "Device not detected.");
               run = QUIT;
@@ -227,6 +236,8 @@ int main(const int argc, char *argv[]) {
         }
       }
 
+#ifndef USE_LIBUSB
+      uint16_t zerobyte_packets = 0; // used to detect device disconnection
       while (1) {
         // read serial port
         const int bytes_read = serial_read(serial_buf, serial_read_size);
@@ -276,7 +287,22 @@ int main(const int argc, char *argv[]) {
           break;
         }
       }
-      render_screen();
+#endif
+
+      uint8_t *command;
+      uint32_t command_size;
+      int draws = 0;
+      while (command_queue_pop(&command, &command_size) > 0) {
+        if (command_size == 0) {
+          run = QUIT;
+          break;
+        }
+        process_command(command, command_size);
+        draws++;
+      }
+      if (draws > 0) {
+        render_screen();
+      }
       SDL_Delay(conf.idle_ms);
     }
   } while (run > QUIT);
@@ -292,5 +318,6 @@ int main(const int argc, char *argv[]) {
   close_serial_port();
   SDL_free(serial_buf);
   SDL_Quit();
+  command_queue_destroy();
   return 0;
 }

src/serial.c

@@ -89,7 +89,7 @@ int check_serial_port() {
   return device_found;
 }
 
-int init_serial(const int verbose, const char *preferred_device) {
+int init_serial(const int verbose, char *preferred_device) {
   if (m8_port != NULL) {
     // Port is already initialized
     return 1;

src/serial.h

@@ -14,7 +14,7 @@ int init_serial_with_file_descriptor(int file_descriptor);
 #define serial_read_size 1024
 #endif
 
-int init_serial(int verbose, const char *preferred_device);
+int init_serial(int verbose, char *preferred_device);
 int list_devices();
 int check_serial_port();
 int reset_display();

src/usb.c

@@ -10,6 +10,8 @@
 #include <stdlib.h>
 #include <string.h>
 
+#include "command_queue.h"
+#include "serial.h"
 #include "usb.h"
 
 static int ep_out_addr = 0x03;
@@ -111,6 +113,61 @@ int bulk_transfer(int endpoint, uint8_t *serial_buf, int count, unsigned int tim
   return actual_length;
 }
 
+int bulk_async_transfer(int endpoint, uint8_t *serial_buf, int count, unsigned int timeout_ms,
+                        void (*f)(struct libusb_transfer *), void *user_data) {
+  struct libusb_transfer *transfer;
+  transfer = libusb_alloc_transfer(1);
+  libusb_fill_bulk_stream_transfer(transfer, devh, endpoint, 0, serial_buf, count, f, user_data,
+                                   timeout_ms);
+  int r = libusb_submit_transfer(transfer);
+
+  if (r < 0) {
+    SDL_Log("Error");
+    libusb_free_transfer(transfer);
+    return r;
+  }
+  return 0;
+}
+
+void async_callback(struct libusb_transfer *xfr) {
+  if (xfr->status != LIBUSB_TRANSFER_COMPLETED) {
+    if (libusb_submit_transfer(xfr) < 0) {
+      SDL_Log("error re-submitting URB\n");
+    }
+    return;
+  }
+
+  int bytes_read = xfr->actual_length;
+  if (bytes_read < 0) {
+    SDL_LogCritical(SDL_LOG_CATEGORY_ERROR, "Error %d reading serial. \n", (int)bytes_read);
+    command_queue_push(NULL, 0);
+  } else if (bytes_read > 0) {
+    uint8_t *serial_buf = xfr->buffer;
+    uint8_t *cur = serial_buf;
+    const uint8_t *end = serial_buf + bytes_read;
+    slip_handler_s *slip = (slip_handler_s *)xfr->user_data;
+    while (cur < end) {
+      // process the incoming bytes into commands and draw them
+      int n = slip_read_byte(slip, *(cur++));
+      if (n != SLIP_NO_ERROR) {
+        if (n == SLIP_ERROR_INVALID_PACKET) {
+          SDL_LogError(SDL_LOG_CATEGORY_ERROR, "Invalid SLIP packet!\n");
+
+        } else {
+          SDL_LogError(SDL_LOG_CATEGORY_ERROR, "SLIP error %d\n", n);
+        }
+      }
+    }
+  }
+  if (libusb_submit_transfer(xfr) < 0) {
+    SDL_Log("error re-submitting URB\n");
+  }
+}
+
+int async_read(uint8_t *serial_buf, int count, slip_handler_s *slip) {
+  return bulk_async_transfer(ep_in_addr, serial_buf, count, 300, &async_callback, slip);
+}
+
 int blocking_write(void *buf, int count, unsigned int timeout_ms) {
   return bulk_transfer(ep_out_addr, buf, count, timeout_ms);
 }

src/usb.h

@@ -2,8 +2,11 @@
 #define M8C_USB_H_
 #ifdef USE_LIBUSB
 
+#include "slip.h"
 #include <libusb.h>
+
 extern libusb_device_handle *devh;
+int async_read(uint8_t *serial_buf, int count, slip_handler_s *slip);
 
 #endif // USE_LIBUSB
 #endif // M8C_USB_H_

src/usb_audio.c

@@ -25,7 +25,7 @@ static void audio_callback(void *userdata, Uint8 *stream, int len) {
 
   // If we didn't read the full len bytes, fill the rest with zeros
   if (read_len < len) {
-    SDL_memset(&stream[read_len], 0, len - read_len);
+    SDL_MixAudio(stream, &stream[read_len], len - read_len, SDL_MIX_MAXVOLUME);
   }
 }
 
@@ -45,6 +45,13 @@ static void cb_xfr(struct libusb_transfer *xfr) {
     const uint8_t *data = libusb_get_iso_packet_buffer_simple(xfr, i);
     if (sdl_audio_device_id != 0) {
       uint32_t actual = ring_buffer_push(audio_buffer, data, pack->actual_length);
+
+      if (audio_buffer->size < audio_buffer->max_size / 4) {
+        SDL_PauseAudio(1);
+      } else if (audio_buffer->size > audio_buffer->max_size / 3) {
+        SDL_PauseAudio(0);
+      }
+
       if (actual == -1) {
         SDL_LogDebug(SDL_LOG_CATEGORY_SYSTEM, "Buffer overflow!");
       }