Rewrite buffer management, not tested

src/algo_naive.c

@@ -65,13 +65,13 @@ int on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   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);
+    mv_buffer_wtof (app_ctx, fdinfo);
     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);
+  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo);
   on_udp_write(ctx, to_fdinfo);
 
   return 0;
@@ -97,7 +97,7 @@ int on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 
   // 3. A whole packet has been written
   // Release the buffer and notify
-  mv_buffer_wtor(app_ctx, fdinfo, bp);
+  mv_buffer_wtof(app_ctx, fdinfo);
   notify_read(ctx, app_ctx);
 
   return 0;
@@ -127,13 +127,13 @@ int on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   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);
+    mv_buffer_wtof (app_ctx, fdinfo);
     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);
+  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo);
   on_tcp_write(ctx, to_fdinfo);
 
   return 0;
@@ -158,7 +158,7 @@ int on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 
   // 3. A whole packet has been written
   // Release the buffer and notify
-  mv_buffer_wtor(app_ctx, fdinfo, bp);
+  mv_buffer_wtof(app_ctx, fdinfo);
   notify_read(ctx, app_ctx);
 
   return 0;
@@ -199,6 +199,7 @@ void algo_naive(struct evt_core_ctx* evt, struct algo_skel* as) {
   memset(ctx, 0, sizeof(struct algo_ctx));
   ctx->free_buffer = g_queue_new ();
   ctx->read_waiting = g_queue_new ();
+  ctx->application_waiting = g_hash_table_new (NULL, NULL);
   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++) {

src/algo_rr.c

@@ -12,7 +12,8 @@ struct waited_pkt {
 
 struct deferred_pkt {
   int link_fd;
-  struct buffer_packet* bp;
+  int idx;
+  uint8_t on;
 };
 
 struct rr_ctx {
@@ -125,7 +126,7 @@ void rr_pkt_register(struct evt_core_ctx* ctx,  struct evt_core_fdinfo* fdinfo,
   if (ring_gt(rr->recv_id, bp->ip.ap.str.id - 1)) {
     // Packet has already been delivered or dropped, we free the buffer
     fprintf(stderr, "Packet %d arrived too late (current: %d)\n", bp->ip.ap.str.id, rr->recv_id);
-    mv_buffer_wtor (app_ctx, fdinfo, bp);
+    mv_buffer_wtof (app_ctx, fdinfo);
     return;
   }
 
@@ -144,9 +145,10 @@ void rr_pkt_register(struct evt_core_ctx* ctx,  struct evt_core_fdinfo* fdinfo,
 
   // 4. We queue the packet
   int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
-  rr->real[idx_real].bp = bp;
+  rr->real[idx_real].on = 1;
+  rr->real[idx_real].idx = idx_real;
   rr->real[idx_real].link_fd = fdinfo->fd;
-  g_hash_table_remove(app_ctx->used_buffer, &fdinfo->fd); // We remove the packet from the reading buffer
+  mv_buffer_rtoa(app_ctx, fdinfo, &rr->real[idx_real].idx);
 
   // 5. We make sure that the remote link is set to up
   char buffer[16];
@@ -155,18 +157,23 @@ void rr_pkt_register(struct evt_core_ctx* ctx,  struct evt_core_fdinfo* fdinfo,
   rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working
 }
 
-void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
+void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct deferred_pkt* dp) {
   struct evt_core_fdinfo *to_fdinfo = NULL;
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
   struct rr_ctx* rr = app_ctx->misc;
   char url[255];
 
+  // 1. Marked the packet as handled
+  dp->on = 0;
+
+  // 2. Get the buffer
+  struct buffer_packet* bp = get_app_buffer (app_ctx, &dp->idx);
   //printf("Selected url %s for pkt %d to be delivered\n", fdinfo->url, bp->ip.ap.str.id);
 
-  // 0. We update our cursor
+  // 3. We update our cursor
   rr->recv_id = bp->ip.ap.str.id;
 
-  // 1. We check that we don't have a running timeout
+  // 4. We check that we don't have a running timeout
   int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
   if (rr->wait[idx_real].on) {
     rr->wait[idx_real].on = 0;
@@ -174,22 +181,24 @@ void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct
     printf("Removed timer for packet %d\n",bp->ip.ap.str.id);
   }
 
-  // 2. We free the buffer if it's a control packet and quit
+  // 5. We free the buffer if it's a control packet and quit
   if (bp->ip.ap.str.flags & PKT_CONTROL) {
-    mv_buffer_wtor (app_ctx, fdinfo, bp);
+    mv_buffer_atof (app_ctx, &dp->idx);
     return;
   }
 
-  // 3. A whole packet has been read, we will find its target
+  // 6. Find its target
   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 udp:write for tcp-read. Dropping packet :( \n", url);
-    mv_buffer_wtor (app_ctx, fdinfo, bp);
+    //mv_buffer_wtor (app_ctx, fdinfo, bp);
+    mv_buffer_atof (app_ctx, &dp->idx);
   }
 
   // 4. We move the buffer and notify the target
-  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
+  //mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp);
+  mv_buffer_atow (app_ctx, &dp->idx, to_fdinfo);
   rr_on_udp_write(ctx, to_fdinfo);
 }
 
@@ -200,9 +209,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
 
   while(1) {
     struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE];
-    if (def->bp == NULL) break;
+    if (!def->on) break;
-    bp = def->bp;
-    def->bp = NULL;
     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 fd\n");
@@ -210,7 +217,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
       continue;
     }
 
-    rr_deliver(ctx, fdinfo, bp);
+    rr_deliver(ctx, fdinfo, def);
   }
 }
 
@@ -257,7 +264,7 @@ int rr_on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 
   // 3. A whole packet has been written
   // Release the buffer and notify
-  mv_buffer_wtor(app_ctx, fdinfo, bp);
+  mv_buffer_wtof(app_ctx, fdinfo);
   notify_read(ctx, app_ctx);
 
   return 0;
@@ -323,13 +330,13 @@ int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   // 4. A whole packet has been read, we will find someone to write it
   if (to_fdinfo == NULL) {
     fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", fdinfo->url);
-    mv_buffer_wtor (app_ctx, fdinfo, bp);
+    mv_buffer_wtof (app_ctx, fdinfo);
     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);
+  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo);
   rr_on_tcp_write(ctx, to_fdinfo);
 
   return 0;
