Rewrite measlat

CMakeLists.txt

@@ -37,6 +37,8 @@ list(APPEND CSOURCES
 	src/capture_traffic.c
 	src/cap_utils.h
 	src/cap_utils.c
+	src/measure.h
+	src/measure.c
 )
 
 add_executable(donar ${CSOURCES} src/donar.c)

r/thunder_configure.R

@@ -3,7 +3,7 @@ library(sqldf)
 library(plyr)
 library(cowplot)
 
-thunder_ms <- read.csv("thunder_23.csv")
+thunder_ms <- read.csv("thunder_22.csv")
 thunder_ms <- sqldf("select run,ident,jmax,links,latency, CAST(latency as real) / 1000. as lat_ms from thunder_ms")
 thunder_ms$links <- as.factor(thunder_ms$links)
 thunder_ms$jmax <- as.factor(thunder_ms$jmax)
@@ -61,7 +61,7 @@ thunder_bw$links <- as.factor(thunder_bw$links)
 v3 <- ggplot(data = thunder_bw, aes(x = jmax, y=sent_ratio, fill=links)) +
   geom_boxplot(outlier.size=0.1) +
   #scale_y_log10() +
-  coord_cartesian(ylim = c(1,2)) +
+  coord_cartesian(ylim = c(1,4)) +
   scale_fill_grey() +
   ylab("bandwidth ratio") +
   xlab("max allowed jitter") +

src/meas_lat.c

@@ -8,93 +8,33 @@
 #include "net_tools.h"
 #include "socks5.h"
 #include "utils.h"
+#include "measure.h"
 
 struct measlat_ctx {
-  int count, size, interval, verbose;
+  struct measure_conf mc;
+  int verbose;
   char *host, *port, *transport;
 };
 
-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_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   return 1;
 }
 
 int on_udp(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   ssize_t res;
-  uint64_t micro_sec;
+  struct measlat_ctx* mctx = fdinfo->cat->app_ctx;
-  struct timespec curr;
+  res = read(fdinfo->fd, mctx->mc.payload, mctx->mc.payload_size);
-  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);
-  }
-  time_t now;
-  time(&now);
-  char* ctime_no_newline = strtok(ctime(&now), "\n");
 
-  micro_sec = elapsed_micros (&head->emit_time, &curr);
+  measure_parse (res, &mctx->mc);
-  printf("[%s] Packet %llu latency %luµs\n", ctime_no_newline, (unsigned long long)head->counter, micro_sec);
 
-  if (mc->max_measure > 0 && head->counter >= mc->max_measure) {
-    printf("Measurement done\n");
-    exit(EXIT_SUCCESS);
-  }
   return 0;
 }
 
-void free_timer_conf(void* v) {
-  struct measure_conf* mc = v;
-  free(mc->payload);
-  free(mc);
-}
-
 int on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   ssize_t s;
   uint64_t ticks = 0;
-  struct measure_conf* mc = fdinfo->other;
+  struct measlat_ctx* mctx = fdinfo->cat->app_ctx;
 
   s = read(fdinfo->fd, &ticks, sizeof(uint64_t));
   if (s == -1 && errno == EAGAIN) return 1;
@@ -106,23 +46,15 @@ int on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   if (ticks != 1) {
     fprintf(stderr, "Has ticked %lu times, expected 1 time. This is a bug\n", ticks);
   }
-  mc->counter++;
+  mctx->mc.counter++;
 
-  memset(mc->payload, 0, mc->payload_size);
+  struct measure_packet* head = (struct measure_packet*)mctx->mc.payload;
-  struct packet_header* head = (struct packet_header*)mc->payload;
+  head->counter = mctx->mc.counter;
-  head->counter = mc->counter;
   if (clock_gettime(CLOCK_MONOTONIC, &head->emit_time) == -1) {
     perror("clock_gettime error");
     exit(EXIT_FAILURE);
   }
 
