Working UDP and socket cleaning

src/algo_naive.c

@@ -23,7 +23,7 @@ void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
   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);
+  //printf("fd=%d accepts, creating fds=%d,%d\n", fd, conn_sock1, conn_sock2);
   evt_core_add_fd (ctx, "tcp-read", conn_sock1);
   evt_core_add_fd (ctx, "tcp-write", conn_sock2);
 
@@ -63,7 +63,7 @@ void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
   return;
 
 co_error:
-  perror("Failed to handle read write for tcp-data");
+  perror("Failed to handle read write for tcp_to_udp");
   exit(EXIT_FAILURE);
 }
 
@@ -91,13 +91,14 @@ void udp_to_tcp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) {
     nwrite = write(tcp_fd, buffer, nread);
     if (nwrite == -1 && errno == EAGAIN) return;
     if (nwrite == -1) goto co_error;
+    printf("written to tcp_fd=%d\n", nwrite);
     ring_buffer_ack_read (rb, nwrite);
   }
 
   return;
 
 co_error:
-  perror("Failed to handle read write for udp-data");
+  perror("Failed to handle read write for udp_to_tcp");
   exit(EXIT_FAILURE);
 }
 
@@ -110,7 +111,7 @@ void algo_naive(struct algo_skel* as) {
   as->on_tcp_co.socklist = NULL;
 
   as->on_tcp_read.name = "tcp-read";
-  as->on_tcp_read.flags = EPOLLIN | EPOLLET;
+  as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP;
   as->on_tcp_read.app_ctx = malloc(sizeof(struct naive_ctx));
   as->on_tcp_read.free_app_ctx = free_naive;
   as->on_tcp_read.cb = tcp_to_udp;
@@ -122,13 +123,13 @@ void algo_naive(struct algo_skel* as) {
   as->on_udp_read.flags = EPOLLIN | EPOLLET;
   as->on_udp_read.app_ctx = malloc(sizeof(struct naive_ctx));
   as->on_udp_read.free_app_ctx = free_naive;
-  as->on_udp_read.cb = tcp_to_udp;
+  as->on_udp_read.cb = udp_to_tcp;
   as->on_udp_read.socklist = NULL;
   if (as->on_udp_read.app_ctx == NULL) goto init_err;
   memset(as->on_udp_read.app_ctx, 0, sizeof(struct naive_ctx));
 
   as->on_tcp_write.name = "tcp-write";
-  as->on_tcp_write.flags = EPOLLOUT | EPOLLET;
+  as->on_tcp_write.flags = EPOLLOUT | EPOLLET | EPOLLRDHUP;
   as->on_tcp_write.app_ctx = as->on_udp_read.app_ctx;
   as->on_tcp_write.free_app_ctx = free_nothing;
   as->on_tcp_write.cb = udp_to_tcp;

src/evt_core.c

@@ -103,14 +103,16 @@ void evt_core_loop(struct evt_core_ctx* ctx) {
     }
 
     for (n = 0 ; n < num_fd; n++) {
-      if ((events[n].events & EPOLLERR) || (events[n].events & EPOLLHUP)) {
-        /* An error has occured on this fd, or the socket is not
-           ready for reading (why were we notified then?) */
+      if (events[n].events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
         int err_fd = events[n].data.fd;
-	      fprintf (stderr, "An epoll error occured with fd=%d\n", err_fd);
+        int evt = events[n].events;
+        if (evt & EPOLLRDHUP) fprintf(stderr, "Epoll Read Hup Event. ");
+        if (evt & EPOLLHUP) fprintf(stderr, "Epoll Hup Event. ");
+        if (evt & EPOLLERR) fprintf(stderr, "Epoll Err Event. ");
+	      fprintf (stderr, "Epoll error or stream closed for fd=%d. Updating our data structures...\n", err_fd);
         struct evt_core_cat* cat = g_hash_table_lookup (ctx->socklist, &err_fd);
         if (cat != NULL) {
-          fprintf(stderr, "The fd was belonging to category %s\n", cat->name);
+          fprintf(stderr, "The fd was belonging to category %s, clearing it.\n", cat->name);
           g_hash_table_remove(ctx->socklist, &err_fd);
           for (int i = 0; i < cat->socklist->len; i++) {
             if (g_array_index(cat->socklist, int, i) == err_fd) {

src/net_tools.c

@@ -4,7 +4,7 @@
 // (which seems to be copied from https://banu.com/blog/2/how-to-use-epoll-a-complete-example-in-c/epoll-example.c )
 
 int create_ip_client(char* host, char* service, int type) {
-    int err, sock;
+    int err, sock, enable;
     struct addrinfo conf;
     struct addrinfo *result, *cursor;
 
@@ -14,6 +14,7 @@ int create_ip_client(char* host, char* service, int type) {
     conf.ai_flags = 0;
     conf.ai_protocol = 0;
 
+    enable = 1;
     err = getaddrinfo(host, service, &conf, &result);
     if (err != 0) {
         fprintf(stderr, "Error with getaddrinfo() for %s:%s\n",host,service);
@@ -23,6 +24,11 @@ int create_ip_client(char* host, char* service, int type) {
     for (cursor = result; cursor != NULL; cursor = cursor->ai_next) {
         sock = socket(cursor->ai_family, cursor->ai_socktype, cursor->ai_protocol);
         if (sock == -1) continue;
+        err = setsockopt (sock, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int));
+        if (err < 0) {
+          fprintf(stderr, "Error setting socket to SO_REUSEADDR\n");
+          exit(EXIT_FAILURE);
+        }
         if (connect(sock, cursor->ai_addr, cursor->ai_addrlen) != -1) break;
         close(sock);
     }
@@ -46,7 +52,7 @@ int create_udp_client(char* host, char* service) {
 }
 
 int create_tcp_server(char* service) {
-  int err, sock;
+  int err, sock, enable;
   struct addrinfo conf;
   struct addrinfo *result, *cursor;
 
@@ -56,6 +62,7 @@ int create_tcp_server(char* service) {
   conf.ai_flags = 0; // AI_PASSIVE to listen on 0.0.0.0
   conf.ai_protocol = 0;
 
+  enable = 1;
   err = getaddrinfo(NULL, service, &conf, &result);
   if (err != 0) {
     fprintf(stderr, "Error with getaddrinfo()\n");
@@ -65,8 +72,13 @@ int create_tcp_server(char* service) {
   for (cursor = result; cursor != NULL; cursor = cursor->ai_next) {
     sock = socket(cursor->ai_family, cursor->ai_socktype, cursor->ai_protocol);
     if (sock == -1) continue;
+    err = setsockopt (sock, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int));
+    if (err < 0) {
+      fprintf(stderr, "Error setting socket to SO_REUSEADDR\n");
+      exit(EXIT_FAILURE);
+    }
     if (bind(sock, cursor->ai_addr, cursor->ai_addrlen) != -1) break;
-    perror("bind failed");
+    perror("Bind failed");
     close(sock);
   }
 

torrc_simple

@@ -0,0 +1 @@
+ControlPort 9051

udp-echo.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""
+A simple UDP echo server.
+"""
+import argparse
+import itertools
+import logging
+import socket
+
+logger = logging.getLogger(__name__)
+
+# the buffer for receiving incoming messages
+BUFFER_SIZE = 4096
+
+def receive_next(sock):
+    "Repeatedly tries receiving on the given socket until some data comes in."
+    logger.debug("Waiting to receive data...")
+    while True:
+        try:
+            return sock.recvfrom(BUFFER_SIZE)
+        except socket.timeout:
+            logger.debug("No data received yet: retrying.")
+            pass
+
+def receive_and_send_one(sock):
+    "Waits for a single datagram over the socket and echoes it back."
+    input_data, addr = receive_next(sock)
+    message = input_data.decode()
+    logger.info("Received message from %s: %s (%s bytes).", addr, message, len(input_data))
+    output_len = sock.sendto(input_data, addr)
+    logger.info("Echoed message back to %s: %s (%s bytes).", addr, message, output_len)
+
+def start(args):
+    "Runs the server."
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.settimeout(5) # seconds
+    sock.bind((args.host, args.port))
+    logger.info("Listening on %s:%s.", args.host, args.port)
+    try:
+        for i in itertools.count(1):
+            receive_and_send_one(sock)
+    finally:
+        logger.info("Shutting down.")
+        sock.close()
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument('--host', help='The host that the server socket should bind to.', default="0.0.0.0")
+    parser.add_argument('--port', help='The port that the server socket should bind to.', type=int, default=123)
+    parser.add_argument('--verbose', '-v', help="Increases the logging verbosity level.", action='count')
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
+                        format='%(asctime)s %(levelname)s %(message)s')
+    start(args)