From 5e291f21a741483041d2eca2d6e5157efffa8db7 Mon Sep 17 00:00:00 2001
From: Quentin Dufour <quentin@dufour.io>
Date: Mon, 18 Feb 2019 21:55:53 +0100
Subject: [PATCH] Compiling solution

---
 src/algo_naive.c   | 343 ++++++++++++++++++++++++++-------------------
 src/algo_skel.h    |   1 -
 src/donar.c        |   2 +-
 src/donar_client.c |   1 -
 src/donar_server.c |   1 -
 src/evt_core.c     |  21 +--
 src/evt_core.h     |   4 +-
 src/meas_lat.c     |   6 +-
 src/packet.c       |   8 +-
 src/packet.h       |   7 +-
 10 files changed, 222 insertions(+), 172 deletions(-)

diff --git a/src/algo_naive.c b/src/algo_naive.c
index 04593f0..0f9d8f8 100644
--- a/src/algo_naive.c
+++ b/src/algo_naive.c
@@ -2,14 +2,27 @@
 
 struct naive_ctx {
   int ref_count;
-  GHashTable* packet_buffer;
+  struct buffer_packet bps[10];
+  GQueue* free_buffer; // Available buffers
+  GHashTable* used_buffer; // Buffers used for reading or writing
+  GQueue* read_waiting; // Who wait to be notified for a read
+  GHashTable* write_waiting; // Structure to track packets waiting to be written
 };
 
+void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
+void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
+void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
+void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo);
+
 void free_nothing(void* app_ctx) {}
 void free_naive(void* app_ctx) {
   struct naive_ctx* ctx = (struct naive_ctx*) app_ctx;
   ctx->ref_count--;
-  if (ctx->ref_count <= 0) free(ctx);
+  if (ctx->ref_count > 0) return;
+  g_queue_free(ctx->free_buffer);
+  g_queue_free(ctx->read_waiting);
+  g_hash_table_destroy (ctx->used_buffer);
+  free(ctx);
 }
 
 char* get_port(char* out, char* in) {
@@ -17,39 +30,35 @@ char* get_port(char* out, char* in) {
   return out;
 }
 
-void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
-  int conn_sock1, conn_sock2, port;
+void 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];
+  char url[1024], port[6];
   struct evt_core_cat local_cat = {0};
-  struct evt_core_fdinfo fdinfo = {0};
-  fdinfo.cat = &local_cat;
-  fdinfo.url = url;
+  struct evt_core_fdinfo to_fdinfo = {0};
+  to_fdinfo.cat = &local_cat;
+  to_fdinfo.url = url;
 
   in_len = sizeof(addr);
-  conn_sock1 = accept(fd, (struct sockaddr*)&addr, &in_len);
+  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);
 
-  struct evt_core_fdinfo* listen_info = g_hash_table_lookup(ctx->socklist, &fd);
-  if (listen_info == NULL) goto co_error;
-  sscanf(listen_info->url, "tcp:listen:127.0.0.1:%d", &port);
+  get_port(port, fdinfo->url);
 
-  fdinfo.fd = conn_sock1;
-  fdinfo.cat->name = "tcp-read";
-  sprintf(fdinfo.url, "tcp:read:127.0.0.1:%d", port);
-  evt_core_add_fd (ctx, &fdinfo);
+  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);
 
-  fdinfo.fd = conn_sock2;
-  fdinfo.cat->name = "tcp-write";
-  sprintf(fdinfo.url, "tcp:write:127.0.0.1:%d", port);
-  evt_core_add_fd (ctx, &fdinfo);
-
-  printf("Selected port: %d\n", port);
+  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;
 
@@ -58,59 +67,115 @@ co_error:
   exit(EXIT_FAILURE);
 }
 