-  char *my_msg = "Tu n'es pas tout a fait la misere,\nCar les levres les plus pauvres te denoncent\nPar un sourire.";
-  size_t msg_len = strlen(my_msg);
-  size_t cursor_msg = 0;
-  for (size_t i = sizeof(struct packet_header); i < mc->payload_size; i++) {
-    mc->payload[i] = my_msg[cursor_msg];
-    cursor_msg = (cursor_msg + 1) % msg_len;
-  }
 
   struct evt_core_fdinfo* tgtinfo = evt_core_get_first_from_cat (ctx, "udp-read");
   if (tgtinfo == NULL) tgtinfo = evt_core_get_first_from_cat (ctx, "tcp-read");
@@ -130,7 +62,7 @@ int on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
     printf("No connection yet\n");
     return 1;
   }
-  s = send(tgtinfo->fd, mc->payload, mc->payload_size, 0);
+  s = send(tgtinfo->fd, mctx->mc.payload, mctx->mc.payload_size, 0);
   if (s < 0) {
     perror("Send error");
     //exit(EXIT_FAILURE);
@@ -139,7 +71,7 @@ int on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 }
 
 
-void register_timer(struct evt_core_ctx* evts, int udp, int interval, int count, int size) {
+void register_timer(struct evt_core_ctx* evts) {
   struct timespec now;
   struct itimerspec timer_config;
   char url[1024];
@@ -148,11 +80,18 @@ void register_timer(struct evt_core_ctx* evts, int udp, int interval, int count,
   fdinfo.cat = &cat;
   fdinfo.url = url;
 
+  struct evt_core_cat* ucat = evt_core_get_from_cat (evts, "tcp-read");
+  if (ucat == NULL) {
+    fprintf(stderr, "Category udp-read not found\n");
+    exit(EXIT_FAILURE);
+  }
+  struct measlat_ctx* mctx = ucat->app_ctx;
+
   if (clock_gettime(CLOCK_REALTIME, &now) == -1) {
     perror("clock_gettime");
     exit(EXIT_FAILURE);
   }
-  uint64_t micro_sec = interval;
+  uint64_t micro_sec = mctx->mc.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;
@@ -168,9 +107,7 @@ void register_timer(struct evt_core_ctx* evts, int udp, int interval, int count,
     exit(EXIT_FAILURE);
   }
   fdinfo.cat->name = "timer";
-  fdinfo.other = create_measure_conf(count, size);
+  sprintf(fdinfo.url, "timer:%ld:%ld", mctx->mc.interval, mctx->mc.max_measure);
-  fdinfo.free_other = free_timer_conf;
-  sprintf(fdinfo.url, "timer:%d:%d", interval, count);
   evt_core_add_fd (evts, &fdinfo);
   printf("--- Timer registered\n");
 }
@@ -192,14 +129,12 @@ int on_socks5_success_measlat(struct evt_core_ctx* ctx, struct evt_core_fdinfo*
 
   fdinfo_n.fd = dup(fdinfo->fd);
   fdinfo_n.cat->name = "tcp-read";
-  fdinfo_n.other = create_measure_conf (mctx->count, mctx->size);
-  fdinfo_n.free_other = free_mesure_conf;
   sprintf(fdinfo_n.url, "tcp:read:%s:%d", s5ctx->addr, s5ctx->port);
 
   evt_core_add_fd (ctx, &fdinfo_n);
   printf("--- Tor socket registered\n");
 
-  register_timer(ctx, fdinfo->fd, mctx->interval, mctx->count, mctx->size);
+  register_timer(ctx);
   return 1;
 }
 
@@ -277,13 +212,13 @@ void spawn_udp_socket(struct evt_core_ctx* evts) {
 
   fdinfo.fd = udp_sock;
   fdinfo.cat->name = "udp-read";
-  fdinfo.other = create_measure_conf (mctx->count, mctx->size);
+  fdinfo.other = &mctx->mc;
-  fdinfo.free_other = free_mesure_conf;
+  fdinfo.free_other = NULL;
   sprintf(fdinfo.url, "udp:read:%s:%s", mctx->host, mctx->port);
   evt_core_add_fd (evts, &fdinfo);
   printf("--- UDP socket registered\n");
 
-  register_timer(evts, fdinfo.fd, mctx->interval, mctx->count, mctx->size);
+  register_timer(evts);
 }
 
 int main(int argc, char** argv) {
@@ -310,13 +245,13 @@ int main(int argc, char** argv) {
         mctx.transport = optarg;
         break;
       case 'c': // count
-        mctx.count = atoi(optarg);
+        mctx.mc.max_measure = atoi(optarg);
         break;
       case 's': // size - payload in bytes
-        mctx.size = atoi(optarg);
+        mctx.mc.payload_size = atoi(optarg);
         break;
       case 'i': // interval - every ms
-        mctx.interval = atoi(optarg);
+        mctx.mc.interval = atoi(optarg);
         break;
       default:
         goto usage;
@@ -324,14 +259,11 @@ int main(int argc, char** argv) {
   }
 
   // 2. Check and fix parameters
-  size_t header_size = sizeof(struct packet_header);
+  measure_prepare (&mctx.mc);
-  if (mctx.interval <= 0) mctx.interval = 1000;
-  if (mctx.count <= 0) mctx.count = 1;
-  if (mctx.size < header_size) mctx.size = header_size;
   if (mctx.transport == NULL) mctx.transport = "udp";
   if (mctx.host == NULL || mctx.port == NULL) goto usage;
-  printf("[measlat_conf] host=%s, port=%s, transport=%s, count=%d, size=%d, interval=%d\n",
+  printf("[measlat_conf] host=%s, port=%s, transport=%s, count=%ld, size=%ld, interval=%ld\n",
-          mctx.host, mctx.port, mctx.transport, mctx.count, mctx.size, mctx.interval);
+          mctx.host, mctx.port, mctx.transport, mctx.mc.max_measure, mctx.mc.payload_size, mctx.mc.interval);
 
   // 3. Bind events
   register_categories(&evts, &mctx);

src/measure.c

@@ -0,0 +1,46 @@
+#include "measure.h"
+
+void measure_parse(int size, struct measure_conf* mc) {
+  struct timespec curr;
+  uint64_t micro_sec;
+  if (size != mc->payload_size) {
+    perror("read error");
+    exit(EXIT_FAILURE);
+  }
+  struct measure_packet* head = (struct measure_packet*) mc->payload;
+  if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){
+    perror("clock_gettime error");
+    exit(EXIT_FAILURE);
+  }
+  time_t now;
+  time(&now);
+  char* ctime_no_newline = strtok(ctime(&now), "\n");
+
+  micro_sec = elapsed_micros (&head->emit_time, &curr);
+  printf("[%s] Packet %llu latency %luµs\n", ctime_no_newline, (unsigned long long)head->counter, micro_sec);
+
+  if (mc->max_measure > 0 && head->counter >= mc->max_measure) {
+    printf("Measurement done\n");
+    exit(EXIT_SUCCESS);
+  }
+}
+
+void measure_prepare(struct measure_conf* mc) {
+  if (mc->interval <= 0) mc->interval = 1000;
+  if (mc->max_measure <= 0) mc->max_measure = 1;
+  if (mc->payload_size < sizeof(struct measure_packet)) mc->payload_size = sizeof(struct measure_packet);
+
+  if ((mc->payload = malloc(sizeof(char) * mc->payload_size)) == NULL) {
+    perror("payload malloc failed");
+    exit(EXIT_FAILURE);
+  }
+  memset(mc->payload, 0, mc->payload_size);
+
+  char *my_msg = "Tu n'es pas tout a fait la misere,\nCar les levres les plus pauvres te denoncent\nPar un sourire.";
+  size_t msg_len = strlen(my_msg);
+  size_t cursor_msg = 0;
+  for (size_t i = sizeof(struct measure_packet); i < mc->payload_size; i++) {
+    mc->payload[i] = my_msg[cursor_msg];
+    cursor_msg = (cursor_msg + 1) % msg_len;
+  }
+}

src/measure.h

@@ -0,0 +1,22 @@
+#include <stdlib.h>
+#include <stdio.h>
+#include <stdint.h>
+#include <time.h>
+#include <string.h>
+#include "utils.h"
+
+struct measure_conf {
+  uint64_t max_measure;
+  uint64_t payload_size;
+  uint64_t interval;
+  char* payload;
+  uint64_t counter;
+};
+
+struct measure_packet {
+  uint64_t counter;
+  struct timespec emit_time;
+};
+
+void measure_parse(int size, struct measure_conf* mc);
+void measure_prepare(struct measure_conf* mc);