Implemented basic connection logic

connection.c

@@ -0,0 +1,302 @@
+#include "connection.h"
+
+#include <inttypes.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <string.h>
+
+#include <errno.h>
+#include <sys/epoll.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include "event_loop.h"
+#include "ring_buffer.h"
+#include "utils.h"
+
+/*
+ * Macros
+ */
+#define SEND_BUF_SIZE           (1024 * 200)
+#define RECV_BUF_SIZE           (1024 * 200)
+#define MAX_CONNECTIONS         (100)
+
+/*
+ * Type definitions
+ */
+typedef struct {
+    uint32_t id;
+    uint32_t ev_mask;
+    int fd;
+    rb_handle_t send_buf;
+    bool hangup;
+} conn_t;
+
+/*
+ * Data
+ */
+static uint32_t _free_id = 1;
+static conn_t *_conns[MAX_CONNECTIONS];
+static uint32_t _slots_available = MAX_CONNECTIONS;
+static uint32_t _recv_buf[RECV_BUF_SIZE];
+
+/*
+ * Private API
+ */
+// Get index of the connection with fd.
+// Parameters:
+//  - fd        - file descriptor
+// Returns:
+//  - FOUND: index of the connection in _conns
+//  - NOT FOUND: -1
+static int _get_conn_index_by_fd(int fd) {
+    for (int i = 0; i < MAX_CONNECTIONS; ++i) {
+        if (_conns[i] && _conns[i]->fd == fd) {
+            return i;
+        }
+    }
+    return -1;
+}
+
+// Get index of the connection by ID.
+// Parameters:
+//  - fd        - file descriptor
+// Returns:
+//  - FOUND: index of the connection in _conns
+//  - NOT FOUND: -1
+static int _get_conn_index_by_id(uint32_t id) {
+    for (int i = 0; i < MAX_CONNECTIONS; ++i) {
+        if (_conns[i] && _conns[i]->id == id) {
+            return i;
+        }
+    }
+    return -1;
+}
+
+// Free connection resources.
+// Parameters:
+//  - index     - index of the connection
+// Remarks:
+//  - calls close()
+//  - frees memory
+//  - does NOT call shutdown()
+//  - does NOT check if index is in bounds
+//  - does NOT check if memory is valid
+static void _free_conn(int index) {
+    conn_t *c = _conns[index];
+    if (flag_verbose) {
+        fprintf(
+            stdout,
+            "[I] CONN: #%"PRIu32" Local delete\n",
+            c->id
+        );
+    }
+    close(c->fd);
+    rb_free(c->send_buf);
+    free(c);
+    _conns[index] = NULL;
+    _slots_available++;
+}
+
+/*
+ * Event handlers.
+ * These functions return false if they have deleted the connection.
+ */
+static bool _event_epollin(int index) {
+    conn_t *c = _conns[index];
+    int res = 0;
+    do {
+        res = recv(
+            c->fd,
+            _recv_buf,
+            RECV_BUF_SIZE,
+            0
+        );
+        if (res == -1) {
+            fprintf(stderr, "[!] recv failed: %d\n", errno);
+            _free_conn(index);
+            return false;
+        }
+        if (res == 0) {
+            _free_conn(index);
+            return false;
+        }
+        if (flag_verbose) {
+            fprintf(
+                stdout,
+                "[I] CONN: #%"PRIu32" Local -> Remote (%"PRIu32" B)\n",
+                c->id,
+                res
+            );
+        }
+        // TODO: handle data in _recv_buf
+    } while (c->hangup);
+    return true;
+}
+
+static bool _event_epollout(int index) {
+    conn_t *c = _conns[index];
+    int res;
+    uint32_t to_send = rb_raw_read_size(c->send_buf);
+    res = send(
+        c->fd,
+        rb_raw_read_ptr(c->send_buf),
+        to_send,
+        MSG_NOSIGNAL
+    );
+    if (res == -1) {
+        fprintf(stderr, "[!] send failed: %d\n", errno);
+        _free_conn(index);
+        return false;
+    }
+    if (flag_verbose) {
+        fprintf(
+            stdout,
+            "[I] CONN: #%"PRIu32" Remote -> Local (%"PRIu32" B)\n",
+            c->id,
+            res
+        );
+    }
+    if (!rb_raw_read_advance(c->send_buf, res)) {
+        // no more data can be read from the ring buffer
+        c->ev_mask &= ~(uint32_t)EPOLLOUT;
+        struct epoll_event ev;
+        ev.data.fd = c->fd;
+        ev.events = c->ev_mask;
+        loop_ctl(
+            EPOLL_CTL_MOD,
+            c->fd,
+            &ev
+        );
+    }
+    return true;
+}
+
+static bool _event_epollhup(int index) {
+    if (flag_verbose) {
+        fprintf(
+            stdout,
+            "[I] CONN: #%"PRIu32" Local hangup\n",
+            _conns[index]->id
+        );
+    }
+    _conns[index]->hangup = true;
+    return true;
+}
+
+static bool _event_epollerr(int index) {
+    shutdown(_conns[index]->fd, SHUT_RDWR);
+    _free_conn(index);
+    return false;
+}
+
+/*
+ * Public API
+ */
+uint32_t connection_add(int fd) {
+    if (_slots_available == 0) {
+        return 0;
+    }
+    conn_t *conn = malloc(sizeof(conn_t));
+    if (!conn) {
+        return 0;
+    }
+    conn->id = _free_id++;
+    conn->fd = fd;
+    conn->send_buf = rb_allocate(SEND_BUF_SIZE);
+    conn->hangup = false;
+    if (!conn->send_buf) {
+        free(conn);
+    }
+    for (int i = 0; i < MAX_CONNECTIONS; ++i) {
+        if (!_conns[i]) {
+            _conns[i] = conn;
+            break;
+        }
+    }
+    _slots_available--;
+    conn->ev_mask = EPOLLIN | EPOLLRDHUP;
+    struct epoll_event ev;
+    ev.data.fd = fd;
+    ev.events = conn->ev_mask;
+    loop_ctl(
+        EPOLL_CTL_ADD,
+        fd,
+        &ev
+    );
+    return conn->id;
+}
+
+void connection_cleanup() {
+    for (int i = 0; i < MAX_CONNECTIONS; ++i) {
+        if (_conns[i]) {
+            shutdown(_conns[i]->fd, SHUT_RDWR);
+            _free_conn(i);
+        }
+    }
+}
+
+bool connection_is_socket_managed(int fd) {
+    for (int i = 0; i < MAX_CONNECTIONS; ++i) {
+        if (_conns[i] && _conns[i]->fd == fd) {
+            return true;
+        }
+    }
+    return false;
+}
+
+void connection_event(int fd, uint32_t events) {
+    int index = _get_conn_index_by_fd(fd);
+    if ((events & EPOLLHUP) && !_event_epollhup(index)) {
+        return;
+    }
+    if ((events & EPOLLRDHUP) && !_event_epollhup(index)) {
+        return;
+    }
+    if ((events & EPOLLERR) && !_event_epollerr(index)) {
+        return;
+    }
+    if ((events & EPOLLIN) && !_event_epollin(index)) {
+        return;
+    }
+    if ((events & EPOLLOUT) && !_event_epollout(index)) {
+        return;
+    }
+}
+
+uint32_t connection_send(uint32_t id, const void *data, uint32_t size) {
+    int index = _get_conn_index_by_id(id);
+    if (index == -1)
+        return 0;
+    conn_t *c = _conns[index];
+    uint32_t can_write, will_write;
+    // write to ring buffer
+    can_write = rb_raw_write_size(c->send_buf);
+    while (can_write && size) {
+        will_write = size < can_write ? size : can_write;
+        memcpy(
+            rb_raw_write_ptr(c->send_buf),
+            data,
+            will_write
+        );
+        can_write = rb_raw_write_advance(
+            c->send_buf,
+            will_write
+        );
+        size -= will_write;
+        data += will_write;
+    }
+    // add to epoll if not added
+    if (!(c->ev_mask & EPOLLOUT)) {
+        c->ev_mask |= EPOLLOUT;
+        struct epoll_event ev;
+        ev.data.fd = c->fd;
+        ev.events = c->ev_mask;
+        loop_ctl(
+            EPOLL_CTL_MOD,
+            c->fd,
+            &ev
+        );
+    }
+}

