Buffering multiple can

examples/acf-can/acf-can-talker.c

@@ -34,14 +34,19 @@
 #include <linux/can/raw.h>
 #include <sys/ioctl.h>
 #include <time.h>
+#include <sys/time.h>
+#include <signal.h>
+#include <stdbool.h>
 
 #include <arpa/inet.h>
 #include <stdlib.h>
 #include <argp.h>
 #include <string.h>
+#include <math.h>
 #include <unistd.h>
 #include <stdio.h>
 #include <time.h>
+#include <locale.h>
 
 #include "common/common.h"
 #include "avtp/Udp.h"
@@ -62,27 +67,56 @@ static int priority = -1;
 static uint8_t seq_num = 0;
 static uint8_t use_tscf;
 static uint8_t use_udp;
-static uint8_t multi_can_frames = 1;
 static char can_ifname[IFNAMSIZ] = "STDIN\0";
-
-static char doc[] = "\nacf-can-talker -- a program designed to send CAN messages to \
-                    a remote CAN bus over Ethernet using Open1722 \
-                    \vEXAMPLES\
-                    \n\n  acf-can-talker eth0 aa:bb:cc:ee:dd:ff\
-                    \n\n    (tunnel transactions from STDIN to a remote CAN bus over Ethernet)\
-                    \n\n  acf-can-talker --count 10 eth0 aa:bb:cc:ee:dd:ff\
-                    \n\n    (as above, but pack 10 CAN frames in one Ethernet frame)\
-                    \n\n  acf-can-talker -u 10.0.0.2:17220 vcan1\
-                    \n\n    (tunnel transactions from can1 interface to a remote CAN bus over IP)\
-                    \n\n  candump can1 | acf-can-talker -u 10.0.0.2:17220\
-                    \n\n    (another method to tunnel transactions from vcan1 to a remote CAN bus)";
+static bool verbose_flag = false;
+static bool timeout_flag = false;            // true when user supplies
+static bool count_flag = false;              //    cmd line arg
+static uint32_t buf_timeout = 0;             // override with --timeout
+static uint32_t buf_can_frames = 1;          // override with --count
+volatile sig_atomic_t send_ethernet = false; // true on timeout interrupt
+struct itimerval timer;                      // used with --timeout 
+
+static char doc[] =
+"\n"
+"acf-can-talker -- a program designed to send CAN messages to a remote CAN bus\n"
+"                  over Ethernet using Open1722. The default behavior is to\n"
+"                  send one CAN frame per Ethernet frame. However, a buffer can\n"
+"                  be used that gives the opportunity to pack more than one.\n"
+"\n"
+"OPTIONS\n"
+"\vBUFFERING\n\n"
+"  acf-can-talker, by default, packs only one CAN frame into an Ethernet frame.\n"
+"  Use the --timeout and/or --count options to change this behavior.\n\n"
+"  --timeout <time>   where <time> is an integer followed by units, \n"
+"                     e.g., '1s', or '400us'. Allowable units: 's', 'ms', 'us'.\n"
+"                     The Ethernet frame will be sent <time> after arrival \n"
+"                     of the first CAN message.\n\n"
+"  --count <count>    where <count> is the max number of CAN frames allowed in \n"
+"                     an Ethernet frame.\n\n"
+"  If both buffering options are presented, then whichever occurs first will\n"
+"  trigger the sending of the Ethernet frame.\n\n"
+"  In all cases, when the Ethernet frame is full it is sent.\n\n"
+"EXAMPLES\n"
+"  acf-can-talker eth0 aa:bb:cc:ee:dd:ff\n\n"
+"    (tunnel transactions from STDIN to a remote CAN bus over Ethernet)\n\n\n"
+"  acf-can-talker --count 10 eth0 aa:bb:cc:ee:dd:ff\n\n"
+"    (as above, but send Ethernet frame as soon as we have 10 CAN frames)\n\n\n"
+"  acf-can-talker --timeout 400ms eth0 aa:bb:cc:ee:dd:ff\n\n"
+"    (as above, but send Ethernet frame when 400 msec have passed since \n"
+"     arrival of first CAN frame) \n\n\n"
+"  acf-can-talker -u 10.0.0.2:17220 vcan1\n\n"
+"    (tunnel transactions from can1 interface to a remote CAN bus over IP)\n\n\n"
+"  candump can1 | acf-can-talker -u 10.0.0.2:17220\n\n"
+"    (another method to tunnel transactions from vcan1 to a remote CAN bus)\n\n\n";
 
 static char args_doc[] = "[ifname] dst-mac-address/dst-nw-address:port [can ifname]";
 
 static struct argp_option options[] = {            
     {"tscf", 't', 0, 0, "Use TSCF"},
     {"udp",  'u', 0, 0, "Use UDP" },
-    {"count", 'c', "COUNT", 0, "Set count of CAN messages per Ethernet frame"},
+    {"verbose",  'v', 0, 0, "Show building of Ethernet Frames" },
+    {"timeout", 501, "TIME", 0, "Set time to wait for CAN messages to arrive"},
+    {"count", 502, "COUNT", 0, "Set count of CAN messages per Ethernet frame"},
     {"can ifname", 0, 0, OPTION_DOC, "CAN interface (set to STDIN by default)"},
     {"ifname", 0, 0, OPTION_DOC, "Network interface (If Ethernet)"},
     {"dst-mac-address", 0, 0, OPTION_DOC, "Stream destination MAC address (If Ethernet)"},
@@ -102,8 +136,30 @@ static error_t parser(int key, char *arg, struct argp_state *state)
     case 'u':
         use_udp = 1;
         break;
-    case 'c':
-        multi_can_frames = atoi(arg);
+    case 'v':
+        verbose_flag = true;
+        break;
+    case 501:
+        char units[3];
+        sscanf(arg, "%d%2s", &buf_timeout, units);
+        if (strcmp(units, "s") == 0) {
+            timer.it_value.tv_sec = buf_timeout;
+            timer.it_value.tv_usec = 0;
+        } else if (strcmp(units, "ms") == 0) {
+            timer.it_value.tv_sec = floor(buf_timeout/1e3);
+            timer.it_value.tv_usec = (buf_timeout % 1000) * 1000;
+        } else if (strcmp(units, "us") == 0) {
+            timer.it_value.tv_sec = floor(buf_timeout/1e6);
+            timer.it_value.tv_usec = (buf_timeout % (uint32_t)1e6);
+        } else { 
+           fprintf(stderr, "error with timeout arg format; got: [%s]\n", arg);
+           argp_usage(state);
+        }
+        timeout_flag = true;
+        break;
+    case 502:
+        buf_can_frames = atoi(arg);
+        count_flag = true;
         break;
 
     case ARGP_KEY_NO_ARGS:
@@ -217,44 +273,75 @@ static int prepare_acf_packet(uint8_t* acf_pdu,
 
 static int get_payload(int can_socket, uint8_t* payload, uint32_t *frame_id, uint8_t *length) {
 
-    char stdin_str[1000];
-    char can_str[10];
-    char can_payload[1000];
-    char *token;
-    size_t n;
-    int res;
+  char stdin_str[1000];
+  char can_str[10];
+  char can_payload[1000];
+  char *token;
+  size_t n;
+  int res;
 	struct can_frame frame;
 
-    if (can_socket == 0) {
-        n = read(STDIN_FILENO, stdin_str, 1000);
-        if (n < 0) {
-            return -1;
-        }
+  if (can_socket == 0) {
+      n = read(STDIN_FILENO, stdin_str, 1000);    // EINTR detects interrupt
+      if (n == -1 && errno == EINTR) {            // (our timeout)
+          return -1;
+      } else if (n < 0) {                         // other failure
+        return -2;
+      }
+
+      res = sscanf(stdin_str, "%s %x [%hhu] %[0-9A-F ]s", can_str, frame_id,
+                                                      length, can_payload);
+      if (res < 0) {
+          return -1;
+      }
+
+      token = strtok(can_payload, " ");
+      int index = 0;
+      while (token != NULL) {
+          payload[index++] = (unsigned short)strtol(token, NULL, 16);
+          token = strtok(NULL, " ");
+      }
+  } else {
+      n = read(can_socket, &frame, sizeof(struct can_frame));
+      if (n > 0) {
+          *frame_id = (uint32_t) frame.can_id;
+          *length = (uint8_t) frame.can_dlc;
+          memcpy(payload, frame.data, (size_t) *length);
+      } else if (n == -1 && errno == EINTR) {   // detect our timeout
+        return -1;
+      } else {
+        return -1;
+      }
+  }
+
+  return n;
+}
 
-        res = sscanf(stdin_str, "%s %x [%hhu] %[0-9A-F ]s", can_str, frame_id,
-                                                        length, can_payload);
-        if (res < 0) {
-            return -1;
-        }
+// On timeout signal set a flag for our read loop to detect
+void handle_timeout(int signum) {
+  send_ethernet = true;
+}
 
-        token = strtok(can_payload, " ");
-        int index = 0;
-        while (token != NULL) {
-            payload[index++] = (unsigned short)strtol(token, NULL, 16);
-            token = strtok(NULL, " ");
-        }
-    } else {
-        n = read(can_socket, &frame, sizeof(struct can_frame));
-        if (n > 0) {
-            *frame_id = (uint32_t) frame.can_id;
-            *length = (uint8_t) frame.can_dlc;
-            memcpy(payload, frame.data, (size_t) *length);
-        }
-    }
-
-    return n;
+// Set a signal handler. This is a little more complicated than "normal" in
+// that we need to prevent an interrupted system call (like 'read') from being
+// restarted. This function comes from 'Advanced Programming in the UNIX
+// Environment', figure 10.19.
+typedef void Sigfunc(int);
+Sigfunc * signal_intr(int signo, Sigfunc *func) {
+  struct sigaction act, oact;
+  act.sa_handler = func;
+  sigemptyset(&act.sa_mask);
+  act.sa_flags = 0;
+#ifdef  SA_INTERRUPT
+  act.sa_flags |= SA_INTERRUPT;
+#endif
+  if (sigaction(signo, &act, &oact) < 0) 
+    return(SIG_ERR);
+  return(oact.sa_handler);
 }
 
+
+
 int main(int argc, char *argv[])
 {
 
@@ -266,12 +353,19 @@ int main(int argc, char *argv[])
     uint8_t payload[CAN_PAYLOAD_MAX_SIZE];
     uint8_t payload_length = 0;
     uint32_t frame_id = 0;
-    uint8_t num_acf_msgs = 1;
+    uint32_t num_acf_msgs = 1;
     uint32_t pdu_length;
 
     int can_socket = 0;
-	struct sockaddr_can can_addr;
-	struct ifreq ifr;
+	  struct sockaddr_can can_addr;
+	  struct ifreq ifr;
+
+    struct timespec start_time, end_time;    // for verbose 
+    setlocale(LC_NUMERIC, "");               // allow comma separator in printf
+
+    // used to timeout the packing of Ethernet frame with CAN frames
+    if (signal_intr(SIGALRM, handle_timeout) == SIG_ERR)
+      perror("main(), setting handler: \n");
 
     argp_parse(&argp, argc, argv, 0, NULL, NULL);
 
@@ -283,7 +377,11 @@ int main(int argc, char *argv[])
     if (fd < 0)
         return 1;
 
-    num_acf_msgs = multi_can_frames;
+    // enforce timeout and count constraints, as requested
+    if      (!timeout_flag && !count_flag) { num_acf_msgs = 1; }
+    else if (!timeout_flag &&  count_flag) { num_acf_msgs = buf_can_frames; } 
+    else if ( timeout_flag && !count_flag) { num_acf_msgs = 2^32 - 1; }
+    else                                   { num_acf_msgs = buf_can_frames; } 
 
     // Open a CAN socket for reading frames if required
     if (strcmp(can_ifname, "STDIN\0")) {
@@ -334,15 +432,69 @@ int main(int argc, char *argv[])
             goto err;
         pdu_length += res;
 
-        int i = 0;
-        while (i < num_acf_msgs) {
-            // Get payload -- will 'spin' here until we get the requested number
-            //                of CAN frames.
+
+        bool first_msg_is_next = true;
+        int i = 1;
+        // Get payload -- will loop here until we get the requested number
+        //                of CAN frames, or until timeout.
+        // magic numbers for detecting no more room in the Ethernet frame:
+        // * 24 is max for CLASSIC CAN (CAN MESSAGE INFO + CAN BASE MESSAGE = 16 + 8)
+        // * 80 is max for CAN FD (CAN MESSAGE INFO + CAN BASE MESSAGE = 16 + 64)
+        // and then + 4 when using UDP, so 28 or 84 are the valid magic numbers.
+        while ((i <= num_acf_msgs) && (pdu_length < MAX_PDU_SIZE - 28 )) {
+
+            // last read was interrupted by timeout?
+            if (send_ethernet && i > 1) {
+              if (verbose_flag) {
+                  int width = 55-i;
+                  fprintf(stderr, "%*s", width, "--timeout");
+              }
+              send_ethernet = false;
+              break;
+            }
+
             res = get_payload(can_socket, payload, &frame_id, &payload_length);
-            if (!res) {
-                continue;
+            if (res < 0) { 
+
+              //if (verbose_flag) {
+              //    // Calculate the elapsed time in seconds and nanoseconds
+              //    clock_gettime(CLOCK_REALTIME, &end_time);
+              //    long elapsed_sec = end_time.tv_sec - start_time.tv_sec;
+              //    long elapsed_nsec = end_time.tv_nsec - start_time.tv_nsec;
+              //    if (elapsed_nsec < 0) {
+              //        elapsed_sec--;
+              //        elapsed_nsec += 1000000000;
+              //    }
+              //    // Print the elapsed time
+              //    char strbuff[50];
+              //    sprintf(strbuff, " %ld s %'11ld nsec", elapsed_sec, elapsed_nsec);
+              //    fprintf(stderr, "%*s", 55-i, strbuff);
+              //    //fprintf(stderr, "%*s", 55-i, "read() < 0");
+              //}
+              continue;
+
+            } else if (res == 0) {
+              if (verbose_flag)
+                  fprintf(stderr, "*");
+
+            } else {
+              if (verbose_flag) {
+                  fprintf(stderr, "%d", i%10);
+              }
+              // on reception of 1st msg we need to set our timer if we allow more
+              // than one message per ethernet frame
+              if (first_msg_is_next && num_acf_msgs > 1) {
+                  //clock_gettime(CLOCK_REALTIME, &start_time);
+                  res = setitimer(ITIMER_REAL, &timer, NULL);
+                  //printf("timer is set: sec [%ld] usec [%ld]\n", 
+                  //       timer.it_value.tv_sec, timer.it_value.tv_usec); fflush(stdout);
+                  first_msg_is_next = false;
+              }
+
             }
 
+
+
             uint8_t* acf_pdu = cf_pdu + pdu_length;
             res = prepare_acf_packet(acf_pdu, payload, payload_length, frame_id);
             if (res < 0)
@@ -351,7 +503,18 @@ int main(int argc, char *argv[])
 
             i++;
         }
-
+        alarm(0);      // disarm timer 
+
+        //
+        // done reading CAN for this Ethernet frame
+        //
+
+        if (i > num_acf_msgs && verbose_flag) {
+              int width = 55-i;
+              fprintf(stderr, "%*s", width, "--count  ");
+        }
+        //printf("should send Ethernet here\n");fflush(stdout);
+
         res = update_pdu_length(cf_pdu, pdu_length);
         if (res < 0)
             goto err;
@@ -372,6 +535,8 @@ int main(int argc, char *argv[])
                 goto err;
             }
         }
+        fprintf(stderr, "   ethersize: [%4d]\n", res);
+        fflush(stderr);
     }
 
 err: