Porting IO Ring to linux by implementing io_uring

.gitignore

@@ -7,3 +7,5 @@ Binaries/file_hashes.txt
 file_list.txt
 temp_code.c
 /.cache/clangd/index
+/file_hasher
+/io_uring_test

README.md

@@ -18,9 +18,9 @@ gcc -g -O0 file_hasher.c xxh_x86dispatch.c -o file_hasher
 
 ### Linux:  
 #### Release:  
-clang -O3 -pthread file_hasher.c xxh_x86dispatch.c -o file_hasher  
+clang -O3 file_hasher.c xxhash.c xxh_x86dispatch.c -pthread -luring -o file_hasher  
-gcc -O3 -pthread file_hasher.c xxh_x86dispatch.c -o file_hasher  
+gcc -O3 file_hasher.c xxhash.c xxh_x86dispatch.c -pthread -luring -o file_hasher  
 
 #### Debug:  
-clang -g -O0 -pthread file_hasher.c xxh_x86dispatch.c -o file_hasher  
+clang -g -O0 file_hasher.c xxhash.c xxh_x86dispatch.c -pthread -luring -o file_hasher  
-gcc -g -O0 -pthread file_hasher.c xxh_x86dispatch.c -o file_hasher  
+gcc -g -O0 file_hasher.c xxhash.c xxh_x86dispatch.c -pthread -luring -o file_hasher  

base.h

@@ -1,23 +1,11 @@
 #pragma once
 #define _CRT_SECURE_NO_WARNINGS
 
-#include <assert.h>
-#include <immintrin.h>
-#include <stdatomic.h>
-#include <stdbool.h>
-#include <stddef.h>
-#include <stdint.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <time.h>
-
 #if defined(_WIN32) || defined(_WIN64)
-// #define PLATFORM_WINDOWS 1
+
-// #define WIN32_LEAN_AND_MEAN
+#if defined(_MSC_VER)
-// #define NTDDI_VERSION NTDDI_WIN11
+#pragma comment(lib, "advapi32.lib")
-//
+#endif
-// #pragma comment(lib, "kernel32.Lib")
 
 #include <aclapi.h>
 #include <fcntl.h>
@@ -29,20 +17,36 @@
 #include <windows.h>
 #include <winerror.h>
 
-#if defined(_MSC_VER)
+#elif defined(__linux__)
-#pragma comment(lib, "advapi32.lib")
+
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
 #endif
 
-#define strdup _strdup
-#else
 #include <dirent.h>
 #include <fcntl.h>
+#include <liburing.h>
 #include <pthread.h>
 #include <pwd.h>
+#include <sys/resource.h>
 #include <sys/stat.h>
 #include <unistd.h>
+#include <poll.h>
+#include <sys/eventfd.h>
 #endif
 
+#include <assert.h>
+#include <ctype.h>
+#include <immintrin.h>
+#include <stdatomic.h>
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+
 /* ------------------------------------------------------------
    Base types
    ------------------------------------------------------------ */

binaries/changelog.txt

@@ -50,7 +50,7 @@ Fixing user prompt parsing
 Reorganising the code
 Improving the scan function
 
-5.0: Implementing the IO Ring instead of buffered hashing, huge performance gains. The IO Ring is thread local, uses DMA and direct disk I/O, bypassing the OS cash completely, it supports bashing multiple submissions and can handle multiple files at the same time.
+5.0: Implementing the IO Ring for windows and ui_uring for linux instead of buffered hashing, huge performance gains. The IO Ring is event driven, thread local, uses DMA and direct disk I/O, bypassing the OS cash completely, registred buffers, it supports bashing multiple submissions and can handle multiple files at the same time.
 Hashing small files using XXH3_128bits() instead of the streaming pipeline(XXH3_128bits_reset(), XXH3_128bits_update(), XXH3_128bits_digest()), this reduses the overhead of creating a state and digest, coupled with the IO Ring it improves the hashing of small files whose size is inferior to the size of IO Ring buffers
 fixing the xxh_x86dispatch warnings
 Updating the progress printing function

