tor_multipath_voip/src/meas_lat.c

#include <stdlib.h>
#include <stdio.h>
#include <sys/timerfd.h>
#include <time.h>
#include <unistd.h>
#include <errno.h>
#include "evt_core.h"
#include "net_tools.h"

struct measure_conf {
  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 max_mes, int plsize) {
  struct measure_conf* mc = malloc(sizeof(struct measure_conf));
  if (mc == NULL) {
    perror("Malloc failed");
    exit(EXIT_FAILURE);
  }

  mc->counter = 0;
  mc->max_measure = max_mes;
  mc->payload_size = plsize;
  mc->payload = malloc(mc->payload_size);
  if (mc->payload == NULL) {
    perror("malloc failed");
    exit(EXIT_FAILURE);
  }
  return mc;
}

void free_mesure_conf(void* v) {
  struct measure_conf* mc = (struct measure_conf*)v;
  free(mc->payload);
  free(mc);
}

int 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 == -1 && errno == EAGAIN) return 1;

  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);
  }
  return 0;
}

int 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++;

  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);
  }
  struct evt_core_fdinfo* udpinfo = evt_core_get_first_from_cat (ctx, "udp-read");
  if (udpinfo == NULL) {
    printf("No connection yet\n");
    return 1;
  }
  s = send(udpinfo->fd, mc->payload, mc->payload_size, 0);
  if (s < 0) {
    perror("Send error");
    //exit(EXIT_FAILURE);
  }
  return 1;
}

void free_timer_conf(void* v) {
  struct measure_conf* mc = v;
  free(mc->payload);
  free(mc);
}

void register_categories(struct evt_core_ctx* evts) {
  struct evt_core_cat udp_read = {
    .app_ctx = NULL,
    .free_app_ctx = NULL,
    .cb = on_udp,
    .err_cb = NULL,
    .name = "udp-read",
    .flags = EPOLLIN | EPOLLET,
    .socklist = NULL
  };
  struct evt_core_cat timer = {
    .app_ctx = NULL,
    .free_app_ctx = NULL,
    .cb = on_timer,
    .err_cb = NULL,
    .name = "timer",
    .flags = EPOLLIN | EPOLLET,
    .socklist = NULL
  };

  evt_core_init(evts);
  evt_core_add_cat (evts, &udp_read);
  evt_core_add_cat(evts, &timer);
  printf("--- Categories registered\n");

}

int register_udp_socket(struct evt_core_ctx* evts, char* host, char* port, int count, int size) {
  int udp_sock = create_udp_client (host, port);
  char url[1024];
  struct evt_core_cat cat = {0};
  struct evt_core_fdinfo fdinfo = {0};
  fdinfo.cat = &cat;
  fdinfo.url = url;

  fdinfo.fd = udp_sock;
  fdinfo.cat->name = "udp-read";
  fdinfo.other = create_measure_conf (count, size);
  fdinfo.free_other = free_mesure_conf;
  sprintf(fdinfo.url, "udp:read:%s:%s", host, port);
  evt_core_add_fd (evts, &fdinfo);
  printf("--- UDP socket registered\n");
  return fdinfo.fd;
}

void register_timer(struct evt_core_ctx* evts, int udp, int interval, int count, int size) {
  struct timespec now;
  struct itimerspec timer_config;
  char url[1024];
  struct evt_core_cat cat = {0};
  struct evt_core_fdinfo fdinfo = {0};
  fdinfo.cat = &cat;
  fdinfo.url = url;

  if (clock_gettime(CLOCK_REALTIME, &now) == -1) {
    perror("clock_gettime");
    exit(EXIT_FAILURE);
  }
  uint64_t micro_sec = interval;
  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");
    exit(EXIT_FAILURE);
  }
  if (timerfd_settime (fdinfo.fd, TFD_TIMER_ABSTIME, &timer_config, NULL) == -1) {
    perror("Unable to timerfd_time");
    exit(EXIT_FAILURE);
  }
  fdinfo.cat->name = "timer";
  fdinfo.other = create_measure_conf(count, size);
  fdinfo.free_other = free_timer_conf;
  sprintf(fdinfo.url, "timer:%d:%d", interval, count);
  evt_core_add_fd (evts, &fdinfo);
  printf("--- Timer registered\n");
}

int main(int argc, char** argv) {
  printf("~ measlat ~\n");

  int opt, udp_fd, count = 0, size = 0, interval = 0;
  char *host = NULL, *port = NULL;
  struct evt_core_ctx evts = {0};

  // 1. Parse parameters
  while ((opt = getopt(argc, argv, "h:p:c:s:i:")) != -1) {
      switch(opt) {
      case 'h':
        host = optarg;
        break;
      case 'p':
        port = optarg;
        break;
      case 'c':
        count = atoi(optarg);
        break;
      case 's':
        size = atoi(optarg);
        break;
      case 'i':
        interval = atoi(optarg);
        break;
      default:
        goto usage;
      }
  }

  // 2. Check and fix parameters
  size_t header_size = sizeof(struct packet_header);
  if (interval <= 0) interval = 1000;
  if (count <= 0) count = 1;
  if (size < header_size) size = header_size;
  if (host == NULL || port == NULL) goto usage;

  // 3. Bind events
  register_categories(&evts);
  udp_fd = register_udp_socket(&evts, host, port, count, size);
  register_timer(&evts, udp_fd, interval, count, size);

  // 4. Run main loop
  evt_core_loop(&evts);

  return 0;
usage:
  fprintf(stderr, "Usage: %s -h <host> -p <port> [-c <count>] [-i <ms>] [-s <bytes>]\n", argv[0]);
  exit(EXIT_FAILURE);
}