amir/duplicatefinder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f37e9154895c2a27326ff77683dff7b7fa6637ce
duplicatefinder/platform.c
amir f37e915489 Minor optimisations and bug fixes 
Fix bug in mt_mpmc.c, in Linux mutexes are not recursive.
Add arena_trim_string() to the arena API
Removing arena->path, now paths are pushed to arena->metadata
Replacing fe->owner[128] with char *owner; the owner is not pushed as a
string to arena->metadata and trimed with arena_trim_string()
Improving cache locality in arena->metadata, the memory layout is not
fe; fe->path; fe->owner.
Cache aligning all arenas except HasherContext->arena to sizeof(void *).
Pushing elements one by one instead of snprintf() in finalize_file() and
hash_worker().
Getting the full path of current directory instead of "."
Fixing bug in path formating, this allow us to remove normalize_path()
from the hot loop.
2026-05-08 20:06:48 +01:00

2151 lines
57 KiB
C
Raw Blame History

#pragma once
#include "arena.h"
#include "base.h"
#include "mt_mpmc.h"
#include "arena.c"
#include <stdint.h>
#include <stdio.h>
// xxhash include
#define XXH_STATIC_LINKING_ONLY
#include "xxh_x86dispatch.h"
#include "config.h"
// ----------------------------- Globals ------------------------------------
static atomic_uint_fast64_t g_files_found = 0;
static atomic_uint_fast64_t g_files_hashed = 0;
static atomic_uint_fast64_t g_bytes_processed = 0;
static atomic_int g_scan_done = 0;
#define HASH_STRLEN 33 // 128-bit hex (32 chars) + null terminator
#define MAX_PATHLEN KiB(4)
// ================== OS-agnostic functions abstraction =====================
// --------------------- Timer functions ---------------------
typedef struct {
u64 start;
u64 now;
} HiResTimer;
#if defined(_WIN32) || defined(_WIN64)
static LARGE_INTEGER g_freq;
static void timer_init(void) { QueryPerformanceFrequency(&g_freq); }
static void timer_start(HiResTimer *t) {
LARGE_INTEGER v;
QueryPerformanceCounter(&v);
t->start = v.QuadPart;
}
static double timer_elapsed(HiResTimer *t) {
LARGE_INTEGER v;
QueryPerformanceCounter(&v);
t->now = v.QuadPart;
return (double)(t->now - t->start) / (double)g_freq.QuadPart;
}
#elif defined(__linux__)
void timer_init(void) {}
void timer_start(HiResTimer *t) {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
t->start = ts.tv_sec * 1000000000ULL + ts.tv_nsec;
}
double timer_elapsed(HiResTimer *t) {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
uint64_t now = ts.tv_sec * 1000000000ULL + ts.tv_nsec;
return (double)(now - t->start) / 1e9;
}
#endif
// ------------------- Get HW info --------------------
#if defined(_WIN32) || defined(_WIN64)
size_t platform_physical_cores(void) {
DWORD len = 0;
GetLogicalProcessorInformation(NULL, &len);
SYSTEM_LOGICAL_PROCESSOR_INFORMATION buf[len];
GetLogicalProcessorInformation(buf, &len);
DWORD count = 0;
DWORD n = len / sizeof(SYSTEM_LOGICAL_PROCESSOR_INFORMATION);
for (DWORD i = 0; i < n; i++) {
if (buf[i].Relationship == RelationProcessorCore)
count++;
}
return count ? count : 1;
}
#elif defined(__linux__)
size_t platform_physical_cores(void) {
long n = sysconf(_SC_NPROCESSORS_ONLN);
return n > 0 ? (size_t)n : 1;
}
#endif
const char *get_xxhash_instruction_set(void) {
int vecID = XXH_featureTest();
switch (vecID) {
case XXH_SCALAR:
return "Scalar (portable C)";
case XXH_SSE2:
return "SSE2";
case XXH_AVX2:
return "AVX2";
case XXH_AVX512:
return "AVX-512";
default:
return "Unknown";
}
}
// -------------------- File IO -------------------
#if defined(_WIN32) || defined(_WIN64)
typedef HANDLE FileHandle;
#define FLAG_SEQUENTIAL_READ FILE_FLAG_SEQUENTIAL_SCAN
#define FLAG_ASYNC_DIRECT_READ (FILE_FLAG_OVERLAPPED | FILE_FLAG_NO_BUFFERING)
#define INVALID_FILE_HANDLE INVALID_HANDLE_VALUE
// File open function
static FileHandle os_file_open(const char *path, DWORD flags) {
return CreateFileA(path, GENERIC_READ, FILE_SHARE_READ | FILE_SHARE_WRITE,
NULL, OPEN_EXISTING, flags, NULL);
}
// File read function
static int os_file_read(FileHandle handle, void *buf, size_t count,
uint64_t *bytes_read) {
DWORD read = 0;
BOOL result = ReadFile(handle, buf, (DWORD)count, &read, NULL);
*bytes_read = read;
return (result && read > 0) ? 0 : -1;
}
// File close function
static void os_file_close(FileHandle handle) { CloseHandle(handle); }
#elif defined(__linux__)
typedef int FileHandle;
#define FLAG_SEQUENTIAL_READ (0)
#define FLAG_ASYNC_DIRECT_READ (O_DIRECT)
#define INVALID_FILE_HANDLE (-1)
// File open function
static FileHandle os_file_open(const char *path, int flags) {
// Combine your mandatory flags with the user-provided flag
int fd = open(path, O_RDONLY | O_NOFOLLOW | flags);
// If sequential was requested, advise the kernel
if (fd != -1 && (flags == FLAG_SEQUENTIAL_READ)) {
posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL);
}
return fd;
}
// File read function
static int os_file_read(FileHandle handle, void *buf, size_t count,
uint64_t *bytes_read) {
ssize_t result = read(handle, buf, count);
if (result >= 0) {
*bytes_read = (uint64_t)result;
return 0;
}
*bytes_read = 0;
return -1;
}
// File close function
static void os_file_close(FileHandle handle) { close(handle); }
#endif
// -------------------- Thread abstraction -------------------
// Threads context
typedef struct {
mem_arena *meta_arena;
MPMCQueue *dir_queue;
MPMCQueue *file_queue;
u8 num_threads;
} ScannerContext;
typedef struct {
mem_arena *arena;
MPMCQueue *file_queue;
} HasherContext;
#if defined(_WIN32) || defined(_WIN64)
typedef HANDLE ThreadHandle;
typedef DWORD(WINAPI *ThreadFunc)(void *);
#define THREAD_RETURN DWORD WINAPI
#define THREAD_RETURN_VALUE 0;
typedef struct {
ThreadHandle handle;
int valid; // Track if thread was successfully created
} Thread;
// Thread function wrapper to handle different return types
#define THREAD_FUNCTION(name) DWORD WINAPI name(LPVOID arg)
// Thread creation function
static int thread_create(Thread *thread, ThreadFunc func, void *arg) {
thread->handle =
CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)func, arg, 0, NULL);
return (thread->handle != NULL) ? 0 : -1;
}
// Thread join function
static int thread_join(Thread *thread) {
return (WaitForSingleObject(thread->handle, INFINITE) == WAIT_OBJECT_0) ? 0
: -1;
}
// Thread close/detach function
static void thread_close(Thread *thread) { CloseHandle(thread->handle); }
// Wait for multiple threads
static int thread_wait_multiple(Thread *threads, size_t count) {
HANDLE handles[64]; // Max 64 threads for Windows
for (size_t i = 0; i < count; i++) {
handles[i] = threads[i].handle;
}
return (WaitForMultipleObjects((DWORD)count, handles, TRUE, INFINITE) ==
WAIT_OBJECT_0)
? 0
: -1;
}
#elif defined(__linux__)
typedef pthread_t ThreadHandle;
typedef void *(*ThreadFunc)(void *);
#define THREAD_RETURN void *
#define THREAD_RETURN_VALUE NULL;
typedef struct {
ThreadHandle handle;
int valid; // Track if thread was successfully created
} Thread;
// Thread function wrapper to handle different return types
typedef struct {
void *(*func)(void *);
void *arg;
} ThreadWrapper;
// Thread creation function
static int thread_create(Thread *thread, ThreadFunc func, void *arg) {
int ret = pthread_create(&thread->handle, NULL, func, arg);
if (ret == 0) {
thread->valid = 1;
}
return ret;
}
// Thread join function
static int thread_join(Thread *thread) {
int ret = pthread_join(thread->handle, NULL);
thread->valid = 0;
return ret;
}
// Thread close/detach function
static void thread_close(Thread *thread) {
if (thread->valid) {
pthread_detach(thread->handle);
thread->valid = 0;
}
}
// Wait for multiple threads
static int thread_wait_multiple(Thread *threads, size_t count) {
for (size_t i = 0; i < count; i++) {
if (thread_join(&threads[i]) != 0) {
return -1;
}
}
return 0;
}
#endif
// ======================== Get file metadata ========================
// -------------------- Path helpers -------------------
static void normalize_path(char *p) {
char *src = p;
char *dst = p;
int prev_slash = 0;
while (*src) {
char c = *src++;
if (c == '\\' || c == '/') {
if (!prev_slash) {
*dst++ = '/';
prev_slash = 1;
}
} else {
*dst++ = c;
prev_slash = 0;
}
}
*dst = '\0';
}
static int parse_paths(char *line, char folders[][MAX_PATHLEN],
int max_folders) {
int count = 0;
char *p = line;
while (*p && count < max_folders) {
while (*p && isspace((unsigned char)*p))
p++;
if (!*p)
break;
char *start;
char quote = 0;
if (*p == '"' || *p == '\'') {
quote = *p++;
start = p;
while (*p && *p != quote)
p++;
} else {
start = p;
while (*p && !isspace((unsigned char)*p))
p++;
}
size_t len = p - start;
if (len >= MAX_PATHLEN)
len = MAX_PATHLEN - 1;
memcpy(folders[count], start, len);
folders[count][len] = 0;
normalize_path(folders[count]);
count++;
if (quote && *p == quote)
p++;
}
return count;
}
#if defined(_WIN32) || defined(_WIN64)
bool platform_get_current_directory(char *buffer, size_t size) {
DWORD len = GetCurrentDirectoryA((DWORD)size, buffer);
if (len == 0 || len >= size)
return false;
return true;
}
#elif defined(__linux__)
bool platform_get_current_directory(char *buffer, size_t size) {
return getcwd(buffer, size) != NULL;
}
#endif
typedef struct {
char buffer[MAX_PATHLEN];
char *base_end; // Points to end of base path
char *filename_pos; // Points to where filename should be written
size_t base_len;
} PathBuilder;
static void path_builder_init(PathBuilder *pb, const char *base) {
pb->base_len = strlen(base);
memcpy(pb->buffer, base, pb->base_len);
pb->base_end = pb->buffer + pb->base_len;
// Only add separator if not already present
if (pb->base_len > 0 && *(pb->base_end - 1) != '/') {
*pb->base_end = '/';
pb->base_end++;
pb->base_len++;
}
// Ensure null termination
*pb->base_end = '\0';
pb->filename_pos = pb->base_end;
}
static void path_builder_set_filename(PathBuilder *pb, const char *filename,
size_t name_len) {
memcpy(pb->filename_pos, filename, name_len);
pb->filename_pos[name_len] = '\0'; // Ensure null termination
}
static char *path_builder_dup_arena(PathBuilder *pb, mem_arena *arena,
bool zero) {
// Calculate total length including base + separator + filename + null
// terminator
size_t total_len =
(pb->filename_pos - pb->buffer) + strlen(pb->filename_pos) + 1;
char *dup = arena_push(&arena, total_len, zero);
memcpy(dup, pb->buffer, total_len);
return dup;
}
// ------------------------- File time -------------------------
#if FILE_TIMES
#if defined(_WIN32) || defined(_WIN64)
static void format_time(uint64_t t, char *out, size_t out_sz) {
if (t == 0) {
snprintf(out, out_sz, "N/A");
return;
}
time_t tt = (time_t)t;
struct tm tm;
localtime_s(&tm, &tt);
strftime(out, out_sz, "%Y-%m-%d %H:%M:%S", &tm);
}
// ------------------ Convert filetime to epoch -------------------
static uint64_t filetime_to_epoch(const FILETIME *ft) {
ULARGE_INTEGER ull;
ull.LowPart = ft->dwLowDateTime;
ull.HighPart = ft->dwHighDateTime;
// Windows epoch (1601) ￢ﾆﾒ Unix epoch (1970)
return (ull.QuadPart - 116444736000000000ULL) / 10000000ULL;
}
void platform_get_file_times(const char *path, uint64_t *out_created,
uint64_t *out_modified) {
WIN32_FILE_ATTRIBUTE_DATA fad;
if (GetFileAttributesExA(path, GetFileExInfoStandard, &fad)) {
*out_created = filetime_to_epoch(&fad.ftCreationTime);
*out_modified = filetime_to_epoch(&fad.ftLastWriteTime);
} else {
*out_created = 0;
*out_modified = 0;
}
}
#elif defined(__linux__)
static void format_time(uint64_t t, char *out, size_t out_sz) {
if (t == 0) {
snprintf(out, out_sz, "N/A");
return;
}
time_t tt = (time_t)t;
struct tm tm;
localtime_r(&tt, &tm);
strftime(out, out_sz, "%Y-%m-%d %H:%M:%S", &tm);
}
void platform_get_file_times(const char *path, uint64_t *out_created,
uint64_t *out_modified) {
struct stat st;
if (stat(path, &st) == 0) {
*out_created = (uint64_t)st.st_ctime;
*out_modified = (uint64_t)st.st_mtime;
} else {
*out_created = 0;
*out_modified = 0;
}
}
static int platform_get_file_times_fd(int dir_fd, const char *name,
uint64_t *created, uint64_t *modified) {
struct stat st;
if (fstatat(dir_fd, name, &st, 0) == 0) {
*created = st.st_ctime; // or st.st_birthtime on systems that support it
*modified = st.st_mtime;
return 0;
}
return -1;
}
#endif
#endif
// -------------------- File owner ---------------------
#if defined(_WIN32) || defined(_WIN64)
#if FILE_OWNER
void platform_get_file_owner(const char *path, char *out_owner,
size_t out_owner_size) {
PSID sid = NULL;
PSECURITY_DESCRIPTOR sd = NULL;
if (GetNamedSecurityInfoA(path, SE_FILE_OBJECT, OWNER_SECURITY_INFORMATION,
&sid, NULL, NULL, NULL, &sd) == ERROR_SUCCESS) {
char name[64], domain[64];
DWORD name_len = sizeof(name);
DWORD domain_len = sizeof(domain);
SID_NAME_USE use;
if (LookupAccountSidA(NULL, sid, name, &name_len, domain, &domain_len,
&use)) {
snprintf(out_owner, out_owner_size, "%s\\%s", domain, name);
} else {
snprintf(out_owner, out_owner_size, "UNKNOWN");
}
} else {
snprintf(out_owner, out_owner_size, "UNKNOWN");
}
if (sd)
LocalFree(sd);
}
#endif
#elif defined(__linux__)
#if FILE_OWNER
void platform_get_file_owner(const char *path, char *out_owner,
size_t out_owner_size) {
struct stat st;
const char *owner = "UNKNOWN";
if (stat(path, &st) == 0) {
struct passwd *pw = getpwuid(st.st_uid);
if (pw) {
owner = pw->pw_name;
}
}
snprintf(out_owner, out_owner_size, "%s", owner);
}
static int platform_get_file_owner_fd(int dir_fd, const char *name, char *owner,
size_t owner_size) {
struct stat st;
if (fstatat(dir_fd, name, &st, 0) == 0) {
struct passwd pw;
struct passwd *result;
char buffer[4096]; // Sufficiently large buffer for passwd data
// Reentrant version (thread-safe)
if (getpwuid_r(st.st_uid, &pw, buffer, sizeof(buffer), &result) == 0 &&
result != NULL && result->pw_name != NULL) {
strncpy(owner, result->pw_name, owner_size - 1);
owner[owner_size - 1] = '\0';
} else {
// Fallback to uid
snprintf(owner, owner_size, "uid:%d", st.st_uid);
}
return 0;
}
return -1;
}
#endif
// ----------------------------- File system -----------------------------
#if CHECK_FILE_SYSTEM
typedef enum FileSystemType {
FS_UNKNOWN = 0,
FS_EXT4,
FS_XFS,
FS_BTRFS,
FS_TMPFS,
FS_NFS,
FS_CIFS,
FS_FAT,
FS_EXFAT,
FS_NTFS,
FS_ZFS,
FS_F2FS,
FS_EROFS,
FS_VIRTIOFS,
FS_OVERLAY,
FS_HUGETLBFS,
FS_SQUASHFS,
FS_PROC,
FS_SYSFS,
} FileSystemType;
static inline FileSystemType fs_from_magic(long type) {
switch (type) {
case 0xEF53:
return FS_EXT4;
case 0x58465342:
return FS_XFS;
case 0x9123683E:
return FS_BTRFS;
case 0x01021994:
return FS_TMPFS;
case 0x6969:
return FS_NFS;
case 0xFF534D42:
return FS_CIFS;
case 0x4d44:
return FS_FAT;
case 0x2011BAB0:
return FS_EXFAT;
case 0x5346544E:
return FS_NTFS;
case 0x2FC1211:
return FS_ZFS;
case 0xF2F52010:
return FS_F2FS;
case 0xE0F5E1E2:
return FS_EROFS;
case 0x56495254:
return FS_VIRTIOFS;
case 0x794C764F:
return FS_OVERLAY;
case 0x958458f6:
return FS_HUGETLBFS;
case 0x73717368:
return FS_SQUASHFS;
case 0x9fa0:
return FS_PROC;
case 0x62656572:
return FS_SYSFS;
default:
return FS_UNKNOWN;
}
}
// Yes it is officially called "magic number" or "signature" in the
// documentation
typedef enum {
FS_POLICY_BUFFERED,
FS_POLICY_DIRECT_OK,
} FsPolicy;
static inline FsPolicy fs_policy(FileSystemType fs) {
switch (fs) {
case FS_EXT4:
case FS_XFS:
case FS_BTRFS:
case FS_ZFS:
case FS_F2FS:
case FS_NFS:
case FS_CIFS:
case FS_VIRTIOFS:
return FS_POLICY_DIRECT_OK;
case FS_TMPFS: // Resides in Page Cache; O_DIRECT generally returns EINVAL
case FS_EROFS: // Read-only filesystem; O_DIRECT support is
// uncommon/restricted
case FS_FAT: // Generally does not implement direct_IO address space ops
case FS_EXFAT: // Limited support depending on driver implementation
case FS_NTFS: // Depends on driver (ntfs3 supports it, older ntfs-3g does not)
default:
return FS_POLICY_BUFFERED;
}
}
#endif
#endif
// ----------------------------- Scan helpers -----------------------------
typedef struct FileEntry {
char *path;
uint64_t size_bytes;
#if FILE_TIMES
uint64_t created_time; // epoch
uint64_t modified_time; // epoch seconds
#endif
#if FILE_OWNER
char *owner; // resolved owner name
#endif
#if CHECK_FILE_SYSTEM // Linux only
FileSystemType fs_type;
#endif
} FileEntry;
#if defined(_WIN32) || defined(_WIN64)
void scan_folder(const char *base, ScannerContext *ctx) {
PathBuilder pb;
path_builder_init(&pb, base);
char search[MAX_PATHLEN];
memcpy(search, pb.buffer, pb.base_len + 1); // Copy base + separator
memcpy(search + pb.base_len, "*", 2); // Add "*" and null
WIN32_FIND_DATAA fd;
HANDLE h = FindFirstFileA(search, &fd);
if (h == INVALID_HANDLE_VALUE)
return;
do {
// Skip . and ..
if (fd.cFileName[0] == '.' &&
(fd.cFileName[1] == 0 ||
(fd.cFileName[1] == '.' && fd.cFileName[2] == 0)))
continue;
if (fd.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT)
continue;
size_t name_len = strlen(fd.cFileName);
path_builder_set_filename(&pb, fd.cFileName, name_len);
// If it's a directory:
if (fd.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) {
char *dir = path_builder_dup_arena(&pb, ctx->meta_arena, false);
mpmc_push_work(ctx->dir_queue, dir);
} else {
// else a file:
atomic_fetch_add(&g_files_found, 1);
FileEntry *fe = arena_push(&ctx->meta_arena, sizeof(FileEntry), false);
fe->path = arena_push(&ctx->meta_arena, MAX_PATHLEN, false);
memcpy(fe->path, pb.buffer, (pb.filename_pos - pb.buffer) + name_len + 1);
arena_trim_string(&ctx->meta_arena, fe->path, ARENA_TRIM_NUL);
#if FILE_TIMES
platform_get_file_times(pb.buffer, &fe->created_time, &fe->modified_time);
#endif
#if FILE_OWNER
fe->owner = arena_push(&ctx->meta_arena, MAX_PATHLEN, false);
platform_get_file_owner(pb.buffer, fe->owner, MAX_PATHLEN);
arena_trim_string(&ctx->meta_arena, fe->owner, ARENA_TRIM_NUL);
#endif
fe->size_bytes = ((uint64_t)fd.nFileSizeHigh << 32) | fd.nFileSizeLow;
mpmc_push(ctx->file_queue, fe);
}
} while (FindNextFileA(h, &fd));
FindClose(h);
}
#elif defined(__linux__)
void scan_folder(const char *base, ScannerContext *ctx) {
PathBuilder pb;
path_builder_init(&pb, base);
int dir_fd = open(base, O_RDONLY | O_DIRECTORY | O_NOFOLLOW);
if (dir_fd == -1)
return;
#if CHECK_FILE_SYSTEM
struct statfs fs;
FileSystemType fs_type = FS_UNKNOWN;
if (fstatfs(dir_fd, &fs) == 0) {
fs_type = fs_from_magic(fs.f_type);
}
#endif
DIR *dir = fdopendir(dir_fd);
if (!dir) {
close(dir_fd);
return;
}
struct dirent *entry;
while ((entry = readdir(dir)) != NULL) {
if (entry->d_name[0] == '.' &&
(entry->d_name[1] == 0 ||
(entry->d_name[1] == '.' && entry->d_name[2] == 0)))
continue;
size_t name_len = strlen(entry->d_name);
path_builder_set_filename(&pb, entry->d_name, name_len);
int file_type = DT_UNKNOWN;
#ifdef _DIRENT_HAVE_D_TYPE
file_type = entry->d_type;
#endif
// Fast path using d_type
if (file_type != DT_UNKNOWN) {
if (file_type == DT_LNK)
continue; // Skip symlinks
if (file_type == DT_DIR) {
char *dir_path = path_builder_dup_arena(&pb, ctx->meta_arena, false);
mpmc_push_work(ctx->dir_queue, dir_path);
continue;
}
if (file_type == DT_REG) {
atomic_fetch_add(&g_files_found, 1);
FileEntry *fe = arena_push(&ctx->meta_arena, sizeof(FileEntry), false);
fe->path = arena_push(&ctx->meta_arena, MAX_PATHLEN, false);
memcpy(fe->path, pb.buffer,
(pb.filename_pos - pb.buffer) + name_len + 1);
arena_trim_string(&ctx->meta_arena, fe->path, ARENA_TRIM_NUL);
// Use fstatat for file info
struct stat st;
if (fstatat(dir_fd, entry->d_name, &st, 0) == 0) {
#if FILE_TIMES
platform_get_file_times_fd(dir_fd, entry->d_name, &fe->created_time,
&fe->modified_time);
#endif
#if FILE_OWNER
fe->owner = arena_push(&ctx->meta_arena, MAX_PATHLEN, false);
platform_get_file_owner_fd(dir_fd, pb.buffer, fe->owner, MAX_PATHLEN);
arena_trim_string(&ctx->meta_arena, fe->owner, ARENA_TRIM_NUL);
#endif
fe->size_bytes = (uint64_t)st.st_size;
#if CHECK_FILE_SYSTEM
fe->fs_type = fs_type;
#endif
mpmc_push(ctx->file_queue, fe);
}
continue;
}
}
// Fallback for unknown types
struct stat st;
if (fstatat(dir_fd, entry->d_name, &st, AT_SYMLINK_NOFOLLOW) == 0) {
if (S_ISLNK(st.st_mode))
continue;
if (S_ISDIR(st.st_mode)) {
char *dir_path = path_builder_dup_arena(&pb, ctx->meta_arena, false);
mpmc_push_work(ctx->dir_queue, dir_path);
} else if (S_ISREG(st.st_mode)) {
atomic_fetch_add(&g_files_found, 1);
FileEntry *fe = arena_push(&ctx->meta_arena, sizeof(FileEntry), false);
fe->path = arena_push(&ctx->meta_arena, MAX_PATHLEN, false);
memcpy(fe->path, pb.buffer,
(pb.filename_pos - pb.buffer) + name_len + 1);
arena_trim_string(&ctx->meta_arena, fe->path, ARENA_TRIM_NUL);
#if FILE_TIMES
platform_get_file_times(pb.buffer, &fe->created_time,
&fe->modified_time);
#endif
#if FILE_OWNER
fe->owner = arena_push(&ctx->meta_arena, MAX_PATHLEN, false);
platform_get_file_owner(pb.buffer, fe->owner, MAX_PATHLEN);
arena_trim_string(&ctx->meta_arena, fe->owner, ARENA_TRIM_NUL);
#endif
fe->size_bytes = (uint64_t)st.st_size;
#if CHECK_FILE_SYSTEM
fe->fs_type = fs_type;
#endif
mpmc_push(ctx->file_queue, fe);
}
}
}
closedir(dir);
}
#endif
// ------------------------- Scan worker --------------------------------
static THREAD_RETURN scan_worker(void *arg) {
ScannerContext *ctx = (ScannerContext *)arg;
for (;;) {
char *dir = mpmc_pop(ctx->dir_queue);
if (!dir)
break;
scan_folder(dir, ctx);
mpmc_task_done(ctx->dir_queue, ctx->num_threads);
}
return THREAD_RETURN_VALUE;
}
// ----------------------------- Hashing helpers -----------------------------
static void xxh3_hash_file_stream(const char *path, char *out_hex,
unsigned char *buf) {
XXH128_hash_t h;
XXH3_state_t state;
XXH3_128bits_reset(&state);
FileHandle handle = os_file_open(path, FLAG_SEQUENTIAL_READ);
if (handle == INVALID_FILE_HANDLE) {
strcpy(out_hex, "ERROR");
return;
}
uint64_t bytes_read;
while (os_file_read(handle, buf, READ_BLOCK, &bytes_read) == 0 &&
bytes_read > 0) {
XXH3_128bits_update(&state, buf, (size_t)bytes_read);
atomic_fetch_add(&g_bytes_processed, bytes_read);
}
os_file_close(handle);
h = XXH3_128bits_digest(&state);
snprintf(out_hex, HASH_STRLEN, "%016llx%016llx", (unsigned long long)h.high64,
(unsigned long long)h.low64);
}
// ------------------------- Hash worker --------------------------------
static THREAD_RETURN hash_worker(void *arg) {
HasherContext *ctx = (HasherContext *)arg;
void *buf = malloc(READ_BLOCK);
for (;;) {
FileEntry *fe = mpmc_pop(ctx->file_queue);
if (!fe)
break;
// Hash
char *hash = arena_push(&ctx->arena, HASH_STRLEN, false);
xxh3_hash_file_stream(fe->path, hash, buf);
arena_trim_string(&ctx->arena, hash, ARENA_TRIM_TAB);
// Path
u64 path_len = strlen(fe->path) + 1;
char *path = arena_push(&ctx->arena, path_len, ARENA_TRIM_TAB);
memcpy(path, fe->path, path_len);
arena_trim_string(&ctx->arena, path, ARENA_TRIM_TAB);
// Size
double size_kib = (double)fe->size_bytes / 1024.0;
char *size = arena_push(&ctx->arena, 32, false);
snprintf(size, 32, "%.2f", size_kib);
arena_trim_string(&ctx->arena, size, ARENA_TRIM_NONE);
// Times
char *separator;
#if FILE_TIMES
separator = arena_push(&ctx->arena, 1, false);
*separator = '\t';
u64 time_size = 32;
char *created = arena_push(&ctx->arena, time_size, false);
format_time(fe->created_time, created, time_size);
arena_trim_string(&ctx->arena, created, ARENA_TRIM_TAB);
char *modified = arena_push(&ctx->arena, time_size, false);
format_time(fe->modified_time, modified, time_size);
arena_trim_string(&ctx->arena, modified, ARENA_TRIM_NONE);
#endif
// Owner
#if FILE_OWNER
separator = arena_push(&ctx->arena, 1, false);
*separator = '\t';
char *owner = arena_push(&ctx->arena, 128, false);
memcpy(owner, fe->owner, 128);
arena_trim_string(&ctx->arena, owner, ARENA_TRIM_NONE);
#endif
// Final newline
separator = arena_push(&ctx->arena, 1, false);
*separator = '\n';
atomic_fetch_add(&g_files_hashed, 1);
}
free(buf);
return THREAD_RETURN_VALUE;
}
// ------------------------- Progress display ---------------------------
static THREAD_RETURN progress_thread(void *arg) {
(void)arg;
HiResTimer progress_timer;
timer_start(&progress_timer);
uint64_t last_bytes = 0;
double last_time = 0.0;
double displayed_speed = 0.0;
const double sample_interval = 0.5;
// Hide cursor to prevent flickering
printf("\033[?25l");
for (;;) {
uint64_t found = atomic_load(&g_files_found);
uint64_t hashed = atomic_load(&g_files_hashed);
uint64_t bytes = atomic_load(&g_bytes_processed);
int scan_done = atomic_load(&g_scan_done);
double t = timer_elapsed(&progress_timer);
double dt = t - last_time;
if (dt >= sample_interval) {
uint64_t db = (bytes > last_bytes) ? bytes - last_bytes : 0;
displayed_speed = (double)db / (1024.0 * 1024.0) / dt;
last_bytes = bytes;
last_time = t;
}
printf("\r");
if (!scan_done) {
printf("\033[1mScanning:\033[0m %llu files | Hashed: %llu | \033[32m%.2f "
"MB/s\033[0m ",
(unsigned long long)found, (unsigned long long)hashed,
displayed_speed);
} else {
double pct = found ? (double)hashed / (double)found : 0.0;
int barw = 40;
int filled = (int)(pct * barw);
printf("[");
// Print filled part in Green (\033[32m)
printf("\033[32m");
for (int i = 0; i < filled; i++)
putchar('#');
// Reset color for empty part
printf("\033[0m");
for (int i = filled; i < barw; i++)
putchar('.');
printf("] %6.2f%% (%llu/%llu) \033[32m%.2f MB/s\033[0m ", pct * 100.0,
(unsigned long long)hashed, (unsigned long long)found,
displayed_speed);
}
fflush(stdout);
if (scan_done && hashed == found)
break;
sleep_ms(100);
}
// Restore cursor (\033[?25h) and move to next line
printf("\033[?25h\n");
return THREAD_RETURN_VALUE;
}
// ======================== IO Ring implementation ========================
#if USE_IORING
// -------------------------- Data structures ---------------------------
// Globals
u64 g_ioring_buffer_size = 4096 * 64;
static atomic_uint_fast64_t g_io_ring_fallbacks = 0;
#define IO_PENDING INT_MIN
typedef struct IoBuffer IoBuffer;
#if defined(_WIN32) || defined(_WIN64)
// Windows I/O Ring types
typedef HIORING IoRingHandle;
#define BUILD_READ_RETURN_VALUE HRESULT
#elif defined(__linux__)
// Linux io_uring types
typedef struct {
struct io_uring ring;
struct io_uring_cqe *cqe_cache;
int cqe_cache_index;
int cqe_cache_count;
} IoUring;
typedef IoUring *IoRingHandle;
typedef struct iovec IORING_BUFFER_INFO;
#define BUILD_READ_RETURN_VALUE int
#endif
typedef struct FileReadContext {
FileEntry *fe;
size_t file_size;
// For in-order hashing
size_t next_read_offset;
IoBuffer *head;
IoBuffer *tail;
// Completion tracking
size_t bytes_hashed;
uint32_t reads_hashed;
uint32_t reads_submitted;
uint32_t reads_completed;
uint32_t active_reads;
union {
XXH3_state_t hash_state; // For incremental hash (large files)
XXH128_hash_t single_hash; // For single-shot hash (small files)
};
FileHandle file_handle;
#if USE_REGISTERED_FILES
uint32_t slot_id;
#endif
bool use_incremental_hash;
bool completed;
} FileReadContext;
// -------------------------- Buffer structure ---------------------------
typedef struct IoBuffer {
FileReadContext *file;
void *data;
size_t size;
size_t offset;
size_t bytes_read;
BUILD_READ_RETURN_VALUE result;
int buffer_id;
struct IoBuffer *next;
} IoBuffer;
// Thread-local I/O Ring context
#if defined(_WIN32) || defined(_WIN64)
typedef struct ThreadIoContext {
IoRingHandle ring;
void *completion_event;
void *fallback_buffer;
IoBuffer buffers[NUM_BUFFERS_PER_THREAD];
int buffer_pool[NUM_BUFFERS_PER_THREAD];
int free_count;
int num_submissions;
int active_files;
bool submitting;
#if USE_REGISTERED_FILES
bool use_registered_files;
FileHandle registered_handles[MAX_ACTIVE_FILES];
#endif
} ThreadIoContext;
#elif defined(__linux__)
typedef struct ThreadIoContext {
IoRingHandle ring;
void *fallback_buffer;
IoBuffer buffers[NUM_BUFFERS_PER_THREAD];
int buffer_pool[NUM_BUFFERS_PER_THREAD];
int free_count;
int num_submissions;
int active_files;
bool submitting;
bool use_registered_buffers;
bool use_registered_files;
} ThreadIoContext;
#endif
typedef struct {
uint32_t MaxSubmissionQueueSize;
uint32_t MaxCompletionQueueSize;
uint32_t MaxVersion;
} IoRingCapabilities;
// ------------------------ IO Ring Abstraction -------------------------
#if defined(_WIN32) || defined(_WIN64)
// Windows I/O Ring functions
static void ioring_query_capabilities(IoRingCapabilities *caps) {
IORING_CAPABILITIES win_caps;
QueryIoRingCapabilities(&win_caps);
caps->MaxSubmissionQueueSize = win_caps.MaxSubmissionQueueSize;
caps->MaxCompletionQueueSize = win_caps.MaxCompletionQueueSize;
caps->MaxVersion = win_caps.MaxVersion;
}
static int create_ioring(ThreadIoContext *thread_ctx, uint32_t queue_size) {
IORING_CREATE_FLAGS flags = {0};
HRESULT hr = CreateIoRing(IORING_VERSION_3, flags, queue_size, queue_size * 2,
&thread_ctx->ring);
// Create completion event
thread_ctx->completion_event = CreateEvent(NULL, FALSE, FALSE, NULL);
if (thread_ctx->completion_event)
SetIoRingCompletionEvent(thread_ctx->ring, thread_ctx->completion_event);
return SUCCEEDED(hr) ? 0 : -1;
}
static int close_ioring(ThreadIoContext *thread_ctx) {
if (thread_ctx->completion_event)
CloseHandle(thread_ctx->completion_event);
CloseIoRing(thread_ctx->ring);
return 0;
}
#define USERDATA_REGISTER 1
#define MAKE_BUF_INFO(a, l) \
(IORING_BUFFER_INFO) { .Address = (a), .Length = (uint32_t)(l) }
static int ioring_submit(ThreadIoContext *thread_ctx, uint32_t *submitted) {
// uint32_t wait_count = MIN(thread_ctx->num_submissions, MAX_WAIT_COUNT);
// The wait_count in windows is not implemented yet, so we wait in
// ioring_pop_completion()
HRESULT hr;
// if (wait_count > 0) {
// hr = SubmitIoRing(ring, wait_count, SUBMIT_TIMEOUT_MS, submitted);
// } else {
hr = SubmitIoRing(thread_ctx->ring, 0, SUBMIT_TIMEOUT_MS, submitted);
// }
if (thread_ctx->num_submissions > 0) {
WaitForSingleObject(thread_ctx->completion_event, SUBMIT_TIMEOUT_MS);
}
return SUCCEEDED(hr) ? 0 : -1;
}
static void ioring_register_buffers(ThreadIoContext *thread_ctx,
IORING_BUFFER_INFO *buf_info) {
HRESULT hr = BuildIoRingRegisterBuffers(
thread_ctx->ring, NUM_BUFFERS_PER_THREAD, buf_info, USERDATA_REGISTER);
if (FAILED(hr)) {
char error_msg[256];
FormatMessageA(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM |
FORMAT_MESSAGE_IGNORE_INSERTS,
NULL, hr, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
error_msg, sizeof(error_msg), NULL);
fprintf(stderr, "WARNING: Error registering buffers: %s (0x%08X)\n",
error_msg, (unsigned int)hr);
}
// Submit registration
ioring_submit(thread_ctx, NULL);
}
#if USE_REGISTERED_FILES
static void ioring_register_files(ThreadIoContext *thread_ctx) {
HRESULT hr = BuildIoRingRegisterFileHandles(
thread_ctx->ring, MAX_ACTIVE_FILES, thread_ctx->registered_handles,
USERDATA_REGISTER);
if (FAILED(hr)) {
char error_msg[256];
FormatMessageA(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM |
FORMAT_MESSAGE_IGNORE_INSERTS,
NULL, hr, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
error_msg, sizeof(error_msg), NULL);
fprintf(stderr, "WARNING: File registration failed: %s (0x%08X)\n",
error_msg, (unsigned int)hr);
}
thread_ctx->use_registered_files = (hr == 0);
}
static void ioring_register_files_update(ThreadIoContext *thread_ctx,
FileReadContext *file) {
thread_ctx->registered_handles[file->slot_id] = file->file_handle;
ioring_register_files(thread_ctx);
}
#endif
static BUILD_READ_RETURN_VALUE ioring_build_read(ThreadIoContext *thread_ctx,
FileReadContext *file_ctx,
uint32_t buffer_id,
size_t size, uint64_t offset,
uintptr_t user_data) {
#if USE_REGISTERED_FILES
IORING_HANDLE_REF file_ref;
if (thread_ctx->use_registered_files) {
file_ref = (IORING_HANDLE_REF)IoRingHandleRefFromIndex(file_ctx->slot_id);
} else {
file_ref =
(IORING_HANDLE_REF)IoRingHandleRefFromHandle(file_ctx->file_handle);
}
#else
IORING_HANDLE_REF file_ref = IoRingHandleRefFromHandle(file_ctx->file_handle);
#endif
IORING_BUFFER_REF buffer_ref =
IoRingBufferRefFromIndexAndOffset(buffer_id, 0);
HRESULT hr =
BuildIoRingReadFile(thread_ctx->ring, file_ref, buffer_ref,
(uint32_t)size, offset, user_data, IOSQE_FLAGS_NONE);
if (FAILED(hr)) {
char error_msg[256];
FormatMessageA(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM |
FORMAT_MESSAGE_IGNORE_INSERTS,
NULL, hr, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
error_msg, sizeof(error_msg), NULL);
fprintf(stderr,
"ERROR: Building read error for file '%s' - Error: %s (Code: "
"0x%08X)\n",
file_ctx->fe->path, error_msg, (unsigned int)hr);
}
return hr;
}
static void ioring_process_completions(ThreadIoContext *restrict thread_ctx) {
uint32_t cqe_count = 0;
uint32_t wait_count = MIN(thread_ctx->num_submissions, MAX_WAIT_COUNT);
while (cqe_count < wait_count) {
// ---- Drain all available CQEs (non-blocking) ----
while (1) {
IORING_CQE win_cqe;
HRESULT hr = PopIoRingCompletion(thread_ctx->ring, &win_cqe);
if (hr != S_OK) {
// No more CQEs available right now
break;
}
if (FAILED(hr)) {
fprintf(stderr, "WARNING: PopIoRingCompletion failed (0x%lx)\n", hr);
return;
}
// Skip internal registration completions
if (win_cqe.UserData == USERDATA_REGISTER) {
continue;
}
IoBuffer *restrict buf = (IoBuffer *)win_cqe.UserData;
FileReadContext *restrict file = buf->file;
if (SUCCEEDED(win_cqe.ResultCode)) {
buf->result = 0;
buf->bytes_read = win_cqe.Information;
} else {
buf->result = win_cqe.ResultCode;
buf->bytes_read = 0;
char error_msg[256];
FormatMessageA(
FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS, NULL,
win_cqe.ResultCode, 0, error_msg, sizeof(error_msg), NULL);
fprintf(stderr,
"WARNING: I/O completion error for file '%s' - Error: %s "
"(Code: 0x%lx)\n",
buf->file->fe->path, error_msg, win_cqe.ResultCode);
}
file->active_reads--;
file->reads_completed++;
thread_ctx->num_submissions--;
cqe_count++;
}
// ---- If we already waited enough, exit ----
if (cqe_count >= wait_count) {
break;
}
// ---- Otherwise wait for more completions ----
WaitForSingleObject(thread_ctx->completion_event, SUBMIT_TIMEOUT_MS);
}
}
FileHandle ioring_open_file(FileEntry *fe) {
FileHandle handle = os_file_open(fe->path, FLAG_ASYNC_DIRECT_READ);
if (handle == INVALID_FILE_HANDLE) {
return os_file_open(fe->path, FLAG_SEQUENTIAL_READ);
}
return handle;
}
#elif defined(__linux__)
// Linux io_uring functions implementation
static void ioring_query_capabilities(IoRingCapabilities *caps) {
// Get system limits for io_uring
long max_entries = sysconf(_SC_IOV_MAX);
if (max_entries <= 0)
max_entries = 4096;
caps->MaxSubmissionQueueSize =
(uint32_t)(max_entries < 4096 ? max_entries : 4096);
caps->MaxCompletionQueueSize = caps->MaxSubmissionQueueSize * 2;
caps->MaxVersion = 1;
}
// static int async_io_create_ring(uint32_t queue_size, AsyncIoRing *ring) {
static int create_ioring(ThreadIoContext *thread_ctx, uint32_t queue_size) {
IoUring *ring_impl = (IoUring *)calloc(1, sizeof(IoUring));
if (!ring_impl)
return -1;
// Initialize io_uring
struct io_uring_params params = {0};
params.flags =
IORING_SETUP_CQSIZE |
IORING_SETUP_SINGLE_ISSUER | // Thread local io_uring
IORING_SETUP_DEFER_TASKRUN; // Do not send interupts when a CQE is ready,
// send them when a wait function is called,
// and groupe them according to the nbre or
// entries to wait (reduces syscall overhead)
params.cq_entries = queue_size * 2;
int ret = io_uring_queue_init_params(queue_size, &ring_impl->ring, &params);
if (ret < 0) {
// Fallback to without params
printf("WARNING: Creating io_uring with default params\n");
ret = io_uring_queue_init(queue_size, &ring_impl->ring, 0);
if (ret < 0) {
free(ring_impl);
return -1;
}
}
ring_impl->cqe_cache = NULL;
ring_impl->cqe_cache_index = 0;
ring_impl->cqe_cache_count = 0;
thread_ctx->ring = ring_impl;
return 0;
}
static int close_ioring(ThreadIoContext *thread_ctx) {
IoUring *rimg_impl = (IoUring *)thread_ctx->ring;
if (!rimg_impl)
return -1;
if (thread_ctx->use_registered_buffers) {
io_uring_unregister_buffers(&rimg_impl->ring);
}
io_uring_queue_exit(&rimg_impl->ring);
free(rimg_impl);
return 0;
}
#define MAKE_BUF_INFO(a, l) \
(IORING_BUFFER_INFO) { .iov_base = (a), .iov_len = (size_t)(l) }
static void ioring_register_buffers(ThreadIoContext *thread_ctx,
IORING_BUFFER_INFO *buf_info) {
int ret = io_uring_register_buffers(&((IoUring *)thread_ctx->ring)->ring,
buf_info, NUM_BUFFERS_PER_THREAD);
if (ret < 0) {
if (ret == -ENOMEM) {
struct rlimit limit;
getrlimit(RLIMIT_MEMLOCK, &limit);
fprintf(
stderr,
"WARNING: Buffer registration failed due to Memlock limit, Error: "
"Cannot allocate memory (code: -12, ENOMEM).\n"
"See README for more informations.\n");
} else {
// For any other error (e.g., EFAULT, EBUSY, EINVAL)
fprintf(stderr, "WARNING: Error registering buffers: %s (code: %d)\n",
strerror(-ret), ret);
}
fprintf(stderr, "Falling back to unregistered buffers (performance may "
"be reduced).\n");
}
thread_ctx->use_registered_buffers = (ret == 0);
}
#if USE_REGISTERED_FILES
static void ioring_register_files(ThreadIoContext *thread_ctx) {
int ret = io_uring_register_files_sparse(&((IoUring *)thread_ctx->ring)->ring,
MAX_ACTIVE_FILES);
if (ret < 0) {
fprintf(stderr,
"WARNING: File registeration failed, fallback to unregistered "
"files - Error: %s (code: %d)\n",
strerror(-ret), ret);
}
thread_ctx->use_registered_files = (ret == 0);
}
static void ioring_register_files_update(ThreadIoContext *thread_ctx,
FileReadContext *file) {
// Update the kernel's file table at the specific slot
int ret = io_uring_register_files_update(
&((IoUring *)thread_ctx->ring)->ring,
file->slot_id, // offset - which slot to update
&file->file_handle, // pointer to the fd
1 // number of files to update
);
if (ret < 0) {
fprintf(stderr,
"WARNING: File registration update failed for slot %u updating "
"file '%s' - Error: %s "
"(code: %d)\n"
"Fallback to unregistered files\n",
file->slot_id, file->fe->path, strerror(-ret), ret);
thread_ctx->use_registered_files = false;
}
}
#endif
static int ioring_build_read(ThreadIoContext *thread_ctx,
FileReadContext *file_ctx, uint32_t buffer_id,
size_t size, uint64_t offset,
uintptr_t user_data) {
struct io_uring_sqe *sqe =
io_uring_get_sqe(&((IoUring *)thread_ctx->ring)->ring);
if (!sqe) {
printf("SQE FULL\n");
return -1;
}
void *buf = thread_ctx->buffers[buffer_id].data;
#if USE_REGISTERED_FILES
if (thread_ctx->use_registered_files) {
sqe->flags |= IOSQE_FIXED_FILE;
if (thread_ctx->use_registered_buffers) {
io_uring_prep_read_fixed(sqe, file_ctx->slot_id, buf, size, offset,
buffer_id);
} else {
io_uring_prep_read(sqe, file_ctx->slot_id, buf, size, offset);
}
io_uring_sqe_set_data64(sqe, user_data);
return 0;
}
#endif
// Fallback: use regular file descriptor
if (thread_ctx->use_registered_buffers) {
io_uring_prep_read_fixed(sqe, file_ctx->file_handle, buf, size, offset,
buffer_id);
} else {
io_uring_prep_read(sqe, file_ctx->file_handle, buf, size, offset);
}
io_uring_sqe_set_data64(sqe, user_data);
return 0;
}
static int ioring_submit(ThreadIoContext *thread_ctx, uint32_t *submitted) {
int ret;
uint32_t wait_count = MIN(thread_ctx->num_submissions, MAX_WAIT_COUNT);
if (wait_count > 0) {
ret = io_uring_submit_and_wait(&((IoUring *)thread_ctx->ring)->ring,
wait_count);
} else {
ret = io_uring_submit(&((IoUring *)thread_ctx->ring)->ring);
}
if (ret < 0) {
fprintf(stderr, "ERROR: Submit error: %s (Code: %d)\n", strerror(-ret),
ret);
return -1;
}
if (submitted)
*submitted = (uint32_t)ret;
return 0;
}
static void ioring_process_completions(ThreadIoContext *thread_ctx) {
struct io_uring_cqe *cqes[NUM_BUFFERS_PER_THREAD];
unsigned cqe_count = io_uring_peek_batch_cqe(
&((IoUring *)thread_ctx->ring)->ring, cqes, NUM_BUFFERS_PER_THREAD);
if (cqe_count == 0) {
return;
}
for (unsigned i = 0; i < cqe_count; i++) {
struct io_uring_cqe *cqe = cqes[i];
int res = cqe->res;
IoBuffer *restrict buf = (IoBuffer *)cqe->user_data;
FileReadContext *restrict file = buf->file;
if (res >= 0) {
buf->result = 0;
buf->bytes_read = (uint32_t)res;
} else {
buf->result = res;
buf->bytes_read = 0;
fprintf(stderr,
"WARNING: I/O completion error for file '%s' - Error: %s (Code: "
"%d)\n",
buf->file->fe->path, strerror(-res), res);
}
file->active_reads--;
file->reads_completed++;
thread_ctx->num_submissions--;
}
// Mark CQE as seen, equivalent to io_uring_cqe_seen() but marks multiple CQEs
io_uring_cq_advance(&((IoUring *)thread_ctx->ring)->ring, cqe_count);
}
FileHandle ioring_open_file(FileEntry *fe) {
#if CHECK_FILE_SYSTEM
if (!fs_policy(fe->fs_type)) {
return os_file_open(fe->path, FLAG_SEQUENTIAL_READ);
}
#endif
FileHandle handle = os_file_open(fe->path, FLAG_ASYNC_DIRECT_READ);
if (handle == INVALID_FILE_HANDLE) {
return os_file_open(fe->path, FLAG_SEQUENTIAL_READ);
}
return handle;
}
#endif
// OS-agnostic helper macros
#define IORING_SUCCEEDED(result) ((result) >= 0)
#define IORING_FAILED(result) ((result) < 0)
// ---------------------- FIFO queue operations ---------------------------
typedef struct FileQueue {
FileReadContext files[MAX_ACTIVE_FILES];
int head;
int tail;
int count;
} FileQueue;
static FileReadContext *fq_push(FileQueue *restrict fq) {
if (fq->count == MAX_ACTIVE_FILES)
return NULL;
FileReadContext *restrict file = &fq->files[fq->tail];
#if USE_REGISTERED_FILES
file->slot_id = fq->tail;
#endif
fq->tail = (fq->tail + 1) % MAX_ACTIVE_FILES;
fq->count++;
return file;
}
static FileReadContext *fq_peek_tail(FileQueue *fq) {
if (fq->count == 0)
return NULL;
int idx = (fq->tail - 1 + MAX_ACTIVE_FILES) % MAX_ACTIVE_FILES;
return &fq->files[idx]; // return the newest file
}
static FileReadContext *fq_peek_at(FileQueue *fq, int index) {
if (index < 0 || index >= fq->count)
return NULL;
int idx = (fq->head + index) % MAX_ACTIVE_FILES;
return &fq->files[idx];
}
static void fq_trim(FileQueue *restrict fq) {
while (fq->count > 0) {
FileReadContext *restrict file = &fq->files[fq->head];
if (!file->completed)
break;
fq->head = (fq->head + 1) % MAX_ACTIVE_FILES;
fq->count--;
}
}
// ----------------- Initialize thread context -----------------------
static ThreadIoContext *ioring_init_thread(void) {
ThreadIoContext *restrict thread_ctx =
(ThreadIoContext *)calloc(1, sizeof(ThreadIoContext));
if (!thread_ctx)
return NULL;
// Query I/O Ring capabilities
IoRingCapabilities caps;
ioring_query_capabilities(&caps);
uint32_t queue_size = caps.MaxSubmissionQueueSize;
if (queue_size > 4096)
queue_size = 4096; // Cap at 4096 for reasonable memory usage
// Create I/O Ring
if (create_ioring(thread_ctx, queue_size) != 0) {
free(thread_ctx);
thread_ctx = NULL;
return NULL;
}
// Initialize buffer pool
thread_ctx->fallback_buffer = malloc(READ_BLOCK);
IORING_BUFFER_INFO buf_info[NUM_BUFFERS_PER_THREAD];
u64 buf_pool_size = g_ioring_buffer_size * NUM_BUFFERS_PER_THREAD;
// Reserve and Commit memory for buffers
void *base_ptr = plat_mem_reserve(buf_pool_size);
if (base_ptr) {
if (!plat_mem_commit(base_ptr, buf_pool_size)) {
plat_mem_release(base_ptr, 0);
close_ioring(thread_ctx);
free(thread_ctx);
thread_ctx = NULL;
return NULL;
}
} else {
close_ioring(thread_ctx);
free(thread_ctx);
thread_ctx = NULL;
return NULL;
}
for (int i = 0; i < NUM_BUFFERS_PER_THREAD; i++) {
thread_ctx->buffers[i].data = (u8 *)base_ptr + (i * g_ioring_buffer_size);
thread_ctx->buffer_pool[i] = i;
thread_ctx->buffers[i].buffer_id = i;
buf_info[i] =
MAKE_BUF_INFO(thread_ctx->buffers[i].data, g_ioring_buffer_size);
}
thread_ctx->free_count = NUM_BUFFERS_PER_THREAD;
// Register buffers
ioring_register_buffers(thread_ctx, buf_info);
#if USE_REGISTERED_FILES
ioring_register_files(thread_ctx);
#endif
thread_ctx->submitting = true;
thread_ctx->num_submissions = 0;
thread_ctx->active_files = 0;
return thread_ctx;
}
static void ioring_cleanup_thread(ThreadIoContext *thread_ctx) {
if (!thread_ctx)
return;
if (thread_ctx->ring)
close_ioring(thread_ctx);
// Free the buffer pool memory
if (thread_ctx->buffers[0].data) {
u64 buf_pool_size = g_ioring_buffer_size * NUM_BUFFERS_PER_THREAD;
plat_mem_release(thread_ctx->buffers[0].data, buf_pool_size);
}
free(thread_ctx);
thread_ctx = NULL;
}
// -------------------------- Buffer get and return ------------------------
static IoBuffer *get_free_buffer(ThreadIoContext *restrict thread_ctx) {
if (thread_ctx->free_count == 0) {
return NULL;
}
int idx = thread_ctx->buffer_pool[--thread_ctx->free_count];
IoBuffer *restrict buf = &thread_ctx->buffers[idx];
buf->bytes_read = 0;
buf->result = IO_PENDING;
buf->next = NULL;
return buf;
}
static void return_buffer(ThreadIoContext *restrict thread_ctx,
IoBuffer *restrict buf) {
if (!buf)
return;
thread_ctx->buffer_pool[thread_ctx->free_count++] = buf->buffer_id;
}
// -------------------- File operations -----------------------
static int init_file(ThreadIoContext *restrict thread_ctx,
FileReadContext *restrict file, FileEntry *restrict fe) {
#if USE_REGISTERED_FILES
uint32_t saved_slot_id = file->slot_id;
#endif
memset(file, 0, sizeof(*file));
file->fe = fe;
file->file_size = fe->size_bytes;
file->file_handle = ioring_open_file(fe);
if (file->file_handle == INVALID_FILE_HANDLE) {
#if IORING_DEBUG_PRINTS
printf("ERROR: Could not open file '%s'\n", fe->path);
#endif
return 0;
}
#if USE_REGISTERED_FILES
file->slot_id = saved_slot_id;
if (thread_ctx->use_registered_files) {
ioring_register_files_update(thread_ctx, file);
}
#endif
// Determine hash method based on file size
if (file->file_size > g_ioring_buffer_size) {
file->use_incremental_hash = true;
XXH3_128bits_reset(&file->hash_state);
}
return 1;
}
static void finalize_file(ThreadIoContext *restrict thread_ctx,
HasherContext *worker_ctx,
FileReadContext *restrict file) {
FileEntry *restrict fe = file->fe;
os_file_close(file->file_handle);
char *hash = arena_push(&worker_ctx->arena, HASH_STRLEN, false);
if (file->bytes_hashed == file->file_size) {
if (file->use_incremental_hash) {
// Large file: digest the accumulated hash state
XXH128_hash_t h = XXH3_128bits_digest(&file->hash_state);
snprintf(hash, HASH_STRLEN, "%016llx%016llx",
(unsigned long long)h.high64, (unsigned long long)h.low64);
} else {
// Small file: hash already computed, stored directly in single_hash
snprintf(hash, HASH_STRLEN, "%016llx%016llx",
(unsigned long long)file->single_hash.high64,
(unsigned long long)file->single_hash.low64);
}
} else {
#if IORING_DEBUG_PRINTS
printf("WARNING: Fallback for path: %s\n", fe->path);
#endif
atomic_fetch_add(&g_io_ring_fallbacks, 1);
xxh3_hash_file_stream(fe->path, hash, thread_ctx->fallback_buffer);
}
arena_trim_string(&worker_ctx->arena, hash, ARENA_TRIM_TAB);
// Path
u64 path_len = strlen(fe->path) + 1;
char *path = arena_push(&worker_ctx->arena, path_len, ARENA_TRIM_TAB);
memcpy(path, fe->path, path_len);
arena_trim_string(&worker_ctx->arena, path, ARENA_TRIM_TAB);
// Size
double size_kib = (double)fe->size_bytes / 1024.0;
char *size = arena_push(&worker_ctx->arena, 32, false);
snprintf(size, 32, "%.2f", size_kib);
arena_trim_string(&worker_ctx->arena, size, ARENA_TRIM_NONE);
// Time
char *separator;
#if FILE_TIMES
separator = arena_push(&worker_ctx->arena, 1, false);
*separator = '\t';
u64 time_size = 32;
char *created = arena_push(&worker_ctx->arena, time_size, false);
format_time(fe->created_time, created, time_size);
arena_trim_string(&worker_ctx->arena, created, ARENA_TRIM_TAB);
char *modified = arena_push(&worker_ctx->arena, time_size, false);
format_time(fe->modified_time, modified, time_size);
arena_trim_string(&worker_ctx->arena, modified, ARENA_TRIM_NONE);
#endif
// Owner
#if FILE_OWNER
separator = arena_push(&worker_ctx->arena, 1, false);
*separator = '\t';
char *owner = arena_push(&worker_ctx->arena, 128, false);
memcpy(owner, fe->owner, 128);
arena_trim_string(&worker_ctx->arena, owner, ARENA_TRIM_NONE);
#endif
separator = arena_push(&worker_ctx->arena, 1, false);
*separator = '\n';
atomic_fetch_add(&g_files_hashed, 1);
}
// -------------------- Hash files -----------------------
static void hash_ready_files(ThreadIoContext *restrict thread_ctx,
FileQueue *restrict fq,
HasherContext *worker_ctx) {
for (int i = 0; i < fq->count; i++) {
FileReadContext *restrict file = fq_peek_at(fq, i);
if (!file || file->completed)
continue;
// ---- HASH READY BUFFERS IN ORDER ----
while (file->head) {
IoBuffer *restrict buf = file->head;
// CQE not received yet
if (buf->result == IO_PENDING)
break;
// Consume buffer
file->head = buf->next;
if (IORING_SUCCEEDED(buf->result) && buf->bytes_read > 0) {
size_t bytes_to_hash = buf->bytes_read;
if (buf->offset + buf->bytes_read > file->file_size) {
bytes_to_hash = file->file_size - buf->offset;
}
if (bytes_to_hash > 0) {
if (file->use_incremental_hash) {
XXH3_128bits_update(&file->hash_state, buf->data, bytes_to_hash);
} else {
file->single_hash = XXH3_128bits(buf->data, bytes_to_hash);
}
file->bytes_hashed += bytes_to_hash;
atomic_fetch_add(&g_bytes_processed, bytes_to_hash);
}
file->reads_hashed++;
} else if (buf->bytes_read == 0 && IORING_SUCCEEDED(buf->result)) {
file->reads_hashed++; // EOF
} else {
finalize_file(thread_ctx, worker_ctx, file);
file->completed = true;
}
return_buffer(thread_ctx, buf);
}
// ---- FINALIZE ----
if (!file->completed && file->active_reads == 0 &&
file->bytes_hashed >= file->file_size) {
finalize_file(thread_ctx, worker_ctx, file);
file->completed = true;
thread_ctx->active_files--;
}
}
// Clean up completed files from the head
fq_trim(fq);
}
// ------------------ Build pending reads ----------------------
static void build_pending_reads(ThreadIoContext *restrict thread_ctx,
FileQueue *restrict fq,
HasherContext *worker_ctx) {
MPMCQueue *file_queue = worker_ctx->file_queue;
FileReadContext *restrict file = fq_peek_tail(fq);
for (;;) {
// BUILD READS FOR CURRENT FILE
if (file) {
while (file->next_read_offset < file->file_size) {
IoBuffer *restrict buf = get_free_buffer(thread_ctx);
if (!buf)
return;
size_t remaining = file->file_size - file->next_read_offset;
size_t bytes_to_read;
if (remaining >= g_ioring_buffer_size) {
bytes_to_read = g_ioring_buffer_size;
} else {
bytes_to_read = ALIGN_UP_POW2(remaining, g_pagesize);
}
// Initialize buffer
buf->file = file;
buf->offset = file->next_read_offset;
buf->size = bytes_to_read;
// Chain buffer
if (!file->head) {
file->head = buf;
} else {
file->tail->next = buf;
}
file->tail = buf;
BUILD_READ_RETURN_VALUE hr =
ioring_build_read(thread_ctx, file, buf->buffer_id, bytes_to_read,
buf->offset, (uintptr_t)buf);
if (IORING_FAILED(hr)) {
// mark failure and stop this file
return_buffer(thread_ctx, buf);
finalize_file(thread_ctx, worker_ctx, file);
file->completed = true;
break;
}
file->active_reads++;
file->reads_submitted++;
thread_ctx->num_submissions++;
file->next_read_offset += bytes_to_read;
}
}
// ADD NEW FILE
if (!thread_ctx->submitting)
return;
if (fq->count >= MAX_ACTIVE_FILES)
return;
FileEntry *fe = mpmc_pop(file_queue);
if (!fe) {
thread_ctx->submitting = false;
return;
}
FileReadContext *newfile = fq_push(fq);
if (!init_file(thread_ctx, newfile, fe)) {
finalize_file(thread_ctx, worker_ctx, newfile);
newfile->completed = true;
continue;
}
file = newfile;
thread_ctx->active_files++;
}
}
// -------------------------- Hash worker I/O Ring ---------------------------
static THREAD_RETURN hash_worker_ioring(void *arg) {
HasherContext *worker_ctx = (HasherContext *)arg;
// Init IO ring
ThreadIoContext *thread_ctx = ioring_init_thread();
if (!thread_ctx || !thread_ctx->ring) {
printf("I/O Ring unavailable, using buffered I/O\n");
return hash_worker(arg);
}
// Initialize pipeline state
FileQueue fq;
memset(&fq, 0, sizeof(fq));
uint32_t submitted;
// Main pipeline loop
for (;;) {
// Submit new reads
build_pending_reads(thread_ctx, &fq, worker_ctx);
submitted = 0;
ioring_submit(thread_ctx, &submitted);
// Process completions
ioring_process_completions(thread_ctx);
// Hash files
hash_ready_files(thread_ctx, &fq, worker_ctx);
#if IORING_DEBUG_STATS
printf(
"Free buffers: %d, Submissions: %d, Active files: %d, fq count: %d\n",
thread_ctx->free_count, thread_ctx->num_submissions,
thread_ctx->active_files, fq.count);
#endif
// Exit condition
if (!thread_ctx->submitting && thread_ctx->active_files == 0 &&
thread_ctx->num_submissions == 0) {
break;
}
}
ioring_cleanup_thread(thread_ctx);
return THREAD_RETURN_VALUE;
}
#endif
Reference in New Issue View Git Blame Copy Permalink