connection.h

@@ -0,0 +1,55 @@
+#ifndef __CONNECTION_H
+#define __CONNECTION_H
+
+#include <stdbool.h>
+#include <stdint.h>
+
+// Add socket to managed sockets.
+// Parameters:
+//  - fd        - socket fd
+// Returns:
+//  - on success: ID of connection (>0)
+//  - on failure: 0
+// Remarks:
+//  - in case of success, you may forget about
+//    the socket you added because it is completely
+//    managed by this module
+//  - in case of failure you still need to close
+//    the socket
+//  - does not check if fd is already added
+uint32_t connection_add(int fd);
+
+// Cleanup.
+// Remarks:
+//  - closes all connections
+void connection_cleanup();
+
+// Check if socket is managed.
+// Parameters:
+//  - fd        - socket fd
+// Returns:
+//  - true if the socket is managed
+//  - false if the socket is not managed
+bool connection_is_socket_managed(int fd);
+
+// Handle socket events.
+// Parameters:
+//  - fd        - socket fd
+//  - events    - events as returned by epoll_wait
+// Returns:
+//  - true on success
+//  - false on failure
+// Remarks:
+//  - the app should be terminated if this function returns false
+void connection_event(int fd, uint32_t events);
+
+// Send data over connection.
+// Parameters:
+//  - id        - connection ID
+//  - data      - data to send
+//  - size      - size of data
+// Returns:
+//  - count of bytes sent
+uint32_t connection_send(uint32_t id, const void *data, uint32_t size);
+
+#endif