-struct buffer_packet* get_bp(struct evt_core_cat* cat, int fd, int is_read_buffer) {
+/**
+ * Returns a buffer if available, NULL otherwise
+ */
+struct buffer_packet* get_read_buffer(struct naive_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
   struct buffer_packet* bp;
-  struct naive_ctx* ctx = cat->app_ctx;
 
-  bp = g_hash_table_lookup (ctx->packet_buffer, &fd);
+  // 1. Check if we don't have a buffer
+  bp = g_hash_table_lookup (app_ctx->used_buffer, &fdinfo->fd);
   if (bp != NULL) return bp;
-  if (!is_read_buffer) goto alloc_error;
 
-  bp = malloc(sizeof(struct buffer_packet));
-  if (bp == NULL) goto alloc_error;
-  memset(bp, 0, sizeof(struct buffer_packet));
-  bp->fdread = fd;
-  bp->fdwrite = -1;
-  bp->ref_count++;
-  g_hash_table_insert(ctx->packet_buffer, &(bp->fdread), bp);
+  // 2. Get a new buffer otherwise
+  bp = g_queue_pop_head(app_ctx->free_buffer);
+  if (bp == NULL) {
+    // 2.1 If no buffer is available, we subscribe to be notified later
+    g_queue_push_tail (app_ctx->read_waiting, &(fdinfo->fd));
+    return NULL;
+  }
+
+  // 3. Update state
+  g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp);
+
   return bp;
-
-alloc_error:
-  perror("alloc error");
-  exit(EXIT_FAILURE);
 }
 
-void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd);
-void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd);
-void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd);
-void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd);
+/**
+ * Returns a buffer if available, NULL otherwise
+ */
+struct buffer_packet* get_write_buffer(struct naive_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
+  struct buffer_packet* bp;
+  GQueue* q;
 
-void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+  // 1. Check if we don't have a buffer
+  bp = g_hash_table_lookup (app_ctx->used_buffer, &fdinfo->fd);
+  if (bp != NULL) return bp;
+
+  // 2. Check our waiting queue otherwise
+  if ((q = g_hash_table_lookup(app_ctx->write_waiting, &(fdinfo->fd))) == NULL) return NULL;
+  bp = g_queue_pop_head(q);
+  if (bp == NULL) return NULL; // No packet to process
+
+  // 3. Update state
+  g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp);
+
+  return bp;
+}
+
+void mv_buffer_rtow(struct naive_ctx* app_ctx,
+                    struct evt_core_fdinfo* from,
+                    struct evt_core_fdinfo* to,
+                    struct buffer_packet* bp) {
+
+  // 1. We get the target writing queue
+  GQueue* q;
+  q = g_hash_table_lookup(app_ctx->write_waiting, &(to->fd));
+  if (q == NULL) {
+    q = g_queue_new ();
+    g_hash_table_insert(app_ctx->write_waiting, &(to->fd), q);
+  }
+
+  // 2. We move the buffer to the target queue
+  g_hash_table_remove(app_ctx->used_buffer, &from->fd);
+  g_queue_push_tail(q, bp);
+}
+
+void mv_buffer_wtor(struct naive_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
+  g_queue_push_tail (app_ctx->free_buffer, bp);
+  g_hash_table_remove(app_ctx->used_buffer, &(fdinfo->fd));
+}
+
+void notify_read(struct evt_core_ctx* ctx, struct naive_ctx* app_ctx) {
+  struct evt_core_fdinfo* next_fdinfo = NULL;
+  while (next_fdinfo == NULL) {
+    int fd = GPOINTER_TO_INT(g_queue_pop_head(app_ctx->read_waiting));
+    if (fd == 0) break;
+    next_fdinfo = evt_core_get_from_fd (ctx, fd);
+    if (strcmp(next_fdinfo->cat->name, "tcp-read") == 0) on_tcp_read(ctx, next_fdinfo);
+    else if (strcmp(next_fdinfo->cat->name, "udp-read") == 0) on_udp_read(ctx, next_fdinfo);
+    else {
+      fprintf(stderr, "A fd from category %s can't be stored in read_waiting\n", next_fdinfo->cat->name);
+      exit(EXIT_FAILURE);
+    }
+  }
+}
+
+void 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 naive_ctx* app_ctx = cat->app_ctx;
+  struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
   int read_res = FDS_READY;
   char url[255];
 
-  // 1. Read in our buffer packet
-  bp = get_bp(cat, fd, TRUE);
+  // 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;
+
+  // 2. Try to read a whole packet in the buffer
   while (read_res != FDS_AGAIN && bp->mode == BP_READING) {
-    read_res = read_packet_from_tcp (fd, bp);
+    read_res = read_packet_from_tcp (fdinfo->fd, bp);
     if (read_res == FDS_ERR) goto co_error;
   }