@@ -356,7 +363,7 @@ int rr_on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
 
   // 3. A whole packet has been written
   // Release the buffer and notify
-  mv_buffer_wtor(app_ctx, fdinfo, bp);
+  mv_buffer_wtof(app_ctx, fdinfo);
   notify_read(ctx, app_ctx);
 
   return 0;
@@ -419,6 +426,7 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) {
   memset(ctx, 0, sizeof(struct algo_ctx));
   ctx->free_buffer = g_queue_new ();
   ctx->read_waiting = g_queue_new ();
+  ctx->application_waiting = g_hash_table_new (NULL, NULL);
   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);
   struct rr_ctx* rr = malloc(sizeof(struct rr_ctx));

src/algo_utils.c

@@ -76,31 +76,98 @@ struct buffer_packet* get_write_buffer(struct algo_ctx *app_ctx, struct evt_core
   return bp;
 }
 
-void mv_buffer_rtow(struct algo_ctx* app_ctx,
+void mv_buffer_rtow(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, struct evt_core_fdinfo* to) {
-                    struct evt_core_fdinfo* from,
-                    struct evt_core_fdinfo* to,
-                    struct buffer_packet* bp) {
-
-  // 1. We get the target writing queue
   GQueue* q;
+  struct buffer_packet* bp;
+
+  // 1. We get the packet buffer
+  bp = g_hash_table_lookup (app_ctx->used_buffer, &from->fd);
+  if (bp == NULL) {
+    fprintf(stderr, "Unable to find a buffer for fd=%d url=%s", from->fd, from->url);
+    exit(EXIT_FAILURE);
+  }
+
+  // 2. We get the target writing queue
   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
+  // 3. We move the data
   g_hash_table_remove(app_ctx->used_buffer, &from->fd);
   g_queue_push_tail(q, bp);
 }
 
-void mv_buffer_wtor(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
+void mv_buffer_wtof(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo) {
+  struct buffer_packet* bp = g_hash_table_lookup (app_ctx->used_buffer, &(fdinfo->fd));
+  if (bp == NULL) {
+    fprintf(stderr, "Unable to find a buffer for fd=%d url=%s", fdinfo->fd, fdinfo->url);
+    exit(EXIT_FAILURE);
+  }
   bp->mode = BP_READING;
   bp->aread = 0;
   g_queue_push_tail (app_ctx->free_buffer, bp);
   g_hash_table_remove(app_ctx->used_buffer, &(fdinfo->fd));
 }
 
+void mv_buffer_rtoa(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, void* to) {
+  struct buffer_packet* bp;
+  bp = g_hash_table_lookup (app_ctx->used_buffer, &from->fd);
+  if (bp == NULL) {
+    fprintf(stderr, "Unable to find a buffer for fd=%d url=%s\n", from->fd, from->url);
+    exit(EXIT_FAILURE);
+  }
+  g_hash_table_remove(app_ctx->used_buffer, &from->fd);
+  if (g_hash_table_contains(app_ctx->application_waiting, to)) {
+    fprintf(stderr, "Data already exist for this entry\n");
+    exit(EXIT_FAILURE);
+  }
+  g_hash_table_insert(app_ctx->application_waiting, to, bp);
+}
+
+void mv_buffer_atow(struct algo_ctx* app_ctx, void* from, struct evt_core_fdinfo* to) {
+  GQueue* q;
+  struct buffer_packet* bp;
+
+  // 1. We get the buffer
+  bp = g_hash_table_lookup (app_ctx->application_waiting, from);
+  if (bp == NULL) {
+    fprintf(stderr, "Unable to find this application buffer\n");
+    exit(EXIT_FAILURE);
+  }
+
+  // 2. We get the target writing queue
+  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);
+  }
+
+  // 3. We move the buffer
+  g_hash_table_remove (app_ctx->application_waiting, from);
+  g_queue_push_tail(q, bp);
+}
+
+void mv_buffer_atof(struct algo_ctx* app_ctx, void* from) {
+  struct buffer_packet* bp;
+
+  // 1. We get the buffer
+  bp = g_hash_table_lookup (app_ctx->application_waiting, from);
+  if (bp == NULL) {
+    fprintf(stderr, "Unable to find this application buffer\n");
+    exit(EXIT_FAILURE);
+  }
+
+  // 2. We move it
+  g_hash_table_remove (app_ctx->application_waiting, from);
+  g_queue_push_tail (app_ctx->free_buffer, bp);
+}
+
+struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx) {
+  return g_hash_table_lookup (app_ctx->application_waiting, idx);
+}
+
 void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
   struct evt_core_fdinfo* next_fdinfo = NULL;
   while (next_fdinfo == NULL) {

src/algo_utils.h

@@ -10,22 +10,27 @@ typedef void (*algo_ctx_free_misc)(void*);
 struct algo_ctx {
   int ref_count;
   struct buffer_packet bps[PACKET_BUFFER_SIZE];
-  GQueue* free_buffer;          // Available buffers
+  GQueue* free_buffer;              // Available buffers
-  GHashTable* used_buffer;      // Buffers used for reading or writing
+  GHashTable* used_buffer;          // Buffers used for reading or writing
-  GQueue* read_waiting;         // Who wait to be notified for a read
+  GQueue* read_waiting;             // Who wait to be notified for a read
-  GHashTable* write_waiting;    // Structure to track packets waiting to be written
+  GHashTable* application_waiting;  // Structure that can be used by the algo for its internal logic
-  void* misc;                   // Additional structures
+  GHashTable* write_waiting;        // Structure to track packets waiting to be written
-  algo_ctx_free_misc free_misc; // Fx ptr to free misc
+  void* misc;                       // Additional structures
+  algo_ctx_free_misc free_misc;     // Fx ptr to free misc
 };
 
-void mv_buffer_rtow(struct algo_ctx* app_ctx,
+void mv_buffer_rtow(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, struct evt_core_fdinfo* to);
-                    struct evt_core_fdinfo* from,
+void mv_buffer_wtof(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from);
-                    struct evt_core_fdinfo* to,
+void mv_buffer_rtoa(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, void* to);
-                    struct buffer_packet* bp);
+void mv_buffer_atow(struct algo_ctx* app_ctx, void* from, struct evt_core_fdinfo* to);
-void mv_buffer_wtor(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp);
+void mv_buffer_atof(struct algo_ctx* app_ctx, void* from);
+
 struct buffer_packet* get_write_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo);
 struct buffer_packet* get_read_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo);
+struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx);
+
+void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx);
+
 void free_naive(void* app_ctx);
 void free_nothing(void* app_ctx);
-void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx);
 void naive_free_simple(void* v);