src/tlsproxy.c: Add thread support.

[tlsproxy/tlsproxy.git] / src / tlsproxy.c

diff --git a/src/tlsproxy.c b/src/tlsproxy.c
index 508b09dfa9d0d0f6bf0e67e3bbb57d7ff804948a..6fe9dfa17448bbc073418ba770f659b3e7413735 100644 (file)
--- a/src/tlsproxy.c
+++ b/src/tlsproxy.c
@@ -17,6 +17,8 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <config.h>
+
 #include <stdlib.h>
 #include <stdio.h>
 /* socket(), bind(), accept(), listen() */
@@ -34,19 +36,34 @@
 #include <signal.h>
 /* poll() */
 #include <poll.h>
+/* errno */
+#include <errno.h>
+/* pthread_*() */
+#include <pthread.h>
 
-#include <config.h>
-
+#include "sem.h"
 
 /* Maximum line of the request line. Longer request lines are aborted with an
  * error. The standard doesn't specify a maximum line length but this should
  * be a good limit to make processing simpler. */
 #define MAX_REQUEST_LINE 4096
 
+/* Size of ringbuffer. */
+#define RINGBUFFER_SIZE 10
+
 
 /* Server should shut down. Set by SIGINT handler. */
 static volatile int done;
 
+/* Number of threads. */
+static size_t thread_count;
+/* Synchronized ring buffer storing accept()ed client sockets. */
+static int ringbuffer[RINGBUFFER_SIZE];
+static int ringbuffer_read;
+static int ringbuffer_write;
+static SEM *ringbuffer_full; /* At least one element in the buffer? */
+static SEM *ringbuffer_free; /* Space for another element in the buffer? */
+static SEM *ringbuffer_lock; /* Read lock. */
 /* Proxy hostname and port if specified on the command line. */
 static char *use_proxy_host;
 static char *use_proxy_port;
@@ -57,6 +74,7 @@ static void sigint_handler(int signal);
 static void parse_arguments(int argc, char **argv);
 static void print_usage(const char *argv);
 
+static void worker_thread(void);
 static void handle_connection(int socket);
 static int read_http_request(FILE *client_fd, char *request, size_t length);
 static void send_close_bad_request(FILE *client_fd);
@@ -76,6 +94,9 @@ int main(int argc, char **argv) {
     int client_socket, server_socket;
     struct sockaddr_in6 server_in;
 
+    size_t i;
+    pthread_t *threads;
+
     struct sigaction action;
 
     parse_arguments(argc, argv);
@@ -94,6 +115,40 @@ int main(int argc, char **argv) {
     action.sa_flags   = 0;
     sigaction(SIGINT, &action, NULL);
 
+    /* Initialize ring buffer. */
+    ringbuffer_read  = 0;
+    ringbuffer_write = 0;
+    ringbuffer_full  = sem_init(0);
+    ringbuffer_free  = sem_init(RINGBUFFER_SIZE);
+    ringbuffer_lock  = sem_init(1);
+    if (NULL == ringbuffer_full
+            || NULL == ringbuffer_free
+            || NULL == ringbuffer_lock) {
+        perror("sem_init()");
+        return EXIT_FAILURE;
+    }
+
+    /* Spawn worker threads to handle requests. */
+    threads = (pthread_t *)malloc(thread_count * sizeof(pthread_t));
+    if (NULL == threads) {
+        perror("thread malloc failed");
+        return EXIT_FAILURE;
+    }
+    for (i = 0; i < thread_count; i++) {
+        int result;
+        pthread_t thread;
+
+        result = pthread_create(&thread, NULL,
+                                (void * (*)(void *))&worker_thread,
+                                NULL);
+        if (0 != result) {
+            printf("failed to create worker thread: %s\n", strerror(result));
+            return EXIT_FAILURE;
+        }
+
+        threads[i] = thread;
+    }
+
     server_socket = socket(PF_INET6, SOCK_STREAM, 0);
     if (-1 == server_socket) {
         perror("socket()");
@@ -141,11 +196,36 @@ int main(int argc, char **argv) {
             break;
         }
 
-        handle_connection(client_socket);
+        /* No lock, we only have one producer! */
+        P(ringbuffer_free);
+        ringbuffer[ringbuffer_write] = client_socket;
+        ringbuffer_write = (ringbuffer_write + 1) % RINGBUFFER_SIZE;
+        V(ringbuffer_full);
     }
 
     close(server_socket);
 
+    /* Poison all threads and shut them down. */
+    for (i = 0; i < thread_count; i++) {
+        P(ringbuffer_free);
+        ringbuffer[ringbuffer_write] = -1; /* poison */
+        ringbuffer_write = (ringbuffer_write + 1) % RINGBUFFER_SIZE;
+        V(ringbuffer_full);
+    }
+    for (i = 0; i < thread_count; i++) {
+        errno = pthread_join(threads[i], NULL);
+        if (0 != errno) {
+            perror("pthread_join()");
+            continue;
+        }
+    }
+
+    free(ringbuffer_full);
+    free(ringbuffer_free);
+    free(ringbuffer_lock);
+
+    free(threads);
+
     free(use_proxy_host);
     free(use_proxy_port);
 
@@ -161,7 +241,10 @@ static void sigint_handler(int signal_number) {
 static void parse_arguments(int argc, char **argv) {
     int option;
 
-    while (-1 != (option = getopt(argc, argv, "p:h?"))) {
+    /* Default values. */
+    thread_count = 10;
+
+    while (-1 != (option = getopt(argc, argv, "p:t:h?"))) {
         switch (option) {
             case 'p': {
                 char *position;
@@ -193,6 +276,14 @@ static void parse_arguments(int argc, char **argv) {
 
                 break;
             }
+            case 't': {
+                if (0 >= atoi(optarg)) {
+                    fprintf(stderr, "-t positive number required\n");
+                    exit(EXIT_FAILURE);
+                }
+                thread_count = (size_t)atoi(optarg);
+                break;
+            }
             case 'h':
             default: /* '?' */
                 print_usage(argv[0]);
@@ -209,6 +300,28 @@ static void print_usage(const char *argv) {
     fprintf(stderr, "Usage: %s [-p host:port] port\n", argv);
     fprintf(stderr, "\n");
     fprintf(stderr, "-p proxy hostname and port\n");
+    fprintf(stderr, "-t number of threads [default: 10]\n");
+}
+
+static void worker_thread(void) {
+    int client_socket;
+
+    for (;;) {
+        /* Get next element from ring buffer. */
+        P(ringbuffer_full);
+        P(ringbuffer_lock);
+        client_socket = ringbuffer[ringbuffer_read];
+        ringbuffer_read = (ringbuffer_read + 1) % RINGBUFFER_SIZE;
+        V(ringbuffer_lock);
+        V(ringbuffer_free);
+
+        /* Negative value indicates we should shut down our thread. */
+        if (client_socket < 0) {
+            break;
+        }
+
+        handle_connection(client_socket);
+    }
 }
 
 static void handle_connection(int client_socket) {