feat: implement io_uring based event dispatcher for linux

CMakeLists.txt

@@ -17,8 +17,16 @@ if(APPLE)
   add_executable(fkvs-benchmark src/fkvs-benchmark.c src/client.c src/core/list.c src/networking/networking.c src/commands/common/command_parser.c src/commands/client/client_command_handlers.c)
   target_compile_definitions(fkvs-benchmark PRIVATE CLI)
 elseif(LINUX)
-  add_executable(fkvs-server src/memory.c src/counter.c src/client.c src/core/list.c src/config.c src/networking/networking.c src/server.c src/core/hashtable.c src/commands/common/command_registry.c src/commands/server/server_command_handlers.c src/io/event_dispatcher_epoll.c)
-  target_compile_definitions(fkvs-server PRIVATE SERVER)
+  find_library(LIBURING liburing)
+
+  if(LIBURING)
+    add_executable(fkvs-server src/memory.c src/counter.c src/client.c src/core/list.c src/config.c src/networking/networking.c src/server.c src/core/hashtable.c src/commands/common/command_registry.c src/commands/server/server_command_handlers.c src/io/event_dispatcher_io_uring.c)
+    target_link_libraries(fkvs-server PRIVATE LIBURING)
+    target_compile_definitions(fkvs-server PRIVATE SERVER IO_URING_ENABLED)
+  else()
+    add_executable(fkvs-server src/memory.c src/counter.c src/client.c src/core/list.c src/config.c src/networking/networking.c src/server.c src/core/hashtable.c src/commands/common/command_registry.c src/commands/server/server_command_handlers.c src/io/event_dispatcher_epoll.c)
+    target_compile_definitions(fkvs-server PRIVATE SERVER)
+  endif()
 
   add_executable(fkvs-benchmark src/fkvs-benchmark.c src/client.c src/core/list.c src/networking/networking.c src/commands/common/command_parser.c src/commands/client/client_command_handlers.c)
   target_compile_definitions(fkvs-benchmark PRIVATE CLI)

Dockerfile

@@ -19,7 +19,7 @@ COPY . .
 RUN cmake -S . -B build -DCMAKE_BUILD_TYPE=Release; \
     cmake --build build -j;
 
-FROM debian:stable-slim AS runtime
+FROM ubuntu:24.04 AS runtime
 ENV DEBIAN_FRONTEND=noninteractive
 
 RUN apt-get update; \
@@ -29,18 +29,23 @@ RUN apt-get update; \
 ARG UID=10001
 ARG GID=10001
 RUN groupadd -g "${GID}" fkvs && useradd -u "${UID}" -g "${GID}" -m -s /usr/sbin/nologin fkvs
+RUN mkdir -p /var/run/fkvs && \
+    chmod 777 /var/run/fkvs && \
+    chown fkvs:fkvs /var/run/fkvs
+USER fkvs
 
 ENV PATH="/usr/local/bin:${PATH}"
 
 COPY --from=builder /src/logo.txt /usr/local/bin/logo.txt
 COPY --from=builder /src/server.conf /home/fkvs/server.conf
 COPY --from=builder /src/client.conf /home/fkvs/client.conf
 COPY --from=builder /src/build/fkvs-server /usr/local/bin/fkvs-server
+COPY --from=builder /src/build/fkvs-benchmark /usr/local/bin/fkvs-benchmark
 COPY --from=builder /src/build/fkvs-cli    /usr/local/bin/fkvs-cli
 
 EXPOSE 5995
 HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \
-  CMD nc -z 127.0.0.1 5995 || exit 1
+    CMD /usr/local/bin/fkvs-cli -h 127.0.0.1 -p 5995 --non-interactive | grep 'PONG' || exit 1
 
 USER fkvs
 WORKDIR /app

README.md

