Implement pad

src/algo_thunder.c

@@ -5,6 +5,9 @@
 #include "proxy.h"
 #include "timer.h"
 
+// A Tor cell size is 512 bytes but handle only 498 bytes of data
+#define TOR_CELL_SIZE 498
+
 struct thunder_ctx {
   uint16_t recv_id;
   uint16_t emit_id;
@@ -21,7 +24,7 @@ uint64_t compute_delta(struct timespec* prev_time, uint64_t max) {
   int secs, nsecs;
   uint64_t mili_sec;
 
-  // 4. We compute the time difference
+  // 1. We compute the time difference
   if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){
     perror("clock_gettime error");
     exit(EXIT_FAILURE);
@@ -39,9 +42,13 @@ void prepare(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
   struct thunder_ctx* thunderc = app_ctx->misc;
 
+  // 1. Put the raw buffer in cache
+  thunderc->emit_id++;
+  uint64_t ref = 0l + thunderc->emit_id;
+  dup_buffer_toa (&app_ctx->br, bp, (void *)ref);
+
   uint64_t delta_pkt = compute_delta(&thunderc->prev_packet_time, 200);
 
-  thunderc->emit_id++;
   union abstract_packet metadata = {
     .fmt.headers.cmd = CMD_UDP_METADATA_THUNDER,
     .fmt.headers.size = sizeof(metadata.fmt.headers) + sizeof(metadata.fmt.content.udp_metadata_thunder),
@@ -57,18 +64,28 @@ void pad(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer
   struct thunder_ctx* thunderc = app_ctx->misc;
   uint64_t ref = 0l + thunderc->emit_id;
 
-  dup_buffer_toa (&app_ctx->br, bp, (void *)ref);
-
+  // 1. Clean old buffers
   if (ref > 10 && get_app_buffer (&app_ctx->br, (void *)ref - 10l)) {
     mv_buffer_atof (&app_ctx->br, (void *)ref - 10l);
   }
 
-  //@FIXME we must add the delta t of the current packet
-  for (uint64_t add_ref = ref; add_ref >= 0 && get_app_buffer (&app_ctx->br, (void *)add_ref); add_ref--) {
-    struct buffer_packet *bpa = get_app_buffer (&app_ctx->br, (void *)add_ref);
-
+  // 2. Append abstract packets stored in our buffers
+  uint64_t add_ref = ref;
+  while(1) {
+    if (add_ref < 1) break;
+    add_ref--;
+    struct buffer_packet *bp_iter = get_app_buffer (&app_ctx->br, (void *)add_ref);
+    if (bp_iter == NULL) break;
+    union abstract_packet *ap = buffer_first_ap (bp_iter);
+    if (ap->fmt.headers.cmd != CMD_UDP_ENCAPSULATED || ap->fmt.headers.flags & FLAG_READ_NEXT) {
+      fprintf(stderr, "Invalid buffer!\n");
+      exit(EXIT_FAILURE);
     }
 
+    if (buffer_full_size (bp) + ap->fmt.headers.size > TOR_CELL_SIZE - thunderc->monit_pkt_size) break;
+
+    buffer_append_ap (bp, ap);
+  }
 }
 
 int schedule(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {

src/packet.c

@@ -7,8 +7,12 @@ union abstract_packet* ap_next(union abstract_packet* ap) {
   return NULL;
 }
 
+union abstract_packet* buffer_first_ap(struct buffer_packet* bp) {
+   return (union abstract_packet*) &bp->ip;
+}
+
 union abstract_packet* buffer_last_ap(struct buffer_packet* bp) {
-  union abstract_packet* ap = (union abstract_packet*) &bp->ip, *apn = NULL;
+  union abstract_packet* ap = buffer_first_ap (bp), *apn = NULL;
   while ((apn = ap_next(ap)) != NULL) ap = apn;
 
   return ap;
@@ -28,7 +32,7 @@ size_t buffer_count_ap(struct buffer_packet* bp) {
   return s;
 }
 
-size_t get_full_size(struct buffer_packet* bp) {
+size_t buffer_full_size(struct buffer_packet* bp) {
   return &(buffer_free_ap (bp))->raw - &bp->ip[0];
 }
 
@@ -93,8 +97,8 @@ enum FD_STATE write_packet_to_tcp(struct evt_core_fdinfo* fdinfo, struct buffer_
 
   //dump_buffer_packet (bp);
   if (bp->mode != BP_WRITING) return FDS_ERR;
-  while (bp->awrite < get_full_size(bp)) {
-    nwrite = send(fdinfo->fd, &(ap->raw) + bp->awrite, get_full_size(bp) - bp->awrite, 0);
+  while (bp->awrite < buffer_full_size(bp)) {
+    nwrite = send(fdinfo->fd, &(ap->raw) + bp->awrite, buffer_full_size(bp) - bp->awrite, 0);
     if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN;
     if (nwrite == -1) return FDS_ERR;
     bp->awrite += nwrite;
@@ -190,7 +194,7 @@ enum FD_STATE read_packet_from_udp (struct evt_core_fdinfo* fdinfo, struct buffe
 
 void dump_buffer_packet(struct buffer_packet* bp) {
   printf("<Buffer Packet>\n");
-  printf("  mode=%d, aread=%d, awrite=%d, ap_count=%d, usage=%ld/%ld\n", bp->mode, bp->aread, bp->awrite, bp->ap_count, get_full_size (bp), sizeof(bp->ip));
+  printf("  mode=%d, aread=%d, awrite=%d, ap_count=%d, usage=%ld/%ld\n", bp->mode, bp->aread, bp->awrite, bp->ap_count, buffer_full_size (bp), sizeof(bp->ip));
   union abstract_packet* ap = (union abstract_packet*) &bp->ip;
   for (int i = 0; i < bp->ap_count; i++) {
     dump_abstract_packet(ap);

src/packet.h

@@ -32,9 +32,9 @@ enum BP_MODE {
 };
 
 enum PKT_CMD {
-  CMD_UDP_ENCAPSULATED,
-  CMD_LINK_MONITORING_THUNDER,
-  CMD_UDP_METADATA_THUNDER,
+  CMD_UDP_ENCAPSULATED = 1,
+  CMD_LINK_MONITORING_THUNDER = 2,
+  CMD_UDP_METADATA_THUNDER = 3,
 };
 
 enum PKT_FLAGS {
@@ -89,8 +89,10 @@ struct udp_target {
 size_t get_full_size(struct buffer_packet* bp);
 
 union abstract_packet* buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap);
-union abstract_packet* buffer_free_ptr(struct buffer_packet* bp);
-union abstract_packet* buffer_last_ptr(struct buffer_packet* bp);
+union abstract_packet* buffer_free_ap(struct buffer_packet* bp);
+union abstract_packet* buffer_first_ap(struct buffer_packet* bp);
+union abstract_packet* buffer_last_ap(struct buffer_packet* bp);
+size_t buffer_full_size(struct buffer_packet* bp);
 union abstract_packet* ap_next(union abstract_packet* ap);
 
 enum FD_STATE read_packet_from_tcp(struct evt_core_fdinfo* fd, struct buffer_packet* bp);