file_hasher.c

@@ -87,7 +87,7 @@ int main(int argc, char **argv) {
   printf("  Selected instruction set: %s\n", get_xxhash_instruction_set());
 
   // Align IO Ring block size to the system page size
-  g_ioring_buffer_size = ALIGN_UP_POW2(IORING_BUFFER_SIZE, g_pagesize);
+  g_ioring_buffer_size = ALIGN_UP_POW2(g_ioring_buffer_size, g_pagesize);
   // -------------------------------
   // Scanning and hashing
   // -------------------------------
@@ -253,7 +253,7 @@ int main(int argc, char **argv) {
 
   FILE *f = fopen(FILE_HASHES_TXT, "wb");
 
-  for (int i = 0; i < num_threads; i++) {
+  for (int i = 0; i < num_hash_threads; i++) {
     mem_arena *arena = workers[i].arena;
     u8 *arena_base =
         (u8 *)arena + ALIGN_UP_POW2(sizeof(mem_arena), arena->align);
@@ -265,14 +265,13 @@ int main(int argc, char **argv) {
   // -------------------------------
   // Print summary
   // -------------------------------
-  // DEBUG
   uint64_t incomplete = atomic_load(&g_io_ring_fallbacks);
   if (incomplete > 0) {
-    printf(
+    printf("\nWARNING: I/O Ring incomplete files: %llu (fallback to buffered "
-        "\nI/O Ring incomplete files: %llu (fallback to buffered I/O used)\n",
+           "I/O used)\n",
            (unsigned long long)incomplete);
   }
-  //
+
   double total_seconds = timer_elapsed(&total_timer);
 
   printf("Completed hashing %zu files\n", total_found);

io_uring_test.c

@@ -0,0 +1,454 @@
+/*
+# Compile
+gcc -o io_uring_test io_uring_test.c -luring
+
+# Run
+./io_uring_test
+   */
+#include "base.h"
+#include <stdint.h>
+#define _GNU_SOURCE
+#include <errno.h>
+#include <fcntl.h>
+#include <liburing.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#define TEST_FILE "test_io_uring.txt"
+#define BUFFER_SIZE 4096
+#define NUM_BUFFERS 4
+
+// Colors for output
+#define COLOR_GREEN "\033[0;32m"
+#define COLOR_RED "\033[0;31m"
+#define COLOR_YELLOW "\033[0;33m"
+#define COLOR_BLUE "\033[0;34m"
+#define COLOR_RESET "\033[0m"
+
+// Test result tracking
+typedef struct {
+  int passed;
+  int failed;
+} TestResults;
+
+static void print_success(const char *step) {
+  printf(COLOR_GREEN "[✓] SUCCESS: %s" COLOR_RESET "\n", step);
+}
+
+static void print_failure(const char *step, const char *error) {
+  printf(COLOR_RED "[✗] FAILED: %s - %s" COLOR_RESET "\n", step, error);
+}
+
+static void print_info(const char *msg) {
+  printf(COLOR_BLUE "[i] INFO: %s" COLOR_RESET "\n", msg);
+}
+
+static void print_step(const char *step) {
+  printf(COLOR_YELLOW "\n>>> Testing: %s" COLOR_RESET "\n", step);
+}
+
+// Create a test file with known content
+static int create_test_file(void) {
+  const char *test_content =
+      "Hello, io_uring! This is a test file for async I/O operations.\n"
+      "Line 2: Testing reads with registered buffers.\n"
+      "Line 3: The quick brown fox jumps over the lazy dog.\n"
+      "Line 4: ABCDEFGHIJKLMNOPQRSTUVWXYZ\n"
+      "Line 5: 0123456789\n";
+
+  FILE *f = fopen(TEST_FILE, "w");
+  if (!f) {
+    perror("Failed to create test file");
+    return -1;
+  }
+
+  fprintf(f, "%s", test_content);
+  fclose(f);
+
+  print_info("Test file created successfully");
+  return 0;
+}
+
+// Test 1: Create io_uring instance
+static int test_io_uring_create(struct io_uring *ring, TestResults *results) {
+  print_step("io_uring creation");
+
+  int ret = io_uring_queue_init(256, ring, 0);
+  if (ret < 0) {
+    print_failure("io_uring_queue_init", strerror(-ret));
+    results->failed++;
+    return -1;
+  }
+
+  print_success("io_uring instance created");
+  results->passed++;
+  return 0;
+}
+
+// Test 2: Register buffers
+static int test_register_buffers(struct io_uring *ring, void **buffers,
+                                 struct iovec *iovs, TestResults *results) {
+  print_step("Buffer registration");
+
+  // Allocate and prepare buffers
+  size_t total_size = BUFFER_SIZE * NUM_BUFFERS;
+  *buffers = aligned_alloc(4096, total_size); // Page-aligned for O_DIRECT
+  if (!*buffers) {
+    print_failure("Buffer allocation", strerror(errno));
+    results->failed++;
+    return -1;
+  }
+
+  // Initialize iovecs
+  for (int i = 0; i < NUM_BUFFERS; i++) {
+    iovs[i].iov_base = (char *)*buffers + (i * BUFFER_SIZE);
+    iovs[i].iov_len = BUFFER_SIZE;
+    memset(iovs[i].iov_base, 0, BUFFER_SIZE);
+  }
+
+  int ret = io_uring_register_buffers(ring, iovs, NUM_BUFFERS);
+  if (ret < 0) {
+    print_failure("io_uring_register_buffers", strerror(-ret));
+    results->failed++;
+    return -1;
+  }
+
+  print_success("Buffers registered successfully");
+  results->passed++;
+  return 0;
+}
+
+// Test 3: Open file
+// Modified test_open_file function
+static int test_open_file(int *fd, TestResults *results) {
+  print_step("File opening");
+
+  // Get file size
+  struct stat st;
+  if (stat(TEST_FILE, &st) != 0) {
+    print_failure("stat", strerror(errno));
+    results->failed++;
+    return -1;
+  }
+
+  // Check if file size is page-aligned
+  int page_size = plat_get_pagesize();
+  size_t file_size = st.st_size;
+
+  printf("  File size: %zu bytes\n", file_size);
+  printf("  Page size: %d bytes\n", page_size);
+
+  if (file_size % page_size != 0) {
+    printf("  Extending read size from %zu to %zu bytes\n", file_size,
+           ALIGN_UP_POW2(file_size, page_size));
+  }
+
+  // Try O_DIRECT first
+  *fd = open(TEST_FILE, O_RDONLY | O_DIRECT);
+  if (*fd < 0) {
+    print_info("O_DIRECT failed, trying without it");
+    *fd = open(TEST_FILE, O_RDONLY);
+    if (*fd < 0) {
+      print_failure("open", strerror(errno));
+      results->failed++;
+      return -1;
+    }
+    print_info("Using buffered I/O (O_DIRECT not available)");
+  } else {
+    print_success("File opened with O_DIRECT");
+  }
+
+  results->passed++;
+  return 0;
+}
+
+// Test 4: Build and submit read operation
+static int test_submit_read(struct io_uring *ring, int fd, struct iovec *iovs,
+                            int buffer_id, uint64_t user_data,
+                            TestResults *results) {
+  print_step("Building and submitting read operation");
+
+  // Get file size for proper alignment
+  struct stat st;
+  if (fstat(fd, &st) != 0) {
+    print_failure("fstat", strerror(errno));
+    results->failed++;
+    return -1;
+  }
+
+  u32 page_size = plat_get_pagesize();
+  size_t file_size = st.st_size;
+  size_t read_size = BUFFER_SIZE;
+
+  // For O_DIRECT, ensure read size is sector-aligned
+  if (read_size > file_size) {
+    read_size = ALIGN_UP_POW2(file_size, page_size);
+    printf("  Adjusted read size to %zu bytes for O_DIRECT alignment\n",
+           read_size);
+  }
+
+  struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
+  if (!sqe) {
+    print_failure("io_uring_get_sqe", "No available SQE");
+    results->failed++;
+    return -1;
+  }
+
+  // Prepare read operation using registered buffer
+  io_uring_prep_read_fixed(sqe, fd, iovs[buffer_id].iov_base, read_size, 0,
+                           buffer_id);
+  io_uring_sqe_set_data64(sqe, user_data);
+
+  int ret = io_uring_submit(ring);
+  if (ret < 0) {
+    print_failure("io_uring_submit", strerror(-ret));
+    results->failed++;
+    return -1;
+  }
+
+  print_success("Read operation submitted successfully");
+  results->passed++;
+  return 0;
+}
+
+// Test 5: Wait for completion
+static int test_wait_completion(struct io_uring *ring,
+                                struct io_uring_cqe **cqe,
+                                TestResults *results) {
+  print_step("Waiting for completion");
+
+  int ret = io_uring_wait_cqe(ring, cqe);
+  if (ret < 0) {
+    print_failure("io_uring_wait_cqe", strerror(-ret));
+    results->failed++;
+    return -1;
+  }
+
+  print_success("Completion received");
+  results->passed++;
+  return 0;
+}
+
+// Test 6: Process completion
+static int test_process_completion(struct io_uring_cqe *cqe,
+                                   uint64_t expected_user_data,
+                                   TestResults *results) {
+  print_step("Processing completion");
+
+  uint64_t user_data = io_uring_cqe_get_data64(cqe);
+  int res = cqe->res;
+
+  printf("  Completion data:\n");
+  printf("    User data: %lu (expected: %lu)\n", user_data, expected_user_data);
+  printf("    Result: %d bytes read\n", res);
+
+  if (user_data != expected_user_data) {
+    print_failure("User data mismatch",
+                  "User data doesn't match expected value");
+    results->failed++;
+    return -1;
+  }
+
+  if (res < 0) {
+    print_failure("Read operation", strerror(-res));
+    results->failed++;
+    return -1;
+  }
+
+  print_success("Completion processed successfully");
+  results->passed++;
+  return res; // Return number of bytes read
+}
+
+// Test 7: Verify read data
+static int test_verify_data(struct iovec *iovs, int buffer_id, int bytes_read,
+                            TestResults *results) {
+  print_step("Data verification");
+
+  char *data = (char *)iovs[buffer_id].iov_base;
+
+  printf("  Read data (first 200 chars):\n");
+  printf("  ---\n");
+  for (int i = 0; i < bytes_read && i < 200; i++) {
+    putchar(data[i]);
+  }
+  if (bytes_read > 200)
+    printf("...");
+  printf("\n  ---\n");
+
+  // Check if data is not empty
+  if (bytes_read == 0) {
+    print_failure("Data verification", "No data read");
+    results->failed++;
+    return -1;
+  }
+
+  // Check if data contains expected content
+  if (strstr(data, "io_uring") == NULL) {
+    print_failure("Data verification", "Expected content not found");
+    results->failed++;
+    return -1;
+  }
+
+  print_success("Data verified successfully");
+  results->passed++;
+  return 0;
+}
+
+// Test 8: Test multiple concurrent reads
+static int test_concurrent_reads(struct io_uring *ring, int fd,
+                                 struct iovec *iovs, TestResults *results) {
+  print_step("Concurrent reads test");
+
+  int num_reads = 3;
+  int submitted = 0;
+
+  // Submit multiple reads
+  for (int i = 0; i < num_reads; i++) {
+    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
+    if (!sqe) {
+      print_failure("Getting SQE for concurrent read", "No available SQE");
+      results->failed++;
+      return -1;
+    }
+
+    off_t offset = i * 100; // Read from different offsets
+    io_uring_prep_read_fixed(sqe, fd, iovs[i].iov_base, BUFFER_SIZE, offset, i);
+    io_uring_sqe_set_data64(sqe, i);
+    submitted++;
+  }
+
+  int ret = io_uring_submit(ring);
+  if (ret != submitted) {
+    char msg[64];
+    snprintf(msg, sizeof(msg), "Expected %d, got %d", submitted, ret);
+
+    print_failure("Submitting concurrent reads", msg);
+    results->failed++;
+    return -1;
+  }
+
+  print_success("Concurrent reads submitted");
+
+  // Wait for and process completions
+  for (int i = 0; i < submitted; i++) {
+    struct io_uring_cqe *cqe;
+    ret = io_uring_wait_cqe(ring, &cqe);
+    if (ret < 0) {
+      print_failure("Waiting for concurrent read completion", strerror(-ret));
+      results->failed++;
+      return -1;
+    }
+
+    uint64_t user_data = io_uring_cqe_get_data64(cqe);
+    int res = cqe->res;
+
+    printf("  Concurrent read %lu completed: %d bytes read\n", user_data, res);
+    io_uring_cqe_seen(ring, cqe);
+  }
+
+  print_success("Concurrent reads completed successfully");
+  results->passed++;
+  return 0;
+}
+
+// Cleanup function
+static void cleanup(struct io_uring *ring, int fd, void *buffers) {
+  if (fd >= 0)
+    close(fd);
+  if (buffers) {
+    io_uring_unregister_buffers(ring);
+    free(buffers);
+  }
+  io_uring_queue_exit(ring);
+  remove(TEST_FILE);
+}
+
+int main() {
+  TestResults results = {0, 0};
+  struct io_uring ring;
+  int fd = -1;
+  void *buffers = NULL;
+  struct iovec iovs[NUM_BUFFERS];
+
+  printf(COLOR_BLUE "\n========================================\n");
+  printf("    io_uring Test Suite\n");
+  printf("========================================\n" COLOR_RESET);
+
+  // Create test file
+  if (create_test_file() != 0) {
+    return 1;
+  }
+
+  // Test 1: Create io_uring
+  if (test_io_uring_create(&ring, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Test 2: Register buffers
+  if (test_register_buffers(&ring, &buffers, iovs, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Test 3: Open file
+  if (test_open_file(&fd, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Test 4: Submit read
+  uint64_t test_user_data = 12345;
+  if (test_submit_read(&ring, fd, iovs, 0, test_user_data, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Test 5: Wait for completion
+  struct io_uring_cqe *cqe;
+  if (test_wait_completion(&ring, &cqe, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Test 6: Process completion
+  int bytes_read = test_process_completion(cqe, test_user_data, &results);
+  if (bytes_read < 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+  io_uring_cqe_seen(&ring, cqe);
+
+  // Test 7: Verify data
+  if (test_verify_data(iovs, 0, bytes_read, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Test 8: Concurrent reads
+  if (test_concurrent_reads(&ring, fd, iovs, &results) != 0) {
+    cleanup(&ring, fd, buffers);
+    return 1;
+  }
+
+  // Print summary
+  printf(COLOR_BLUE "\n========================================\n");
+  printf("           TEST SUMMARY\n");
+  printf("========================================\n" COLOR_RESET);
+  printf("  Total tests: %d\n", results.passed + results.failed);
+  printf(COLOR_GREEN "  Passed: %d\n" COLOR_RESET, results.passed);
+  if (results.failed > 0) {
+    printf(COLOR_RED "  Failed: %d\n" COLOR_RESET, results.failed);
+  } else {
+    printf(COLOR_GREEN "  ✓ ALL TESTS PASSED!\n" COLOR_RESET);
+  }
+
+  // Cleanup
+  cleanup(&ring, fd, buffers);
+
+  return results.failed > 0 ? 1 : 0;
+}