Snapshot that doesn't compile

CMakeLists.txt

@@ -16,6 +16,8 @@ list(APPEND CSOURCES
 	src/donar_server.c
 	src/evt_core.h
 	src/evt_core.c
+	src/algo_skel.h
+	src/algo_naive.c
 )
 
 add_executable(donar-proxy ${CSOURCES} src/donar_proxy.c)

src/algo_naive.c

@@ -0,0 +1,90 @@
+#include "algo_skel.h"
+#define NAIVE_BUFFER 128
+
+struct naive_ctx {
+  char buffer[NAIVE_BUFFER];
+  int start;
+  int end;
+};
+
+void free_naive(void* app_ctx) {
+  if (app_ctx != NULL) free(app_ctx);
+}
+
+void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+  int conn_sock, err;
+  struct sockaddr in_addr;
+  socklen_t in_len;
+  struct epoll_event current_event;
+
+  in_len = sizeof(in_addr);
+
+  //while (1) {
+  conn_sock = accept(fd, &in_addr, &in_len);
+  //if (conn_sock == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) break;
+  if (conn_sock == -1) goto co_error;
+  evt_core_add_fd (ctx, "tcp-data", fd);
+  //}
+
+  return;
+
+co_error:
+  perror("Failed to handle new connection");
+  exit(EXIT_FAILURE);
+}
+
+void on_tcp_data(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+  struct naive_ctx* app_ctx = (struct naive_ctx*) (cat->app_ctx);
+
+  app_ctx->end = read(fd, app_ctx->buffer, sizeof(char) * NAIVE_BUFFER);
+  if (app_ctx->end == 0) return;
+  if (app_ctx->end == -1 && errno == EAGAIN) return;
+  if (app_ctx->end == -1) goto co_error;
+  app_ctx->start = write(fd, app_ctx->buffer, app_ctx->end);
+  if (app_ctx->end == -1 && errno == EAGAIN) {
+    printf("Lost data EAGAIN\n");
+    return;
+  }
+  if (app_ctx->end == -1) goto co_error;
+  if (app_ctx->start != app_ctx->end) {
+    printf("Lost data not everything has been written\n");
+    return;
+  }
+  return;
+
+co_error:
+  perror("Failed to handle read write");
+  exit(EXIT_FAILURE);
+}
+
+void on_udp_data(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
+
+}
+
+void algo_naive(struct algo_skel* as) {
+  as->on_tcp_data.name = "tcp-data";
+  as->on_tcp_data.flags = 0;
+  as->on_tcp_data.app_ctx = malloc(sizeof(struct naive_ctx));
+  as->on_tcp_data.free_app_ctx = free_naive;
+  as->on_tcp_data.cb = on_tcp_data;
+  as->on_tcp_data.socklist = NULL;
+
+  as->on_tcp_co.name = "tcp-listen";
+  as->on_tcp_co.flags = 0;
+  as->on_tcp_co.app_ctx = NULL;
+  as->on_tcp_co.free_app_ctx = free_naive;
+  as->on_tcp_co.cb = on_tcp_co;
+  as->on_tcp_co.socklist = NULL;
+
+  as->on_udp_data.name = "udp-data";
+  as->on_udp_data.flags = 0;
+  as->on_udp_data.app_ctx = malloc(sizeof(struct naive_ctx));
+  as->on_udp_data.free_app_ctx = free_naive;
+  as->on_udp_data.cb = on_udp_data;
+  as->on_udp_data.socklist = NULL;
+
+  if (as->on_tcp_data.app_ctx == NULL || as->on_udp_data.app_ctx == NULL) {
+    fprintf(stderr, "Failed to malloc naive_ctx\n");
+    exit(EXIT_FAILURE);
+  }
+}

src/algo_skel.h

@@ -0,0 +1,13 @@
+#pragma once
+#include <stdlib.h>
+#include <stdio.h>
+#include <errno.h>
+#include "evt_core.h"
+
+struct algo_skel {
+  struct evt_core_cat on_udp_data;
+  struct evt_core_cat on_tcp_data;
+  struct evt_core_cat on_tcp_co;
+};
+
+void algo_naive(struct algo_skel* as);

