wip

examples/queue.c

@@ -0,0 +1,302 @@
+#include "queue.h"
+
+typedef struct queue_entry_t queue_entry_t;
+typedef struct queue_event_t queue_event_t;
+
+struct queue_entry_t {
+    queue_entry_t *prev;
+    queue_entry_t *next;
+    void *data;
+    int   events;
+    microtcp_queue_t  *queue;
+    microtcp_socket_t *socket;
+    queue_event_t *event_entry;
+};
+
+struct queue_event_t {
+
+    queue_event_t *prev;
+    queue_event_t *next;
+
+    int events;
+    queue_entry_t *socket_entry;
+};
+
+struct microtcp_queue_t {
+    
+    microtcp_t *mtcp;
+
+    pthread_cond_t something_happened;
+
+    queue_event_t *queue_head;
+    queue_event_t *queue_tail;
+
+    queue_entry_t *entry_free_list;
+    queue_entry_t  entry_pool[MICROTCP_QUEUE_ENTRIES_MAX];
+
+    queue_event_t *event_free_list;
+    queue_event_t  event_pool[MICROTCP_QUEUE_ENTRIES_MAX];
+};
+
+static event_entry_t*
+pop_event_entry_from_queue(microtcp_queue_t *queue)
+{
+    event_entry_t *event;
+
+    if (queue->queue_tail) {
+        
+        // An event is present in the queue. Pop it.
+
+        event = queue->queue_tail;
+
+        if (event->prev)
+            event->prev->next = NULL;
+        else
+            queue->queue_head = NULL;
+        queue->queue_tail = event->prev;
+
+    } else
+        event = NULL;
+
+    return event;
+}
+
+microtcp_event_t microtcp_queue_next(microtcp_queue_t *queue, bool no_block)
+{
+    microtcp_t *mtcp;
+
+    event_entry_t *event = pop_event_entry_from_queue(queue);
+
+    while (!event && !no_block) {
+        pthread_cond_wait(&queue->something_happened, &mtcp->lock);
+        event = pop_event_entry_from_queue(queue);
+    }
+
+    microtcp_event_t result;
+
+    if (event) {
+        queue_entry_t *socket_entry = event->socket_entry;
+
+        result.type   = event->events;
+        result.data   = socket_entry->data;
+        result.socket = socket_entry->socket;
+
+    } else {
+        result.type = 0;
+        result.data = NULL;
+        result.socket = NULL;
+    }
+
+    return result;
+}
+
+static void signal_socket_events_to_queues(microtcp_socket_t *socket, int events)
+{
+    queue_entry_t *entry = socket->queue_entry;
+    while (entry) {
+        
+        int interesting_events = events & entry->events;
+
+        if (interesting_events) {
+            // Some events that are of interest to 
+            // this queue are being signaled.
+            // Push them onto the queue's event queue
+            
+            microtcp_queue_t *queue = entry->queue;
+
+            if (entry->event_entry) {
+                // At least one event was already signaled, so
+                // an event structure was already created. 
+                // We just need to add to it the newly signaled
+                // events.
+                event_entry_t *event = entry->event_entry;
+                event->events |= interesting_events;
+            } else {
+
+                // No events related to this socket are in the
+                // queue, so we need to create a new event structure.
+                //
+                // Unused event structure are organized in a free
+                // list, so we just need to pop one from there.
+                // 
+                // We know for sure that a free event structure is
+                // available at this point because there is one for
+                // each socket registered into the queue.
+                assert(queue->event_free_list);
+
+                event_entry_t *event = queue->event_free_list;
+                queue->event_free_list = event->next;
+
+                // Set-up the event structure
+                event->prev = NULL;
+                event->next = NULL;
+                event->events = interesting_events;
+                event->socket_entry = entry;
+
+                // Now push the event structure into the queue's queue
+                event->next = queue->queue_head;
+                if (queue->queue_head)
+                    queue->queue_head->prev = event;
+                else
+                    queue->queue_tail = event;
+                queue->queue_head = event;
+            }
+            
+            pthread_cond_signal(&queue->something_happened);
+        }
+
+        entry = entry->next;
+    }
+}
+
+microtcp_queue_t *microtcp_queue_create(microtcp_t *mtcp)
+{
+    microtcp_queue_t *queue = malloc(sizeof(microtcp_queue_t));
+    if (!queue)
+        return NULL;
+
+    queue->mtcp = mtcp;
+    queue->queue_head = NULL;
+    queue->queue_tail = NULL;
+    queue->entry_free_list = queue->entry_pool;
+    queue->event_free_list = queue->event_pool;
+
+    for (size_t i = 0; i < MICROTCP_QUEUE_ENTRIES_MAX-1; i++) {
+        mtcp->entry_pool[i].mtcp = NULL;
+        mtcp->entry_pool[i].prev = NULL;
+        mtcp->entry_pool[i].next = mtcp->entry_pool + i+1;
+
+        mtcp->event_pool[i].mtcp = NULL;
+        mtcp->event_pool[i].prev = NULL;
+        mtcp->event_pool[i].next = mtcp->event_pool + i+1;
+    }
+    mtcp->entry_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].mtcp = NULL;
+    mtcp->entry_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].prev = NULL;
+    mtcp->entry_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].next = NULL;
+
+    mtcp->event_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].prev = NULL;
+    mtcp->event_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].next = NULL;
+
+    if (pthread_cond_init(&queue->something_happened, NULL)) {
+        free(queue);
+        return NULL;
+    }
+
+    return queue;
+}
+
+void microtcp_queue_destroy(microtcp_queue_t *queue)
+{
+    for (size_t i = 0; i < MICROTCP_QUEUE_ENTRIES_MAX; i++) {
+        if (queue->entries[i].mtcp == NULL)
+            continue;
+        microtcp_queue_unregister(queue, queue->entries[i].socket, MICROTCP_EVENT_ALL);
+    }
+    free(queue);
+}
+
+bool microtcp_queue_register(microtcp_queue_t *queue, microtcp_socket_t *socket, void *data, int events)
+{
+    if (!events)
+        return true; // No events registered
+
+    if (!queue->free_list)
+        return false; // Registration limit reached
+
+    queue_entry_t *entry = queue->free_list;
+    queue->free_list = entry->next;
+
+    entry->prev = NULL;
+    entry->next = NULL;
+    entry->data = data;
+    entry->queue  = queue;
+    entry->events = events;
+    entry->socket = socket;
+
+    entry->next = socket->queue_entry;
+    if (socket->queue_entry)
+        socket->queue_entry->prev = entry;
+    socket->queue_entry = entry;
+
+    return true;
+}
+
+static queue_entry_t *find_entry(microtcp_queue_t *queue, microtcp_socket_t *socket)
+{
+    queue_entry_t *entry = socket->queue_entry;
+    while (entry) {
+        if (entry->queue == queue)
+            return entry;
+        entry = entry->next;
+    }
+    return NULL;
+}
+
+static void 
+unregister_socket_from_all_queues(microtcp_socket_t *socket)
+{
+    while (socket->queue_entry)
+        microtcp_queue_unregister(socket->queue_entry->queue, socket, MICROTCP_EVENT_ALL);
+}
+
+bool microtcp_queue_unregister(microtcp_queue_t *queue, microtcp_socket_t *socket, int unregister_events)
+{
+    if (!queue || !socket || queue->mtcp != socket->mtcp)
+        return false;
+
+    queue_entry_t *entry = find_entry(queue, socket);
+    if (!entry)
+        return false; // Socket wasn't registered in this queue
+
+    int remaining_events = entry->events & ~unregister_events;
+
+    if (!remaining_events) {
+
+        // All events were unregistered, so the 
+        // socket must be removed from the queue.
+
+        // Unlink any events related to this socket
+        // from the queue
+        queue_event_t *event = entry->event_entry;
+        if (event) {
+
+            // Remove the event structure from the queue
+            if (event->prev)
+                event->prev->next = event->next;
+            else
+                queue->queue_head = event->next;
+
+            if (event->next)
+                event->next->prev = event->prev;
+            else
+                queue->queue_tail = event->prev;
+
+            // Push the now unlinked event structure onto
+            // the free list
+            event->prev = NULL;
+            event->next = queue->event_free_list;
+            queue->event_free_list = event;
+        }
+
+        // Unlink the entry from the socket's list
+        if (entry->prev)
+            entry->prev->next = entry->next;
+        else
+            socket->queue_entry = entry->next;
+
+        if (entry->next)
+            entry->next->prev = entry->prev;
+
+        // Put the entry structure back into the 
+        // free list of the queue
+        entry->mtcp = NULL; // It's important to set this to NULL because
+                            // this way the queue knows it's not used anymore.
+        entry->prev = NULL;
+        entry->next = queue->free_list;
+        queue->free_list = entry;   
+    
+    } else
+        entry->events = remaining_events;
+
+    return true;
+}