-  if (bp->mode != BP_SWITCH_WRITE) return;
+  if (bp->mode != BP_WRITING) return;
 
-  // 2. Packet has just been read, choose a destination
+  // 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) goto co_error;
 
-  // 3. Configure packet buffer for the destination
-  bp->mode = BP_WRITING;
-  bp->fdwrite = to_fdinfo->fd;
-  bp->ref_count++;
-  g_hash_table_insert(app_ctx->packet_buffer, &(bp->fdwrite), bp);
-  on_udp_write(ctx, to_fdinfo->cat, to_fdinfo->fd);
+  // 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;
 co_error:
@@ -118,140 +183,136 @@ co_error:
   exit(EXIT_FAILURE);
 }
 
-void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   struct buffer_packet* bp;
-  struct evt_core_fdinfo *to_fdinfo;
+  struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
   int write_res = FDS_READY;
-  char url[255];
 
-  bp = get_bp(cat, fd, FALSE);
-  if (bp == NULL) return;
+  // 1. Get current write buffer OR a buffer from the waiting queue OR leave
+  if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return;
 
+  // 2. Write data from the buffer to the socket
   while (write_res != FDS_AGAIN && bp->mode == BP_WRITING) {
-    write_res = write_packet_to_tcp(fd, bp);
+    write_res = write_packet_to_tcp(fdinfo->fd, bp);
     if (write_res == FDS_ERR) goto co_error;
   }
+  if (bp->mode != BP_READING) return;
 
+  // 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;
 co_error:
   perror("Failed to TCP write");
   exit(EXIT_FAILURE);
 }
 
-void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
-
-}
-
-void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
-
-}
-
-void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
-  // Init data structures for the transfer
+void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct buffer_packet* bp;
+  struct evt_core_fdinfo *to_fdinfo;
+  struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
+  int read_res = FDS_READY;
   char url[255];
-  struct naive_ctx* app_ctx = cat->app_ctx;
-  struct buffer_packet* bp = &(app_ctx->tcp_to_udp);
-  struct evt_core_fdinfo* fdinfo;
-  int read_res = FDS_READY;
-  int write_res = FDS_READY;
 
-  // Either we can read something, either we can write something
-  while ((read_res != FDS_AGAIN && bp->mode == BP_READING)
-         || (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) {
+  // 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. Read packet from TCP socket
-    if (bp->mode == BP_READING) {
-      read_res = read_packet_from_tcp (fd, bp);
-      if (read_res == FDS_ERR) goto co_error;
-    }
+  // 2. Read packet from socket
+  read_res = read_packet_from_udp (fdinfo->fd, bp);
+  if (read_res == FDS_ERR) goto co_error;
+  if (bp->mode != BP_WRITING) return;
 
-    // 2. Write packet to UDP socket
-    if (bp->mode == BP_WRITING) {
-      sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port);
-      fdinfo = evt_core_get_from_url(ctx, url);
-      if (fdinfo == NULL) goto co_error;
-      write_res = write_packet_to_udp(fdinfo->fd, bp);
-      if (read_res == FDS_ERR) goto co_error;
-    }
-  }
-  return;
+  // 3. A whole packet has been read, we will find someone to write it
+  sprintf(url, "tcp:write:127.0.0.1:7500");
+  to_fdinfo = evt_core_get_from_url (ctx, url);
+  if (to_fdinfo == NULL) goto co_error;
+
+  // 4. We move the buffer and notify the target
+  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
+  on_tcp_write(ctx, to_fdinfo);
 
 co_error:
-  perror("Failed to handle read write for tcp_to_udp");
+  perror("Failed to UDP read");
   exit(EXIT_FAILURE);
 }
 