main.c

@@ -10,6 +10,7 @@
 #include <sys/signalfd.h>
 
 #include "event_loop.h"
+#include "connection.h"
 #include "server.h"
 #include "utils.h"
 
@@ -179,6 +180,10 @@ static void on_fd_event(int fd, uint32_t events) {
             _to_work = false;
         }
     }
+    // Connection event
+    else if (connection_is_socket_managed(fd)) {
+        connection_event(fd, events);
+    }
     else {
         fprintf(stderr, "[!] Event has happened on an unknown file descriptor\n");
     }
@@ -216,6 +221,7 @@ int main(int argc, char **argv) {
     if (_is_server) {
         server_close();
     }
+    connection_cleanup();
     signals_deinit();
     loop_deinit();
     return 0;

ring_buffer.c

@@ -0,0 +1,99 @@
+#include "ring_buffer.h"
+
+#include <stdlib.h>
+#include <stdint.h>
+#include <string.h>
+
+/*
+ * Type definitions
+ */
+typedef struct {
+    char *buffer;
+    uint32_t head;
+    uint32_t tail;
+    uint32_t size;
+} ring_buffer_t;
+
+/*
+ * Public API
+ */
+rb_handle_t rb_allocate(uint32_t size) {
+    ring_buffer_t *rb = calloc(1, sizeof(ring_buffer_t));
+    if (!rb) {
+        return NULL;
+    }
+    rb->buffer = malloc(size);
+    if (!rb->buffer) {
+        free(rb);
+        rb = NULL;
+    }
+    rb->size = size;
+    return rb;
+}
+
+void rb_free(rb_handle_t _rb) {
+    if (!_rb) {
+        return;
+    }
+    ring_buffer_t *rb = _rb;
+    free(rb->buffer);
+    free(rb);
+}
+
+uint32_t rb_get_free_bytes(rb_handle_t _rb) {
+    ring_buffer_t *rb = _rb;
+    if (rb->head == rb->tail) {
+        return rb->size - 1;
+    }
+    if (rb->head < rb->tail) {
+        return rb->tail - rb->head - 1;
+    }
+    return rb->size - (rb->head - rb->tail) - 1;
+}
+
+uint32_t rb_raw_read_size(rb_handle_t _rb) {
+    ring_buffer_t *rb = _rb;
+    if (rb->head == rb->tail) {
+        return 0;
+    }
+    if (rb->head < rb->tail) {
+        return rb->size - rb->tail;
+    }
+    return rb->head - rb->tail;
+}
+
+const void *rb_raw_read_ptr(rb_handle_t _rb) {
+    ring_buffer_t *rb = _rb;
+    return rb->buffer + rb->tail;
+}
+
+uint32_t rb_raw_read_advance(rb_handle_t _rb, uint32_t size) {
+    if (size) {
+        ring_buffer_t *rb = _rb;
+        rb->tail += size;
+        rb->tail %= rb->size;
+    }
+    return rb_raw_read_size(_rb);
+}
+
+uint32_t rb_raw_write_size(rb_handle_t _rb) {
+    ring_buffer_t *rb = _rb;
+    if (rb->head < rb->tail) {
+        return rb->tail - rb->head -1;
+    }
+    return rb->size - rb->head;
+}
+
+void *rb_raw_write_ptr(rb_handle_t _rb) {
+    ring_buffer_t *rb = _rb;
+    return rb->buffer + rb->head;
+}
+
+uint32_t rb_raw_write_advance(rb_handle_t _rb, uint32_t size) {
+    if (size) {
+        ring_buffer_t *rb = _rb;
+        rb->head += size;
+        rb->head %= rb->size;
+    }
+    return rb_raw_write_size(_rb);
+}

ring_buffer.h

@@ -0,0 +1,92 @@
+#ifndef __RING_BUFFER_H
+#define __RING_BUFFER_H
+
+#include <stdint.h>
+
+/*
+ * Type definitions
+ */
+typedef void* rb_handle_t;
+
+/*
+ * Public API
+ */
+// Allocate a new ring buffer.
+// Parameters:
+//  - size      - size of buffer in bytes
+// Returns:
+//  - ring buffer handle on success
+//  - NULL on out-of-memory
+rb_handle_t rb_allocate(uint32_t size);
+
+// Free ring buffer.
+// Parameters:
+//  - rb        - ring buffer
+// Remarks:
+//  - no-op if rb is NULL
+void rb_free(rb_handle_t rb);
+
+// Get free size of ring buffer.
+// Parameters:
+//  - rb        - ring buffer
+// Returns:
+//  - count of free bytes in ring buffer
+uint32_t rb_get_free_bytes(rb_handle_t rb);
+
+
+// RAW ACCESS. Get count of bytes you can read in one call.
+// Parameters:
+//  - rb        - ring buffer
+// Returns:
+//  - count of bytes you can copy from rb_raw_read_ptr()
+// Remarks:
+//  - you may read until this function returns zero
+uint32_t rb_raw_read_size(rb_handle_t rb);
+
+// RAW ACCESS. Get pointer to the area you can read from.
+// Parameters:
+//  - rb        - ring buffer
+// Returns:
+//  - memory you can read from (use rb_raw_read_size to get max read size)
+const void *rb_raw_read_ptr(rb_handle_t rb);
+
+// RAW ACCESS. Notify the ring buffer that a read has happened.
+// Parameters:
+//  - rb        - ring buffer
+//  - size      - how much bytes were read
+// Returns:
+//  - count of bytes you can still read
+// Remarks:
+//  - no-op if size == 0 (but return value is still valid)
+//  - this function will not check for overflow
+uint32_t rb_raw_read_advance(rb_handle_t rb, uint32_t size);
+
+
+// RAW ACCESS. Get count of bytes you can write in one call.
+// Parameters:
+//  - rb        - ring buffer
+// Returns:
+//  - count of bytes you can write to rb_raw_write_ptr()
+// Remarks:
+//  - you may write until this function returns zero
+uint32_t rb_raw_write_size(rb_handle_t rb);
+
+// RAW ACCESS. Get pointer to the area you can write to.
+// Parameters:
+//  - rb        - ring buffer
+// Returns:
+//  - memory you can write to (use rb_raw_write_size to get max write size)
+void *rb_raw_write_ptr(rb_handle_t rb);
+
+// RAW ACCESS. Notify the ring buffer that a write has happened.
+// Parameters:
+//  - rb        - ring buffer
+//  - size      - how much bytes were written
+// Returns:
+//  - count of bytes you can still write
+// Remarks:
+//  - no-op if size == 0 (but return value is still valid)
+//  - this function will not check for overflow
+uint32_t rb_raw_write_advance(rb_handle_t rb, uint32_t size);
+
+#endif

server.c

@@ -10,6 +10,9 @@
 #include <sys/socket.h>
 #include <unistd.h>
 
+#include "connection.h"
+#include "utils.h"
+
 /*
  * Macros
  */
@@ -24,7 +27,21 @@ static int _fd = -1;
  * Private API
  */
 bool _accept() {
-    close(accept(_fd, 0, 0));
+    int fd = accept(_fd, 0, 0);
+    if (fd == -1) {
+        fprintf(stderr, "[!] accept failed: %d\n", errno);
+        return false;
+    }
+    uint32_t id = connection_add(fd);
+    if (!id) {
+        close(fd);
+        fprintf(stderr, "[!] connection_add failed\n");
+        // no returning false, because it would terminate the app
+    }
+    else {
+        const char* data = "Test data lalalala";
+        connection_send(id, data, strlen(data));
+    }
     return true;
 }