WIP timer

src/algo_naive.c

@@ -193,7 +193,7 @@ int on_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   return 0;
 }
 
-void algo_naive(struct algo_skel* as) {
+void algo_naive(struct evt_core_ctx* evt, struct algo_skel* as) {
   struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx));
   if (ctx == NULL) goto init_err;
   memset(ctx, 0, sizeof(struct algo_ctx));

src/algo_rr.c

@@ -18,6 +18,7 @@ struct rr_ctx {
   uint8_t current_link;
   struct deferred_pkt real[LINK_COUNT];
   struct deferred_pkt stub[LINK_COUNT];
+  struct buffer_packet stub_pool[LINK_COUNT];
 };
 
 int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
@@ -179,7 +180,7 @@ void rr_pkt_recv(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struc
       def->bp = NULL;
       struct evt_core_fdinfo* fdinfo = evt_core_get_from_fd (ctx, def->link_fd);
       if (fdinfo == NULL) {
-        fprintf(stderr, "An error occured as the link seems to be closed for the requested link\n");
+        fprintf(stderr, "An error occured as the link seems to be closed for the requested fd\n");
         rr->recv_id++;
       }
     } while (fdinfo == NULL);
@@ -234,7 +235,33 @@ co_error:
   exit(EXIT_FAILURE);
 }
 
-void algo_rr(struct algo_skel* as) {
+int rr_on_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
+  struct buffer_packet* bp;
+
+  // 1. If has a "used" buffer, remove it
+  bp = g_hash_table_lookup (app_ctx->used_buffer, &(fdinfo->fd));
+  if (bp != NULL) {
+    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);
+  }
+
+  // 2. If appears in the write waiting queue, remove it
+  GQueue* writew = g_hash_table_lookup (app_ctx->write_waiting, &(fdinfo->fd));
+  while (writew != NULL && (bp = g_queue_pop_head (writew)) != NULL) {
+    memset(bp, 0, sizeof(struct buffer_packet));
+    g_queue_push_tail(app_ctx->free_buffer, bp);
+  }
+  g_hash_table_remove (app_ctx->write_waiting, &(fdinfo->fd));
+
+  // 3. If appears in the read waiting queue, remove it
+  g_queue_remove_all (app_ctx->read_waiting, &(fdinfo->fd));
+
+  return 0;
+}
+
+void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) {
   struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx));
   if (ctx == NULL) goto init_err;
   memset(ctx, 0, sizeof(struct algo_ctx));
@@ -253,20 +280,20 @@ void algo_rr(struct algo_skel* as) {
     g_queue_push_tail(ctx->free_buffer, &(ctx->bps[i]));
   }
 
-/*
+
   as->on_tcp_co.name = "tcp-listen";
   as->on_tcp_co.flags = EPOLLIN;
   as->on_tcp_co.free_app_ctx = free_nothing;
-  as->on_tcp_co.cb = on_tcp_co;
+  as->on_tcp_co.cb = rr_on_tcp_co;
 
   as->on_tcp_read.name = "tcp-read";
   as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP;
   as->on_tcp_read.app_ctx = ctx;
   as->on_tcp_read.free_app_ctx = free_naive;
-  as->on_tcp_read.cb = on_tcp_read;
-  as->on_tcp_read.err_cb = on_err;
+  as->on_tcp_read.cb = rr_on_tcp_read;
+  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;
@@ -282,15 +309,21 @@ void algo_rr(struct algo_skel* as) {
   as->on_tcp_write.cb = on_tcp_write;
   as->on_tcp_write.err_cb = on_err;
   ctx->ref_count++;
+*/
 
   as->on_udp_write.name = "udp-write";
   as->on_udp_write.flags = EPOLLOUT | EPOLLET;
   as->on_udp_write.app_ctx = ctx;
   as->on_udp_write.free_app_ctx = free_naive;
-  as->on_udp_write.cb = on_udp_write;
-  as->on_udp_write.err_cb = on_err;
+  as->on_udp_write.cb = rr_on_udp_write;
+  as->on_udp_write.err_cb = rr_on_err;
   ctx->ref_count++;
-*/
+
+  struct evt_core_cat tcat = {
+
+  };
+  evt_core_add_cat(evt, &tcat);
+
   return;
 init_err:
   fprintf(stderr, "Failed to init algo naive\n");

src/algo_skel.c

@@ -1,9 +1,9 @@
 #include "algo_skel.h"
 
-void init_algo(struct algo_skel* as, char* name) {
+void init_algo(struct evt_core_ctx* ctx, struct algo_skel* as, char* name) {
   for (int i = 0; i < sizeof(available_algo) / sizeof(available_algo[0]); i++) {
     if (strcmp(available_algo[i].name, name) == 0) {
-      available_algo[i].init(as);
+      available_algo[i].init(ctx, as);
       return;
     }
   }

src/algo_skel.h

@@ -16,10 +16,10 @@ struct algo_skel {
   struct evt_core_cat on_tcp_co;
 };
 
-typedef void (*algo_init)(struct algo_skel* as);
+typedef void (*algo_init)(struct evt_core_ctx* ctx, struct algo_skel* as);
 
-void init_algo(struct algo_skel* as, char* name);
-void algo_naive(struct algo_skel* as);
+void init_algo(struct evt_core_ctx* ctx, struct algo_skel* as, char* name);
+void algo_naive(struct evt_core_ctx* ctx, struct algo_skel* as);
 
 struct algo_desc {
   algo_init init;

src/donar.c

@@ -59,14 +59,15 @@ int main(int argc, char** argv) {
     if (algo == NULL) goto in_error;
 
     struct algo_skel as = {0};
-    init_algo(&as, algo);
 
     if (is_server) {
       struct donar_server_ctx ctx;
+      init_algo(&ctx.evts, &as, algo);
       if (exposed_ports->len < 1 && remote_ports->len < 1) goto in_error;
       donar_server(&ctx, &as, exposed_ports, remote_ports);
     } else if (is_client) {
       struct donar_client_ctx ctx;
+      init_algo(&ctx.evts, &as, algo);
       if ((exposed_ports->len < 1 && remote_ports->len < 1) || onion_file == NULL) goto in_error;
       donar_client(&ctx, &as, onion_file, exposed_ports, remote_ports);
     }