examples/queue.h

@@ -0,0 +1,25 @@
+
+enum {
+    MICROTCP_EVENT_NONE   = 0,
+    MICROTCP_EVENT_SEND   = 1,
+    MICROTCP_EVENT_RECV   = 2,
+    MICROTCP_EVENT_ACCEPT = 4,
+
+    MICROTCP_EVENT_ALL = MICROTCP_EVENT_RECV 
+                       | MICROTCP_EVENT_SEND 
+                       | MICROTCP_EVENT_ACCEPT,
+};
+
+typedef struct microtcp_queue_t microtcp_queue_t;
+
+typedef struct {
+    int   type;
+    void *data;
+    microtcp_socket_t *socket;
+} microtcp_event_t;
+
+microtcp_queue_t *microtcp_queue_create(microtcp_t *mtcp);
+void              microtcp_queue_destroy(microtcp_queue_t *queue);
+microtcp_event_t  microtcp_queue_next(microtcp_queue_t *queue, bool no_block);
+bool              microtcp_queue_register(microtcp_queue_t *queue, microtcp_socket_t *socket, void *data, int events);
+bool              microtcp_queue_unregister(microtcp_queue_t *queue, microtcp_socket_t *socket, int unregister_events);

examples/queue2.c

@@ -0,0 +1,54 @@
+
+typedef struct {
+    size_t   size;
+    size_t   used;
+    uint8_t *data;
+} buffer_t;
+
+typedef struct client_t client_t;
+struct client_t {
+    client_t *prev;
+    client_t *next;
+    microtcp_socket_t *socket;
+    buffer_t  input;
+    buffer_t output;
+};
+
+int main(void)
+{
+    microtcp_t *mtcp = microtcp_create();
+
+    microtcp_socket_t *server = microtcp_open(mtcp, 80);
+
+    microtcp_queue_t *queue = microtcp_queue_create(mtcp);
+    microtcp_queue_register(queue, server, NULL, MICROTCP_EVENT_ACCEPT);
+    while (1) {
+        microtcp_event_t event = microtcp_queue_next(queue, no_block);
+        switch (event.type) {
+            
+            case MICROTCP_EVENT_NONE:
+            break;
+            
+            case MICROTCP_EVENT_READ:
+            break;
+
+            case MICROTCP_EVENT_RECV:
+            {
+                microtcp_socket_t *client_socket = event.socket;
+                microtcp_recv(client_socket, );
+            }
+            break;
+            
+            case MICROTCP_EVENT_ACCEPT:
+            {
+                microtcp_socket_t *client_socket = microtcp_accept(event.socket, true);
+                microtcp_queue_register(queue, client_socket, NULL, MICROTCP_EVENT_SEND | MICROTCP_EVENT_RECV);
+            }
+            break;
+        }
+    }
+    microtcp_queue_destroy(queue);
+
+    microtcp_destroy(mtcp);
+    return 0;
+}