Add two ways info

r/lightning_begin.R

@@ -17,6 +17,22 @@ ggplot(data=xx2, aes(x=packet_id, y=lat, color=way)) +
   #geom_point() +
   theme_classic()
 
+xx4 <- sqldf("select packet_id,1.0 * latency / 1000.0 as lat,way from xx where flag = 0")
+ggplot(data=xx4, aes(x=packet_id, y=lat, color=way)) +
+  geom_line() +
+  coord_cartesian(ylim=c(0,2000)) +
+  #geom_point() +
+  theme_classic()
+
+
+xx5 <- sqldf("select packet_id,1.0 * latency / 1000.0 as lat,way from xx where flag = 1")
+ggplot(data=xx5, aes(x=packet_id, y=lat, color=way)) +
+  geom_line() +
+  coord_cartesian(ylim=c(0,2000)) +
+  #geom_point() +
+  theme_classic()
+
+
 prepros <- sqldf(
 "select 
   r.packet_id, 
@@ -35,3 +51,10 @@ where
   s.packet_id = r.packet_id and 
   s.way = r.way and 
   r.lat = s.latency")
+
+xx3 <- sqldf("select packet_id,1.0 * latency / 1000.0 as lat,flag,way from xx")
+xx3$flag <- factor(xx3$flag)
+ggplot(data=xx3, aes(x=lat, group=flag, color=flag)) +
+  stat_ecdf(pad = FALSE) +
+  #coord_cartesian(xlim=c(0,2000)) +
+  theme_classic()

src/algo_lightning.c

@@ -30,6 +30,8 @@ struct timing_entry {
 
 struct light_ctx {
   uint8_t prev_links[MAX_LINKS];
+  uint16_t remote_stats[MAX_LINKS];
+  uint16_t local_stats[MAX_LINKS];
   struct timing_entry historic[HISTORIC_SIZE];
   struct timespec last[MAX_LINKS];
   uint64_t pkt_rcv_id;
@@ -86,7 +88,11 @@ void algo_lightning_init(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx, str
     lightc->prev_links[i] = UINT8_MAX;
 
   union abstract_packet m;
-  lightc->monit_pkt_size = sizeof(m.fmt.headers) + sizeof(m.fmt.content.link_monitoring_lightning) + sizeof(uint8_t) * (lightc->sent_past_links - 1);
+  lightc->monit_pkt_size =
+    sizeof(m.fmt.headers) +
+    sizeof(m.fmt.content.link_monitoring_lightning) +
+    sizeof(uint8_t) * (lightc->sent_past_links - 1) +
+    sizeof(uint16_t) * lightc->total_links;
   timespec_set_unit (&lightc->window, window, MILISEC);
 
   printf("fast_count = %d\n", lightc->fast_count);
@@ -108,9 +114,17 @@ void monitoring(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct
     exit(EXIT_FAILURE);
   }
 
-  uint8_t *prev_links = &ap->fmt.content.link_monitoring_lightning.prev_links;
+  uint8_t *prev_links = &ap->fmt.content.link_monitoring_lightning.dyn_struct;
+  uint16_t *remote_stats = (uint16_t*)(prev_links + sizeof(uint8_t) * lightc->sent_past_links);
+
   int64_t pkt_id = ap->fmt.content.link_monitoring_lightning.id;
   int64_t missing = pkt_id - (lightc->pkt_rcv_id + 1);
+  if (pkt_id > lightc->pkt_rcv_id) {
+    lightc->pkt_rcv_id = pkt_id;
+    memcpy(&lightc->remote_stats, remote_stats, sizeof(uint16_t) * lightc->total_links);
+  }
+  //printf("internal packet %ld (%ld)\n", pkt_id, missing);
+
   struct timespec now;
   set_now(&now);
 
@@ -174,11 +188,13 @@ void algo_lightning_update_stats (struct light_ctx *lightc, struct stat_entry *s
   set_now(&now);
   timespec_diff (&now, &lightc->window, &not_before);
 
+  // Init
   for (int i = 0; i < lightc->total_links; i++) {
     stats[i].link_id = i;
     stats[i].max_ooo = -1;
   }
 
+  // Compute local stats
   for (int i = 0; i < HISTORIC_SIZE; i++) {
     if (timespec_lt(&lightc->historic[i].finished_at, &not_before)) continue;
     uint8_t l = lightc->historic[i].link_id;
@@ -200,6 +216,13 @@ void algo_lightning_update_stats (struct light_ctx *lightc, struct stat_entry *s
     stats[l].max_ooo = delta > stats[l].max_ooo ? delta : stats[l].max_ooo;
   }
 
+  // Set my local stats + merge remote stats
+  for (int i = 0; i < lightc->total_links; i++) {
+    lightc->local_stats[i] = stats[i].max_ooo;
+    stats[i].max_ooo = lightc->remote_stats[i] > stats[i].max_ooo ? lightc->remote_stats[i] : stats[i].max_ooo;
+  }
+
+  // Sort
   qsort(stats, lightc->total_links, sizeof(struct stat_entry), compare_stat_entry_max);
 }
 
@@ -230,10 +253,13 @@ int send_message(struct evt_core_ctx* ctx, struct buffer_packet* bp) {
     .fmt.content.link_monitoring_lightning.id = lightc->pkt_sent_id
   };
   union abstract_packet *ap_buf = buffer_append_ap (bp_dup, &monit);
-  uint8_t *links = &ap_buf->fmt.content.link_monitoring_lightning.prev_links;
+  uint8_t *links = &ap_buf->fmt.content.link_monitoring_lightning.dyn_struct;
+  uint16_t *remote_stats = (uint16_t*)(links + sizeof(uint8_t) * lightc->sent_past_links);
+
   for (int i = 0; i < lightc->sent_past_links; i++) {
     links[i] = lightc->prev_links[(lightc->pkt_sent_id - (i + 1)) % MAX_LINKS];
   }
+  memcpy(remote_stats, &lightc->local_stats, sizeof(uint16_t) * lightc->total_links);
 
   lightc->prev_links[lightc->pkt_sent_id % MAX_LINKS] = lightc->selected_link;
 

src/packet.h

@@ -59,7 +59,7 @@ union abstract_packet {
     union {
       struct {
         uint64_t id;
-        uint8_t prev_links;
+        uint8_t dyn_struct;
       } link_monitoring_lightning;
       struct {
         uint8_t to_increment;