-void udp_to_tcp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
-  // Get target file descriptor
-  struct evt_core_cat* tcp = g_hash_table_lookup (ctx->catlist, "tcp-write");
-  if (tcp == NULL || tcp->socklist->len < 1) goto co_error;
-  int tcp_fd = g_array_index(tcp->socklist, int, 0);
-
-  // Init data structures for the transfer
-  struct naive_ctx* app_ctx = cat->app_ctx;
-  struct buffer_packet* bp = &(app_ctx->udp_to_tcp);
-  int read_res = FDS_READY;
+void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct buffer_packet* bp;
+  struct naive_ctx* app_ctx = fdinfo->cat->app_ctx;
   int write_res = FDS_READY;
 
-  // Either we can read something, either we can write something
-  while ((read_res != FDS_AGAIN && bp->mode == BP_READING)
-         || (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) {
+  // 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. Read packet from UDP socket
-    if (bp->mode == BP_READING) {
-      read_res = read_packet_from_udp (fd, bp);
-      if (read_res == FDS_ERR) goto co_error;
-    }
+  // 2. Write buffer
+  write_res = write_packet_to_udp(fdinfo->fd, bp);
+  if (write_res == FDS_ERR) goto co_error;
+  if (bp->mode != BP_READING) return;
 
-    // 2. Write packet to TCP socket
-    if (bp->mode == BP_WRITING) {
-      write_res = write_packet_to_tcp(tcp_fd, bp);
-      if (write_res == FDS_ERR) goto co_error;
-    }
-  }
-
-  return;
+  // 3. A whole packet has been written
+  // Release the buffer and notify
+  mv_buffer_wtor(app_ctx, fdinfo, bp);
+  notify_read(ctx, app_ctx);
 
 co_error:
-  perror("Failed to handle read write for udp_to_tcp");
+  perror("Failed to UDP write");
   exit(EXIT_FAILURE);
 }
 
-void free_buffer_packet(void *v) {
-  struct buffer_packet* bp = (struct buffer_packet*) v;
-  bp->ref_count--;
-  if (bp->ref_count == 0) free(v);
-}
 
-void free_simple(void* v) {
+void naive_free_simple(void* v) {
   free(v);
 }
 
+void on_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
+  struct naive_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));
+}
+
 void algo_naive(struct algo_skel* as) {
   struct naive_ctx* ctx = malloc(sizeof(struct naive_ctx));
   if (ctx == NULL) goto init_err;
   memset(ctx, 0, sizeof(struct naive_ctx));
-  ctx->packet_buffer = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, free_buffer_packet);
-
-  as->on_init.name = "init";
-  as->on_init.flags = 0;
-  as->on_init.app_ctx = NULL;
-  as->on_init.free_app_ctx = NULL;
-  as->on_init.cb = NULL;
-  as->on_init.socklist = NULL;
+  ctx->free_buffer = g_queue_new ();
+  ctx->read_waiting = g_queue_new ();
+  ctx->used_buffer = g_hash_table_new(g_int_hash, g_int_equal);
+  ctx->write_waiting = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, naive_free_simple);
+  for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) {
+    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.app_ctx = NULL;
   as->on_tcp_co.free_app_ctx = free_nothing;
   as->on_tcp_co.cb = on_tcp_co;
-  as->on_tcp_co.socklist = NULL;
 
   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.socklist = NULL;
+  as->on_tcp_read.err_cb = on_err;
   ctx->ref_count++;
 
   as->on_udp_read.name = "udp-read";
@@ -259,7 +320,7 @@ void algo_naive(struct algo_skel* as) {
   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.socklist = NULL;
+  as->on_udp_read.err_cb = on_err;
   ctx->ref_count++;
 
   as->on_tcp_write.name = "tcp-write";
@@ -267,7 +328,7 @@ void algo_naive(struct algo_skel* as) {
   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.socklist = NULL;
+  as->on_tcp_write.err_cb = on_err;
   ctx->ref_count++;
 
   as->on_udp_write.name = "udp-write";
@@ -275,7 +336,7 @@ void algo_naive(struct algo_skel* as) {
   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.socklist = NULL;
+  as->on_udp_write.err_cb = on_err;
   ctx->ref_count++;
 
   return;
diff --git a/src/algo_skel.h b/src/algo_skel.h
index bdfe972..e4a21ef 100644
--- a/src/algo_skel.h
+++ b/src/algo_skel.h
@@ -8,7 +8,6 @@
 #include "utils.h"
 
 struct algo_skel {
-  struct evt_core_cat on_init;
   struct evt_core_cat on_udp_read;
   struct evt_core_cat on_tcp_read;
   struct evt_core_cat on_udp_write;
diff --git a/src/donar.c b/src/donar.c
index 8e40703..7dfe2fe 100644
--- a/src/donar.c
+++ b/src/donar.c
@@ -74,8 +74,8 @@ in_error:
 
 terminate:
     if (onion_file != NULL) free(onion_file);
-    if (port != NULL) free(port);
     if (algo != NULL) free(algo);
+    g_ptr_array_free(ports, TRUE);
 
     return errored;
 }
diff --git a/src/donar_client.c b/src/donar_client.c
index 08f9ea9..f21f1eb 100644
--- a/src/donar_client.c
+++ b/src/donar_client.c
@@ -150,7 +150,6 @@ void donar_client(struct donar_client_ctx* ctx, struct algo_skel* algo, char* on
     .socklist = NULL
   };
   evt_core_add_cat (&(ctx->evts), &init_socks5);
-  evt_core_add_cat (&(ctx->evts), &(algo->on_init));
   evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co));
   evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read));
   evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read));