@@ -27,7 +27,15 @@ principles.
 ```shell
 docker build -t fkvs:latest -f Dockerfile .  
 
-docker run --rm -it -p 5995:5995 fkvs:latest
+docker run --rm -it --name fkvs -p 5995:5995 fkvs:latest # if you intend to connecting from host via tcp
+docker run --rm -it --name fkvs fkvs:latest
+
+## Connect to server using 127.0.0.1 and port 5995
+docker exec -it fkvs /usr/local/bin/fkvs-cli -h 127.0.0.1 -p 5995 -d /home/fkvs/client.conf
+
+## Additional commands for running benchmarks from within the container
+docker exec -it fkvs /usr/local/bin/fkvs-benchmark -n 1000000 -t set -c 30 -u # run benchmark using unix domain sockets
+docker exec -it fkvs /usr/local/bin/fkvs-benchmark -n 1000000 -t set -c 30 # run benchmark using tcp
 ```
 
 ## Build fkvs from source
@@ -72,6 +80,13 @@ Type 'exit' to quit.
 "google.com"
 ```
 
+### Running the client in non-interactive mode
+
+```shell
+$ ./fkvs-cli -h 127.0.0.1 -p 5995 --non-interactive
+"PONG"
+```
+
 ### Supported commands
 
 - PING | PING "value"
@@ -155,6 +170,7 @@ verbose false
 daemonize false
 show-logo true
 unixsocket /tmp/fkvs.sock
+use-io-uring false
 ```
 
 Start the server:
