WIP rr

src/algo_rr.c

@@ -1,6 +1,145 @@
+#include <sys/timerfd.h>
 #include "algo_skel.h"
 #include "algo_utils.h"
 
+int rr_on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  int conn_sock1, conn_sock2;
+  struct sockaddr_in addr;
+  socklen_t in_len;
+  char url[1024], port[6];
+  struct evt_core_cat local_cat = {0};
+  struct evt_core_fdinfo to_fdinfo = {0};
+  to_fdinfo.cat = &local_cat;
+  to_fdinfo.url = url;
+
+  in_len = sizeof(addr);
+  conn_sock1 = accept(fdinfo->fd, (struct sockaddr*)&addr, &in_len);
+
+  if (conn_sock1 == -1) goto co_error;
+  conn_sock2 = dup(conn_sock1);
+  if (conn_sock2 == -1) goto co_error;
+  //printf("fd=%d accepts, creating fds=%d,%d\n", fd, conn_sock1, conn_sock2);
+
+  url_get_port(port, fdinfo->url);
+
+  to_fdinfo.fd = conn_sock1;
+  to_fdinfo.cat->name = "tcp-read";
+  sprintf(to_fdinfo.url, "tcp:read:127.0.0.1:%s", port);
+  evt_core_add_fd (ctx, &to_fdinfo);
+
+  to_fdinfo.fd = conn_sock2;
+  to_fdinfo.cat->name = "tcp-write";
+  sprintf(to_fdinfo.url, "tcp:write:127.0.0.1:%s", port);
+  evt_core_add_fd (ctx, &to_fdinfo);
+
+  return 1;
+
+co_error:
+  perror("Failed to handle new connection");
+  exit(EXIT_FAILURE);
+}
+
+void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec) {
+  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);
+  }
+
+  timer_config.it_value.tv_sec = now.tv_sec + micro_sec / 1000;
+  timer_config.it_value.tv_nsec = now.tv_nsec + micro_sec % 1000 * 1000000;
+  timer_config.it_interval.tv_sec = 60;
+  timer_config.it_interval.tv_nsec = 0;
+
+  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 = "timeout";
+  fdinfo.other = NULL;
+  fdinfo.free_other = NULL;
+  sprintf(fdinfo.url, "timer:%ld:1", micro_sec);
+  evt_core_add_fd (evts, &fdinfo);
+}
+
+void rr_handle_recv(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) {
+  struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
+
+  // 1. Update links I can used thanks to target feedback
+  if (bp->ip.ap.str.bitfield ^ app_ctx->my_bitfield) {
+    update_my_bitfield(bp, app_ctx);
+  }
+
+  // 2. If packet arrived too late
+  if (app_ctx->recv_id >  bp->ip.ap.str.id - 1) {
+    // Packet has already been delivered or dropped, we free the buffer
+    g_hash_table_remove (app_ctx->used_buffer, &(fdinfo->fd));
+    memset(bp, 0, sizeof(struct buffer_packet));
+    g_queue_push_tail(app_ctx->free_buffer, bp);
+  }
+
+  // 3. If packet arrived too early
+  else if (app_ctx->recv_id < bp->ip.ap.str.id - 1) {
+    int64_t timeout = app_ctx->mjit - (int64_t) bp->ip.ap.str.deltat;
+    if (timeout <= 0) timeout = 0;
+    //bp->ip.ap.str.id;
+    set_timeout(ctx, timeout);
+  }
+}
+
+int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct buffer_packet* bp;
+  struct evt_core_fdinfo *to_fdinfo = NULL;
+  struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
+  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. Try to read a whole packet in the buffer
+  while (bp->mode == BP_READING) {
+    read_res = read_packet_from_tcp (fdinfo->fd, bp);
+    if (read_res == FDS_ERR) goto co_error;
+    if (read_res == FDS_AGAIN) return 1;
+  }
+
+  // 3. Logic on packet
+  rr_handle_recv(bp, app_ctx, fdinfo);
+
+  /*
+  // 3. A whole packet has been read, we will find someone to write it
+  sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port);
+  to_fdinfo = evt_core_get_from_url (ctx, url);
+  if (to_fdinfo == NULL) {
+    fprintf(stderr, "No fd for URL %s in tcp-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);
+
+  // 4. We move the buffer and notify the target
+  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
+  on_udp_write(ctx, to_fdinfo);
+*/
+  return 0;
+co_error:
+  perror("Failed to TCP read");
+  exit(EXIT_FAILURE);
+}
+
 void algo_rr(struct algo_skel* as) {
   struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx));
   if (ctx == NULL) goto init_err;

src/packet.h

@@ -34,6 +34,9 @@ union abstract_packet {
     uint16_t size;
     uint16_t port;
     uint8_t id;
+    uint8_t bitfield;
+    uint8_t prevlink;
+    uint16_t deltat;
     char payload;
   } str;
 };