diff --git a/src/donar_server.c b/src/donar_server.c
index 8437cb8..bcebeab 100644
--- a/src/donar_server.c
+++ b/src/donar_server.c
@@ -82,7 +82,6 @@ socket_failed:
 
 void donar_server(struct donar_server_ctx* ctx, struct algo_skel* algo, GPtrArray* ports) {
   evt_core_init (&(ctx->evts));
-  evt_core_add_cat (&(ctx->evts), &(algo->on_init));
   evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co));
   evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read));
   evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read));
diff --git a/src/evt_core.c b/src/evt_core.c
index 5a6e457..2354b9b 100644
--- a/src/evt_core.c
+++ b/src/evt_core.c
@@ -52,6 +52,7 @@ void evt_core_add_cat(struct evt_core_ctx* ctx, struct evt_core_cat* cat) {
   dyn->cb = cat->cb;
   dyn->name = strdup(cat->name);
   dyn->flags = cat->flags;
+  dyn->err_cb = cat->err_cb;
   dyn->socklist = g_array_new (FALSE, FALSE, sizeof(struct evt_core_fdinfo*));
 
   if (dyn->name == NULL) {
@@ -133,11 +134,6 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
   struct evt_core_fdinfo* fdinfo;
   struct evt_core_cat* cat;
 
-  cat = g_hash_table_lookup(ctx->catlist, "init");
-  if (cat != NULL) {
-    cat->cb(ctx, cat, -1);
-  }
-
   printf("--- Start main loop\n");
   int num_fd, n = 0;
   while(1) {
@@ -156,9 +152,11 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
         if (evt & EPOLLHUP) fprintf(stderr, "Epoll Hup Event. ");
         if (evt & EPOLLERR) fprintf(stderr, "Epoll Err Event. ");
 
-        cat = evt_core_rm_fd (ctx, err_fd);
-        if (cat != NULL) {
-          fprintf(stderr, "Clearing fd=%d on cat=%s\n", err_fd, cat->name);
+        fdinfo = evt_core_get_from_fd(ctx, err_fd);
+        if (fdinfo != NULL) {
+          fprintf(stderr, "Clearing fd=%d on cat=%s\n", err_fd, fdinfo->cat->name);
+          if (fdinfo->cat->err_cb != NULL) fdinfo->cat->err_cb(ctx, fdinfo);
+          evt_core_rm_fd (ctx, err_fd);
         } else {
           fprintf(stderr, "The file descriptor is not registered in a category, this is probably a logic error\n");
 	        close (err_fd);
@@ -172,16 +170,11 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
         fprintf(stderr, "Ignoring file descriptor %d as it is not registered. This is a bug.\n", events[n].data.fd);
         continue;
       }
-      fdinfo->cat->cb(ctx, fdinfo->cat, events[n].data.fd);
+      fdinfo->cat->cb(ctx, fdinfo);
 
     }
   }
 
-  cat = g_hash_table_lookup(ctx->catlist, "destroy");
-  if (cat != NULL) {
-    cat->cb(ctx, cat, -1);
-  }
-
   evt_core_free(ctx);
 }
 
diff --git a/src/evt_core.h b/src/evt_core.h
index 63b1922..523617c 100644
--- a/src/evt_core.h
+++ b/src/evt_core.h
@@ -12,14 +12,16 @@
 
 struct evt_core_ctx;
 struct evt_core_cat;
+struct evt_core_fdinfo;
 
 typedef void (*evt_core_free_app_ctx)(void*);
-typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_cat*, int fd);
+typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_fdinfo*);
 
 struct evt_core_cat {
   void* app_ctx;
   evt_core_free_app_ctx free_app_ctx;
   evt_core_cb cb;
+  evt_core_cb err_cb;
   char* name;
   int flags;
   GArray* socklist;
diff --git a/src/meas_lat.c b/src/meas_lat.c
index 0d069ba..5f93b68 100644
--- a/src/meas_lat.c
+++ b/src/meas_lat.c
@@ -7,10 +7,10 @@ struct timer_ctx {
   uint64_t counter;
 };
 
-void on_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+void on_udp(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 }
 
-void on_timer(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+void on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 }
 
 int main(int argc, char** argv) {
@@ -23,6 +23,7 @@ int main(int argc, char** argv) {
     .app_ctx = NULL,
     .free_app_ctx = NULL,
     .cb = on_udp,
+    .err_cb = NULL,
     .name = "udp-read",
     .flags = EPOLLIN | EPOLLET,
     .socklist = NULL
@@ -31,6 +32,7 @@ int main(int argc, char** argv) {
     .app_ctx = &apptime,
     .free_app_ctx = NULL,
     .cb = on_timer,
+    .err_cb = NULL,
     .name = "timer",
     .flags = EPOLLIN | EPOLLET,
     .socklist = NULL
diff --git a/src/packet.c b/src/packet.c
index 568d926..e89ad8f 100644
--- a/src/packet.c
+++ b/src/packet.c
@@ -21,7 +21,7 @@ enum FD_STATE read_packet_from_tcp(int fd, struct buffer_packet* bp) {
     bp->aread += nread;
   }
 
-  bp->mode = BP_SWITCH_WRITE;
+  bp->mode = BP_WRITING;
   bp->awrite = 0;
 
   return FDS_READY;
@@ -38,7 +38,7 @@ enum FD_STATE write_packet_to_tcp(int fd, struct buffer_packet* bp) {
     bp->awrite += nwrite;
   }
 
-  bp->mode = BP_SWITCH_READ;
+  bp->mode = BP_READING;
   bp->aread = 0;
 
   return FDS_READY;
@@ -60,7 +60,7 @@ enum FD_STATE write_packet_to_udp(int fd, struct buffer_packet* bp) {
   if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN;
   if (nwrite != bytes_to_send) return FDS_ERR;
 
-  bp->mode = BP_SWITCH_READ;
+  bp->mode = BP_READING;
   bp->aread = 0;
 
   return FDS_READY;
@@ -81,7 +81,7 @@ enum FD_STATE read_packet_from_udp (int fd, struct buffer_packet* bp) {
 
   bp->ip.ap.str.size = nread + pkt_header_size;
 
-  bp->mode = BP_SWITCH_WRITE;
+  bp->mode = BP_WRITING;
   bp->awrite = 0;
 
   return FDS_READY;
diff --git a/src/packet.h b/src/packet.h
index 357e956..f8d1f8f 100644
--- a/src/packet.h
+++ b/src/packet.h
@@ -24,9 +24,7 @@ enum FD_STATE {
 
 enum BP_MODE {
   BP_READING,
-  BP_WRITING,
-  BP_SWITCH_WRITE,
-  BP_SWITCH_READ
+  BP_WRITING
 };
 
 union abstract_packet {
@@ -45,10 +43,7 @@ struct internet_packet {
 };
 
 struct buffer_packet {
-  uint64_t ref_count;
   uint8_t mode;
-  int fdread;
-  int fdwrite;
   uint16_t aread;
   uint16_t awrite;
   struct internet_packet ip;