@@ -177,6 +193,44 @@ Completed: 1000000  Failed: 0
 Elapsed: 3.263411 s  Throughput: 306427.84 req/s
 ```
 
+## Event Dispatchers Overview
+
+fkvs supports multiple event dispatchers, leveraging the available I/O technologies available on
+the host operating system for the best possible I/O performance.
+
+Below, we review the different type `event_dispatcher_kind`'s that fkvs supports and their characteristics.
+
+### Reactive-based Dispatchers
+These are readiness-driven mechanisms: the kernel notifies the fkvs process when a file descriptor 
+becomes ready for I/O (readable or writable).
+
+The server then performs the actual I/O operation by calling the necessary handler.
+
+kqueue:	(macOS -> BSD-style event notification system. Efficient for large numbers of connections)
+
+epoll:	(Linux -> Edge-triggered or level-triggered I/O multiplexer optimized for high concurrency)
+
+In both cases, fkvs waits for readiness notifications and then executes the corresponding read/write 
+operations, meaning a syscall still happens per I/O.
+
+### Pro-reactive-based Dispatcher
+This is an operation-driven mechanism: The fkvs server process submits actual I/O requests to the kernel, and 
+the kernel completes them asynchronously, returning results later via a completion queue.
+
+io_uring: (Linux kernel ≥ 5.6, Enables asynchronous, batched, and zero-syscall I/O via shared ring buffers between user
+and kernel space. Provides significantly higher throughput for small, frequent operations)
+
+Using `io_uring` generally leads to fewer syscalls (batch submission), lower context-switch overhead, 
+improved cache locality, better CPU efficiency, lower tail latency under load.
+
+### Configuration
+To enable `io_uring` on Linux, set the following option in your server.conf file:
+
+```server.conf
+# Enable io_uring for pro-reactive I/O handling on Linux
+use-io-uring true
+```
+
 ## Contributing
 
 Contributions are welcome! Please feel free to submit a Pull Request.

client.conf

@@ -6,4 +6,4 @@ port 5995
 # server the client should connect to
 bind 127.0.0.1
 # Specify the path for the Unix socket
-# unixsocket /tmp/fkvs.sock
+# unixsocket /var/run/fkvs/fkvs.sock

server.conf

@@ -1,10 +1,12 @@
 # Server configuration
-port 5995
+# port 5995
 logs-enabled true
 verbose false
 daemonize false
 show-logo true
 # The event-loop-max-events  configuration defines the maximum number of events that
 # can be processed at one time during an iteration of the event loop
 event-loop-max-events 100000
-# unixsocket /tmp/fkvs.sock
+unixsocket /var/run/fkvs/fkvs.sock
+# Enable io_uring for pro-reactive I/O handling on Linux
+use-io-uring true

src/client.h

@@ -22,6 +22,9 @@ typedef struct client_t {
     enum socket_domain
         socket_domain; // The socket domain we are using (Unix Domain or TCP/IP)
     bool benchmark_mode;
+    char *command_type;
+    char *config_file_path;
+    bool interactive_mode;
     bool verbose; // print additional information during runtime
 } client_t;
 

src/commands/client/client_command_handlers.c

@@ -4,6 +4,8 @@
 #include "../../client.h"
 #include "../../commands/common/command_parser.h"
 #include "../../commands/common/command_defs.h"
+
+#include <assert.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -33,6 +35,10 @@ void cmd_get(const command_args_t args, void (*response_cb)(client_t *client))
         return;
     }
 
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+
     send(args.client->fd, binary_cmd, cmd_len, 0);
     free(binary_cmd);
     response_cb(args.client);
@@ -61,6 +67,10 @@ void cmd_set(const command_args_t args, void (*response_cb)(client_t *client))
         return;
     }
 
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+
     send(args.client->fd, binary_cmd, cmd_len, 0);
     free(binary_cmd);
     response_cb(args.client);
@@ -88,6 +98,10 @@ void cmd_incr(const command_args_t args, void (*response_cb)(client_t *client))
         return;
     }
 
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+
     send(args.client->fd, binary_cmd, cmd_len, 0);
     free(binary_cmd);
     response_cb(args.client);
@@ -117,23 +131,33 @@ void cmd_incr_by(const command_args_t args,
         return;
     }
 
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+
     send(args.client->fd, binary_cmd, cmd_len, 0);
     free(binary_cmd);
     response_cb(args.client);
 }
 
 void cmd_ping(const command_args_t args, void (*response_cb)(client_t *client))
 {
-    if (strncmp(args.cmd, "PING ", 5) != 0) {
+    if (strncmp(args.cmd, "PING ", 5) != 0 && strncmp(args.cmd, "PING", 4) != 0) {
         return;
     }
 
     size_t cmd_len;
+
     char value[VALUE_LEN];
     if (sscanf(args.cmd, "PING \"%127[^\"]\"s", value) == 1) {
 
-    } else if (sscanf(args.cmd, "PING") == 1) {
-
+    } else if (strcmp(args.cmd, "PING") == 0) {
+      // TODO: Find a cleaner alternative for this.
+      value[0] = 'P';
+      value[1] = 'O';
+      value[2] = 'N';
+      value[3] = 'G';
+      value[4] = '\0';
     } else if (sscanf(args.cmd, "PING %127s", value) == 1) {
 
     } else {
@@ -142,12 +166,17 @@ void cmd_ping(const command_args_t args, void (*response_cb)(client_t *client))
         return;
     }
 
+    assert(ARRAY_SIZE(value) == VALUE_LEN);
     unsigned char *binary_cmd = construct_ping_command(value, &cmd_len);
     if (binary_cmd == NULL) {
         fprintf(stderr, "Failed to construct PING command\n");
         return;
     }
 
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+
     send(args.client->fd, binary_cmd, cmd_len, 0);
     free(binary_cmd);
     response_cb(args.client);
@@ -160,15 +189,19 @@ void cmd_info(const command_args_t args, void (*response_cb)(client_t *client))
     }
 
     size_t cmd_len;
-    unsigned char *info_cmd = construct_info_command(&cmd_len);
-    if (!info_cmd) {
+    unsigned char *binary_cmd = construct_info_command(&cmd_len);
+    if (!binary_cmd) {
         fprintf(stderr, "Failed to construct INFO command\n");
         return;
     }
-    send(args.client->fd, info_cmd, cmd_len, 0);
-    free(info_cmd);
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+    send(args.client->fd, binary_cmd, cmd_len, 0);
+    free(binary_cmd);
     response_cb(args.client);
 }
+
 void cmd_decr(const command_args_t args, void (*response_cb)(client_t *client))
 {
     if (strncmp(args.cmd, "DECR ", 4) != 0) {
@@ -191,6 +224,10 @@ void cmd_decr(const command_args_t args, void (*response_cb)(client_t *client))
         return;
     }
 
+    assert(cmd_len != 0);
+    assert(args.client->fd != -1);
+    assert(binary_cmd != NULL);
+
     send(args.client->fd, binary_cmd, cmd_len, 0);
     free(binary_cmd);
     response_cb(args.client);
@@ -206,7 +243,8 @@ void cmd_unknown(const command_args_t args,
 {
     if (strncmp(args.cmd, "INCR ", 5) && strncmp(args.cmd, "INCRBY ", 6) &&
         strncmp(args.cmd, "GET ", 4) && strncmp(args.cmd, "SET ", 4) &&
-        strncmp(args.cmd, "PING ", 5) && strncmp(args.cmd, "DECR ", 5) && strncmp(args.cmd, "INFO", 5)) {
+        strncmp(args.cmd, "PING ", 5) && strncmp(args.cmd, "PING", 4)
+        && strncmp(args.cmd, "DECR ", 5) && strncmp(args.cmd, "INFO", 5)) {
         printf("Unknown command \n");
     }
 }
@@ -258,7 +296,6 @@ void execute_command_benchmark(const char *cmd, client_t *client,
         response_cb(args.client);
     } else if (!strcasecmp(cmd, "ping")) {
         unsigned char *binary_cmd = construct_ping_command("", &cmd_len);
-
         send(args.client->fd, binary_cmd, cmd_len, 0);
         free(binary_cmd);
         response_cb(args.client);

src/commands/server/server_command_handlers.c

@@ -4,6 +4,8 @@
 #include "../../utils.h"
 #include "../common/command_defs.h"
 #include "../common/command_registry.h"
+
+#include <assert.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/errno.h>
@@ -134,10 +136,12 @@ void handle_get_command(int client_fd, unsigned char *buffer, size_t bytes_read)
             }
 
             memcpy(resp_buffer, value->ptr, value_len);
+
+            assert(resp_buffer != NULL);
+            assert(client_fd != -1);
+
             resp_buffer[value_len] = '\0';
-            printf("Returning: %s \n", (const char *)resp_buffer);
             send_reply(client_fd, resp_buffer, value_len);
-            free(resp_buffer);
         } else {
             send_error(client_fd);
         }
@@ -153,6 +157,9 @@ void handle_incr_command(int client_fd, unsigned char *buffer,
     const size_t command_len = (buffer[0] << 8) | buffer[1];
     const size_t key_len = (buffer[3] << 8) | buffer[4];
 
+    assert(key_len >= 1);
+    assert(command_len >= 1);
+
     if (bytes_read - 2 != command_len) {
         fprintf(stderr, "Incomplete command data for INCR.\n");
         send_error(client_fd);
@@ -176,6 +183,7 @@ void handle_incr_command(int client_fd, unsigned char *buffer,
 
     const uint64_t current = strtoull(value->ptr, NULL, 10);
     const uint64_t sum = current + 1;
+    assert(sum == current + 1);
 
     if (server.verbose) {
         printf("Value incremented to %llu\n", sum);

src/config.c

@@ -92,6 +92,16 @@ server_t load_server_config(const char *path)
                 }
             }
         }
+
+        if (strcmp(key, "use-io-uring") == 0) {
+            if (strcmp(value, "true") == 0) {
+                server.use_io_uring = true;
+            } else if (strcmp(value, "false") == 0) {
+                server.use_io_uring = false;
+            } else {
+                ERROR_AND_EXIT("'use-io-uring' expects a truthy value.");
+            }
+        }
     }
 
     fclose(config);
@@ -114,9 +124,15 @@ client_t load_client_config(const char *path)
     client.benchmark_mode = false;
     client.uds_socket_path = NULL;
     client.socket_domain = TCP_IP;
+    client.interactive_mode = true;
     if (client.socket_domain == TCP_IP) {
         client.ip_address = "127.0.0.1";
     }
+    if (path) {
+        client.config_file_path = path;
+    } else {
+        client.config_file_path = DEFAULT_CLIENT_CONFIG_FILE_PATH;
+    }
 
     char line[1024];