src/donar_proxy.c

@@ -3,6 +3,7 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <unistd.h>
+#include "algo_skel.h"
 #include "donar_client.h"
 #include "donar_server.h"
 
@@ -31,7 +32,9 @@ int main(int argc, char** argv) {
 
     if (is_server) {
       struct donar_server_ctx ctx;
-      donar_server(&ctx);
+      struct algo_skel as;
+      algo_naive (&as);
+      donar_server(&ctx, &as);
     } else if (is_client) {
       donar_client();
     }

src/donar_server.c

@@ -24,16 +24,14 @@ void destroy_resources(struct tor_os_str* tos, struct tor_ctl* tctl) {
 
 void init_tcp_servers(struct donar_server_ctx* ctx) {
   char buffer[6];
-  int err = 0;
-  for (int i = 0; i < ctx->connection_count; i++) {
+  int err, sock = 0;
+  for (int i = 0; i < PORT_SIZE; i++) {
     sprintf (buffer, "%d", ctx->ports[i]);
-    ctx->tcp_socks[i] = create_tcp_server (buffer);
-    if (ctx->tcp_socks[i] < 0) goto socket_create_err;
-    // Might be needed only with EPOLLET - might not be needed at all
-    //err = make_socket_non_blocking (ctx->tcp_socks[i]);
-    //if (err != 0) goto socket_create_err;
-    err = listen(ctx->tcp_socks[i], SOMAXCONN);
+    sock = create_tcp_server (buffer);
+    if (sock < 0) goto socket_create_err;
+    err = listen(sock, SOMAXCONN);
     if (err != 0) goto socket_create_err;
+    evt_core_add_fd(&(ctx->evts), "tcp-listen", sock);
   }
   return;
 
@@ -41,15 +39,7 @@ socket_create_err:
   fprintf(stderr, "Unable to create a TCP socket\n");
   exit(EXIT_FAILURE);
 }
-
-int is_listening_socket(struct donar_server_ctx* ctx, int sock) {
-  int i;
-  for (i = 0; i < ctx->connection_count && ctx->is_server; i++) {
-    if (sock == ctx->tcp_socks[i]) return 1;
-  }
-  return 0;
-}
-
+/*
 void handle_new_tcp_client_connection(struct donar_server_ctx* ctx, struct epoll_event* evt) {
   int conn_sock, err;
   struct sockaddr in_addr;
@@ -97,55 +87,22 @@ void handle_new_tcp_client_rw(struct donar_server_ctx* ctx, struct epoll_event*
 co_error:
   perror("Failed to handle read write");
   exit(EXIT_FAILURE);
-}
+}*/
 
-void donar_server(struct donar_server_ctx* ctx) {
-  ctx->is_server = 1;
-  ctx->connection_count = PORT_SIZE;
-  for (uint16_t i = 0; i < ctx->connection_count ; i++) {
+void donar_server(struct donar_server_ctx* ctx, struct algo_skel* algo) {
+  evt_core_init (&(ctx->evts));
+  evt_core_add_cat (&(ctx->evts), &(algo->on_udp_data));
+  evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_data));
+  evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co));
+
+  for (uint16_t i = 0; i < PORT_SIZE ; i++) {
     ctx->ports[i] = 7500 + i;
   }
-  create_onion_services (&(ctx->tos), &(ctx->tctl), ctx->ports, ctx->connection_count);
+  create_onion_services (&(ctx->tos), &(ctx->tctl), ctx->ports, PORT_SIZE);
   printf("--- Onion services created\n");
   init_tcp_servers(ctx);
 
-  ctx->epollfd = epoll_create1(0);
-  if (ctx->epollfd == -1) {
-    perror("Failed to create epoll file descriptor epoll_create1");
-    exit(EXIT_FAILURE);
-  }
+  evt_core_loop (&(ctx->evts));
 
-  struct epoll_event current_event, events[MAX_EVENTS];
-
-  for (int i = 0; i < ctx->connection_count; i++) {
-    current_event.events = EPOLLIN;
-    current_event.data.fd = ctx->tcp_socks[i];
-    if (epoll_ctl (ctx->epollfd, EPOLL_CTL_ADD, ctx->tcp_socks[i], &current_event) == -1) {
-      perror("Failed to add a file descriptor to epoll with epoll_ctl");
-      exit(EXIT_FAILURE);
-    }
-  }
-
-  printf("--- Start main loop\n");
-  int num_fd, n = 0;
-  while(1) {
-    num_fd = epoll_wait(ctx->epollfd, events, MAX_EVENTS, -1);
-    if (num_fd == -1) {
-      perror("Failed to epoll_wait");
-      exit(EXIT_FAILURE);
-    }
-
-    for (n = 0 ; n < num_fd; n++) {
-       if ((events[n].events & EPOLLERR) || (events[n].events & EPOLLHUP) || (!(events[n].events & EPOLLIN))) {
-        /* An error has occured on this fd, or the socket is not
-           ready for reading (why were we notified then?) */
-	      fprintf (stderr, "epoll error\n");
-	      close (events[n].data.fd);
-	      continue;
-	    }
-      else if (is_listening_socket (ctx, events[n].data.fd)) handle_new_tcp_client_connection (ctx, &(events[n]));
-      else handle_new_tcp_client_rw(ctx, &(events[n]));
-    }
-  }
   destroy_resources (&(ctx->tos), &(ctx->tctl));
 }

src/donar_server.h

@@ -6,17 +6,16 @@
 #include "socks5.h"
 #include "tor_os.h"
 #include "tor_ctl.h"
+#include "evt_core.h"
+#include "algo_skel.h"
+
 #define PORT_SIZE 10
-#define MAX_EVENTS 10
 
 struct donar_server_ctx {
   struct tor_os_str tos;
   struct tor_ctl tctl;
-  int connection_count;
+  struct evt_core_ctx evts;
   uint16_t ports[PORT_SIZE];
-  int tcp_socks[PORT_SIZE];
-  int epollfd;
-  int is_server;
 };
 
-void donar_server(struct donar_server_ctx* ctx);
+void donar_server(struct donar_server_ctx* ctx, struct algo_skel* algo);

src/evt_core.c

@@ -1,6 +1,8 @@
 #include "evt_core.h"
 
-void free_int(void* v) {
+void free_fd(void* v) {
+  int* fd = (int*)v;
+  close(*fd);
   free(v);
 }
 
@@ -23,7 +25,7 @@ void evt_core_init(struct evt_core_ctx* ctx) {
   }
 
   ctx->catlist = g_hash_table_new_full(g_str_hash, g_str_equal,free_char,free_cat);
-  ctx->socklist = g_hash_table_new_full(g_int_hash, g_int_equal,free_int, NULL);
+  ctx->socklist = g_hash_table_new_full(g_int_hash, g_int_equal,free_fd, NULL);
 }
 
 void evt_core_add_cat(struct evt_core_ctx* ctx, struct evt_core_cat* cat) {
@@ -54,7 +56,7 @@ void evt_core_add_cat(struct evt_core_ctx* ctx, struct evt_core_cat* cat) {
   g_hash_table_insert (ctx->catlist, key, dyn);
 }
 
-void evt_core_add_fd(struct evt_core_ctx* ctx, int fd, char* name) {
+void evt_core_add_fd(struct evt_core_ctx* ctx, char* name, int fd) {
   int* key = NULL;
   key = malloc(sizeof(int));
 
@@ -71,4 +73,45 @@ void evt_core_add_fd(struct evt_core_ctx* ctx, int fd, char* name) {
   }
 
   g_hash_table_insert(ctx->socklist, key, cat);
+  add_fd_to_epoll(ctx->epollfd, fd);
+}
+
+void evt_core_free(struct evt_core_ctx* ctx) {
+  g_hash_table_destroy(ctx->socklist);
+  g_hash_table_destroy(ctx->catlist);
+}
+
+void evt_core_loop(struct evt_core_ctx* ctx) {
+  struct epoll_event current_event, events[EVT_CORE_MAX_EVENTS];
+  struct evt_core_cat* cat;
+
+  printf("--- Start main loop\n");
+  int num_fd, n = 0;
+  while(1) {
+    num_fd = epoll_wait(ctx->epollfd, events, EVT_CORE_MAX_EVENTS, -1);
+    if (num_fd == -1) {
+      perror("Failed to epoll_wait");
+      exit(EXIT_FAILURE);
+    }
+
+    for (n = 0 ; n < num_fd; n++) {
+       if ((events[n].events & EPOLLERR) || (events[n].events & EPOLLHUP) || (!(events[n].events & EPOLLIN))) {
+        /* An error has occured on this fd, or the socket is not
+           ready for reading (why were we notified then?) */
+	      fprintf (stderr, "epoll error\n");
+	      close (events[n].data.fd);
+	      continue;
+	    }
+
+      cat = g_hash_table_lookup(ctx->socklist, &(events[n].data.fd));
+      if (cat == NULL) {
+        fprintf(stderr, "Ignoring file descriptor %d as it is not registered. This is a bug.\n", events[n].data.fd);
+        continue;
+      }
+      cat->cb(ctx, cat, events[n].data.fd);
+
+    }
+  }
+
+  evt_core_free(ctx);
 }

src/evt_core.h

@@ -1,16 +1,20 @@
 #pragma once
 #include <stdlib.h>
 #include <stdio.h>
+#include <unistd.h>
 #include <sys/epoll.h>
 #include <glib-2.0/glib.h>
 #include <glib-2.0/gmodule.h>
 #include <glib-2.0/glib-object.h>
+#include "net_tools.h"
+
+#define EVT_CORE_MAX_EVENTS 10
 
 struct evt_core_ctx;
 struct evt_core_cat;
 
 typedef void (*evt_core_free_app_ctx)(void*);
-typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_cat*);
+typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_cat*, int fd);
 
 struct evt_core_cat {
   void* app_ctx;
@@ -18,6 +22,7 @@ struct evt_core_cat {
   evt_core_cb cb;
   char* name;
   int flags;
+  GHashTable* socklist;
 };
 
 struct evt_core_ctx {
@@ -27,3 +32,7 @@ struct evt_core_ctx {
 };
 
 void evt_core_init(struct evt_core_ctx* ctx);
+void evt_core_add_cat(struct evt_core_ctx* ctx, struct evt_core_cat* cat);
+void evt_core_add_fd(struct evt_core_ctx* ctx, char* name, int fd);
+void evt_core_free(struct evt_core_ctx* ctx);
+void evt_core_loop(struct evt_core_ctx* ctx);

src/net_tools.c

@@ -113,3 +113,22 @@ void fill_buffer(size_t* written, char* dest, void *src, size_t n) {
     memcpy(dest+*written, src, n);
     *written += n;
 }
+
+/*
+ * Trying with Level Triggered for now --------
+ * Be careful, if configured as edge triggered and not level triggered
+ * You need to read everything before going back to epoll
+ * Which means keeping state too
+ */
+void add_fd_to_epoll(int epollfd, int fd) {
+  make_socket_non_blocking (fd);
+
+  struct epoll_event current_event;
+  //current_event.events = EPOLLIN | EPOLLET;
+  current_event.events = EPOLLIN;
+  current_event.data.fd = fd;
+  if (epoll_ctl (epollfd, EPOLL_CTL_ADD, fd, &current_event) == -1) {
+    perror("Failed to add a file descriptor to epoll with epoll_ctl");
+    exit(EXIT_FAILURE);
+  }
+}

src/net_tools.h

@@ -6,9 +6,11 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/epoll.h>
 
 int create_tcp_client(char* host, char* service);
 int create_tcp_server(char* service);
 int make_socket_non_blocking(int fd);
+void add_fd_to_epoll(int epollfd, int fd);
 int read_entity(int fd, void* entity, int size);
 void fill_buffer(size_t* written, char* dest, void *src, size_t n);