From d48c5753fb9c669503eacaa4ac2e90cb8f707ed8 Mon Sep 17 00:00:00 2001
From: Quentin Dufour <quentin.dufour@inria.fr>
Date: Tue, 19 Mar 2019 16:35:11 +0100
Subject: [PATCH] All the pieces should be here

---
 src/algo_rr.c | 105 +++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 99 insertions(+), 6 deletions(-)

diff --git a/src/algo_rr.c b/src/algo_rr.c
index b748948..d54e165 100644
--- a/src/algo_rr.c
+++ b/src/algo_rr.c
@@ -23,6 +23,7 @@ struct rr_ctx {
   uint16_t recv_id;
   uint16_t sent_id;
   uint8_t current_link;
+  struct timespec emit_time;
   struct deferred_pkt real[PACKET_BUFFER_SIZE];
   struct waited_pkt wait[PACKET_BUFFER_SIZE];
 };
@@ -202,6 +203,8 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
   }
 }
 
+//------
+
 int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   struct buffer_packet* bp;
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
@@ -252,6 +255,97 @@ co_error:
   exit(EXIT_FAILURE);
 }
 
+int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct buffer_packet* bp;
+  struct evt_core_fdinfo *to_fdinfo;
+  struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
+  struct rr_ctx* rr = app_ctx->misc;
+  int read_res = FDS_READY;
+  char url[255];
+
+  // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later
+  if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return 1;
+
+  // 2. Read packet from socket
+  bp->ip.ap.str.port = url_get_port_int (fdinfo->url);
+  read_res = read_packet_from_udp (fdinfo->fd, bp, fdinfo->other);
+  if (read_res == FDS_ERR) goto co_error;
+  if (read_res == FDS_AGAIN) return 1;
+
+  // 3. Prepare RR state and packet values
+  struct timespec curr;
+  int secs, nsecs;
+  uint64_t micro_sec;
+
+  if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){
+    perror("clock_gettime error");
+    exit(EXIT_FAILURE);
+  }
+
+  secs = curr.tv_sec - rr->emit_time.tv_sec;
+  nsecs = curr.tv_nsec - rr->emit_time.tv_nsec;
+  micro_sec = secs * 1000000 + nsecs / 1000;
+  if (micro_sec > rr->mjit) micro_sec = rr->mjit;
+
+  bp->ip.ap.str.id = rr->sent_id;
+  bp->ip.ap.str.flags = 0;
+  bp->ip.ap.str.deltat = micro_sec;
+  bp->ip.ap.str.bitfield = rr->remote_links;
+  bp->ip.ap.str.prevlink = rr->current_link;
+
+  do {
+    rr->current_link = (rr->current_link) + 1 % 10;
+  } while (!(rr->my_links & (1 << rr->current_link)));
+  rr->emit_time = curr;
+  rr->sent_id++;
+
+  // 4. A whole packet has been read, we will find someone to write it
+  sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + rr->current_link); //@FIXME Hardcoded
+  to_fdinfo = evt_core_get_from_url (ctx, url);
+  if (to_fdinfo == NULL) {
+    fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", url);
+    mv_buffer_wtor (app_ctx, fdinfo, bp);
+    return 1;
+  }
+  //printf("Pass packet from %s to %s\n", fdinfo->url, url);
+
+  // 5. We move the buffer and notify the target
+  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
+  rr_on_tcp_write(ctx, to_fdinfo);
+
+  return 0;
+
+co_error:
+  perror("Failed to UDP read");
+  exit(EXIT_FAILURE);
+}
+
+int rr_on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct buffer_packet* bp;
+  struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
+  int write_res = FDS_READY;
+
+  // 1. Get current write buffer OR a buffer from the waiting queue OR leave
+  if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return 1;
+
+  // 2. Write data from the buffer to the socket
+  while (bp->mode == BP_WRITING) {
+    write_res = write_packet_to_tcp(fdinfo->fd, bp);
+    if (write_res == FDS_ERR) goto co_error;
+    if (write_res == FDS_AGAIN) return 1;
+  }
+
+  // 3. A whole packet has been written
+  // Release the buffer and notify
+  mv_buffer_wtor(app_ctx, fdinfo, bp);
+  notify_read(ctx, app_ctx);
+
+  return 0;
+co_error:
+  perror("Failed to TCP write");
+  exit(EXIT_FAILURE);
+}
+
 int rr_on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
   struct rr_ctx* rr = app_ctx->misc;
@@ -327,23 +421,22 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) {
   as->on_tcp_read.err_cb = rr_on_err;
   ctx->ref_count++;
 
-/*
+
   as->on_udp_read.name = "udp-read";
   as->on_udp_read.flags = EPOLLIN | EPOLLET;
   as->on_udp_read.app_ctx = ctx;
   as->on_udp_read.free_app_ctx = free_naive;
-  as->on_udp_read.cb = on_udp_read;
-  as->on_udp_read.err_cb = on_err;
+  as->on_udp_read.cb = rr_on_udp_read;
+  as->on_udp_read.err_cb = rr_on_err;
   ctx->ref_count++;
 
   as->on_tcp_write.name = "tcp-write";
   as->on_tcp_write.flags = EPOLLOUT | EPOLLET | EPOLLRDHUP;
   as->on_tcp_write.app_ctx = ctx;
   as->on_tcp_write.free_app_ctx = free_naive;
-  as->on_tcp_write.cb = on_tcp_write;
-  as->on_tcp_write.err_cb = on_err;
+  as->on_tcp_write.cb = rr_on_tcp_write;
+  as->on_tcp_write.err_cb = rr_on_err;
   ctx->ref_count++;
-*/
 
   as->on_udp_write.name = "udp-write";
   as->on_udp_write.flags = EPOLLOUT | EPOLLET;