Latency measurement tool

src/meas_lat.c

@@ -1,23 +1,121 @@
 #include <stdlib.h>
 #include <stdio.h>
+#include <sys/timerfd.h>
+#include <time.h>
+#include <unistd.h>
 #include "evt_core.h"
 #include "net_tools.h"
 
-struct timer_ctx {
+struct measure_conf {
+  int udp_fd;
+  uint64_t max_measure;
+  uint64_t payload_size;
+  char* payload;
   uint64_t counter;
 };
 
+struct packet_header {
+  uint64_t counter;
+  struct timespec emit_time;
+};
+
+struct measure_conf* create_measure_conf(int udp_sock, char* max_mes, char* plsize) {
+  struct measure_conf* mc = malloc(sizeof(struct measure_conf));
+  if (mc == NULL) {
+    perror("Malloc failed");
+    exit(EXIT_FAILURE);
+  }
+
+  if (udp_sock >= 0) {
+    mc->udp_fd = dup(udp_sock);
+    if (mc->udp_fd == -1) {
+      perror("Dup failed");
+      exit(EXIT_FAILURE);
+    }
+  }
+  mc->counter = 0;
+  mc->max_measure = atoi(max_mes);
+  mc->payload_size = atoi(plsize);
+  mc->payload = malloc(mc->payload_size);
+  if (mc->payload == NULL) {
+    perror("malloc failed");
+    exit(EXIT_FAILURE);
+  }
+  return mc;
+}
+
 void on_udp(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  ssize_t res;
+  int secs, nsecs;
+  uint64_t micro_sec;
+  struct timespec curr;
+  struct measure_conf* mc = fdinfo->other;
+  res = read(fdinfo->fd, mc->payload, mc->payload_size);
+  if (res != mc->payload_size) {
+    perror("read error");
+    exit(EXIT_FAILURE);
+  }
+  struct packet_header* head = (struct packet_header*) mc->payload;
+  if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){
+    perror("clock_gettime error");
+    exit(EXIT_FAILURE);
+  }
+
+  secs = curr.tv_sec - head->emit_time.tv_sec;
+  nsecs = curr.tv_nsec - head->emit_time.tv_nsec;
+  micro_sec = secs * 1000000 + nsecs / 1000;
+  printf("Packet %llu latency %luµs\n", (unsigned long long)head->counter, micro_sec);
+
+  if (head->counter >= mc->max_measure) {
+    printf("Measurement done\n");
+    exit(EXIT_SUCCESS);
+  }
 }
 
 void on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  ssize_t s;
+  uint64_t ticks;
+  struct measure_conf* mc = fdinfo->other;
+
+  s = read(fdinfo->fd, &ticks, sizeof(uint64_t));
+  if (s != sizeof(uint64_t)) {
+    perror("Read error");
+    exit(EXIT_FAILURE);
+  }
+
+  if (ticks != 1) {
+    fprintf(stderr, "Has ticked %lu times, expected 1 time. This is a bug\n", ticks);
+  }
+  mc->counter += ticks;
+
+  memset(mc->payload, 0, mc->payload_size);
+  struct packet_header* head = (struct packet_header*)mc->payload;
+  head->counter = mc->counter;
+  if (clock_gettime(CLOCK_MONOTONIC, &head->emit_time) == -1) {
+    perror("clock_gettime error");
+    exit(EXIT_FAILURE);
+  }
+  s = send(mc->udp_fd, mc->payload, mc->payload_size, 0);
+  if (s < 0) {
+    perror("Send error");
+    exit(EXIT_FAILURE);
+  }
+}
+
+void free_timer_conf(void* v) {
+  struct measure_conf* mc = v;
+  close(mc->udp_fd);
+  free(mc->payload);
+  free(mc);
 }
 
 int main(int argc, char** argv) {
   printf("~ measlat ~\n");
-  if (argc < 3) exit(EXIT_FAILURE);
+  if (argc < 6){
+    fprintf(stderr, "%s host port interval count payload-size\n", argv[0]);
+    exit(EXIT_FAILURE);
+  }
 
-  struct timer_ctx apptime = {0};
   struct evt_core_ctx evts = {0};
   struct evt_core_cat udp_read = {
     .app_ctx = NULL,
@@ -29,7 +127,7 @@ int main(int argc, char** argv) {
     .socklist = NULL
   };
   struct evt_core_cat timer = {
-    .app_ctx = &apptime,
+    .app_ctx = NULL,
     .free_app_ctx = NULL,
     .cb = on_timer,
     .err_cb = NULL,
@@ -52,8 +150,42 @@ int main(int argc, char** argv) {
 
   fdinfo.fd = udp_sock;
   fdinfo.cat->name = "udp-read";
+  fdinfo.other = create_measure_conf (-1, argv[4], argv[5]);
   sprintf(fdinfo.url, "udp:read:%s:%s", argv[1], argv[2]);
   evt_core_add_fd (&evts, &fdinfo);
+  printf("--- UDP socket registered\n");
+
+  struct timespec now;
+  struct itimerspec timer_config;
+
+  if (clock_gettime(CLOCK_REALTIME, &now) == -1) {
+    perror("clock_gettime");
+    return 1;
+  }
+
+  uint64_t micro_sec = atoi(argv[3]);
+  timer_config.it_value.tv_sec = now.tv_sec + 1;
+  timer_config.it_value.tv_nsec = now.tv_nsec;
+  timer_config.it_interval.tv_sec = micro_sec / 1000;
+  timer_config.it_interval.tv_nsec = micro_sec % 1000 * 1000000;
+
+  fdinfo.fd = timerfd_create(CLOCK_REALTIME, 0);
+  if (fdinfo.fd == -1) {
+    perror("Unable to timerfd_create");
+    return 1;
+  }
+  if (timerfd_settime (fdinfo.fd, TFD_TIMER_ABSTIME, &timer_config, NULL) == -1) {
+    perror("Unable to timerfd_time");
+    return 1;
+  }
+  fdinfo.cat->name = "timer";
+  fdinfo.other = create_measure_conf(udp_sock, argv[4], argv[5]);
+  fdinfo.free_other = free_timer_conf;
+  sprintf(fdinfo.url, "timer:%s:%s", argv[3], argv[4]);
+  evt_core_add_fd (&evts, &fdinfo);
+  printf("--- Timer registered\n");
+
+  evt_core_loop(&evts);
 
   return 0;
 }

udp-echo.py

@@ -25,10 +25,10 @@ def receive_next(sock):
 def receive_and_send_one(sock):
     "Waits for a single datagram over the socket and echoes it back."
     input_data, addr = receive_next(sock)
-    message = input_data.decode()
+    #message = input_data.decode()
-    logger.info("Received message from %s: %s (%s bytes).", addr, message, len(input_data))
+    #logger.info("Received message from %s: %s (%s bytes).", addr, message, len(input_data))
     output_len = sock.sendto(input_data, addr)
-    logger.info("Echoed message back to %s: %s (%s bytes).", addr, message, output_len)
+    #logger.info("Echoed message back to %s: %s (%s bytes).", addr, message, output_len)
 
 def start(args):
     "Runs the server."