amir/filehasher
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Experimenting with the restrict keyword

Browse Source
This commit is contained in:
amir
2026-05-04 17:27:01 +01:00
parent b8104b0fc7
commit 7d2a24d0be
2 changed files with 32 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CMakeLists.txt
View File

@@ -116,7 +116,7 @@ set(HEADERS
arena.h arena.h
base.h base.h
xxhash.h xxhash.h
lf_mpmc.h mt_mpmc.h
) )
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -281,4 +281,4 @@ message(STATUS " Compiler: ${CMAKE_C_COMPILER}")
message(STATUS " Build Type: ${CMAKE_BUILD_TYPE}") message(STATUS " Build Type: ${CMAKE_BUILD_TYPE}")
message(STATUS " Generator: ${CMAKE_GENERATOR}") message(STATUS " Generator: ${CMAKE_GENERATOR}")
message(STATUS " Platform: ${PLATFORM_NAME}") message(STATUS " Platform: ${PLATFORM_NAME}")
message(STATUS "----------------------------------------") message(STATUS "----------------------------------------")

78
platform.c
View File

@@ -2,7 +2,7 @@
#include "arena.h" #include "arena.h"
#include "base.h" #include "base.h"
#include "sm_mpmc.h" #include "mt_mpmc.h"
#include "arena.c" #include "arena.c"
#include <stdint.h> #include <stdint.h>
@@ -1295,7 +1295,7 @@ static BUILD_READ_RETURN_VALUE ioring_build_read(ThreadIoContext *thread_ctx,
return hr; return hr;
} }
static void ioring_process_completions(ThreadIoContext *thread_ctx) { static void ioring_process_completions(ThreadIoContext *restrict thread_ctx) {
uint32_t cqe_count = 0; uint32_t cqe_count = 0;
uint32_t wait_count = MIN(thread_ctx->num_submissions, MAX_WAIT_COUNT); uint32_t wait_count = MIN(thread_ctx->num_submissions, MAX_WAIT_COUNT);
@@ -1322,8 +1322,8 @@ static void ioring_process_completions(ThreadIoContext *thread_ctx) {
continue; continue;
} }
IoBuffer *buf = (IoBuffer *)win_cqe.UserData; IoBuffer *restrict buf = (IoBuffer *)win_cqe.UserData;
FileReadContext *file = buf->file; FileReadContext *restrict file = buf->file;
if (SUCCEEDED(win_cqe.ResultCode)) { if (SUCCEEDED(win_cqe.ResultCode)) {
buf->result = 0; buf->result = 0;
@@ -1588,8 +1588,8 @@ static void ioring_process_completions(ThreadIoContext *thread_ctx) {
int res = cqe->res; int res = cqe->res;
IoBuffer *buf = (IoBuffer *)cqe->user_data; IoBuffer *restrict buf = (IoBuffer *)cqe->user_data;
FileReadContext *file = buf->file; FileReadContext *restrict file = buf->file;
if (res >= 0) { if (res >= 0) {
buf->result = 0; buf->result = 0;
@@ -1641,11 +1641,11 @@ typedef struct FileQueue {
int count; int count;
} FileQueue; } FileQueue;
static FileReadContext *fq_push(FileQueue *fq) { static FileReadContext *fq_push(FileQueue *restrict fq) {
if (fq->count == MAX_ACTIVE_FILES) if (fq->count == MAX_ACTIVE_FILES)
return NULL; return NULL;
FileReadContext *file = &fq->files[fq->tail]; FileReadContext *restrict file = &fq->files[fq->tail];
#if USE_REGISTERED_FILES #if USE_REGISTERED_FILES
file->slot_id = fq->tail; file->slot_id = fq->tail;
#endif #endif
@@ -1671,11 +1671,11 @@ static FileReadContext *fq_peek_at(FileQueue *fq, int index) {
return &fq->files[idx]; return &fq->files[idx];
} }
static void fq_trim(FileQueue *fq) { static void fq_trim(FileQueue *restrict fq) {
while (fq->count > 0) { while (fq->count > 0) {
FileReadContext *f = &fq->files[fq->head]; FileReadContext *restrict file = &fq->files[fq->head];
if (!f->completed) if (!file->completed)
break; break;
fq->head = (fq->head + 1) % MAX_ACTIVE_FILES; fq->head = (fq->head + 1) % MAX_ACTIVE_FILES;
@@ -1685,7 +1685,7 @@ static void fq_trim(FileQueue *fq) {
// ----------------- Initialize thread context ----------------------- // ----------------- Initialize thread context -----------------------
static ThreadIoContext *ioring_init_thread(void) { static ThreadIoContext *ioring_init_thread(void) {
ThreadIoContext *thread_ctx = ThreadIoContext *restrict thread_ctx =
(ThreadIoContext *)calloc(1, sizeof(ThreadIoContext)); (ThreadIoContext *)calloc(1, sizeof(ThreadIoContext));
if (!thread_ctx) if (!thread_ctx)
return NULL; return NULL;
@@ -1773,14 +1773,14 @@ static void ioring_cleanup_thread(ThreadIoContext *thread_ctx) {
} }
// -------------------------- Buffer get and return ------------------------ // -------------------------- Buffer get and return ------------------------
static IoBuffer *get_free_buffer(ThreadIoContext *ctx) { static IoBuffer *get_free_buffer(ThreadIoContext *restrict thread_ctx) {
if (ctx->free_count == 0) { if (thread_ctx->free_count == 0) {
return NULL; return NULL;
} }
int idx = ctx->buffer_pool[--ctx->free_count]; int idx = thread_ctx->buffer_pool[--thread_ctx->free_count];
IoBuffer *buf = &ctx->buffers[idx]; IoBuffer *restrict buf = &thread_ctx->buffers[idx];
buf->bytes_read = 0; buf->bytes_read = 0;
buf->result = IO_PENDING; buf->result = IO_PENDING;
buf->next = NULL; buf->next = NULL;
@@ -1788,34 +1788,16 @@ static IoBuffer *get_free_buffer(ThreadIoContext *ctx) {
return buf; return buf;
} }
static void return_buffer(ThreadIoContext *ctx, IoBuffer *buf) { static void return_buffer(ThreadIoContext *restrict thread_ctx, IoBuffer *restrict buf) {
if (!buf) if (!buf)
return; return;
ctx->buffer_pool[ctx->free_count++] = buf->buffer_id; thread_ctx->buffer_pool[thread_ctx->free_count++] = buf->buffer_id;
} }
// -------------------------- Process completions ---------------------------
// static void process_completions(ThreadIoContext *thread_ctx) {
// IoRingCQE cqe;
//
// while (ioring_pop_completion(thread_ctx->ring, &cqe) == 1) {
//
// IoBuffer *buf = (IoBuffer *)cqe.UserData;
// FileReadContext *file = buf->file;
//
// buf->result = cqe.ResultCode;
// buf->bytes_read = cqe.Information;
//
// file->active_reads--;
// file->reads_completed++;
// thread_ctx->num_submissions--;
// }
// }
// -------------------- File operations ----------------------- // -------------------- File operations -----------------------
static int init_file(ThreadIoContext *thread_ctx, FileReadContext *file, static int init_file(ThreadIoContext *restrict thread_ctx, FileReadContext *restrict file,
FileEntry *fe) { FileEntry *restrict fe) {
#if USE_REGISTERED_FILES #if USE_REGISTERED_FILES
uint32_t saved_slot_id = file->slot_id; uint32_t saved_slot_id = file->slot_id;
@@ -1852,10 +1834,10 @@ static int init_file(ThreadIoContext *thread_ctx, FileReadContext *file,
return 1; return 1;
} }
static void finalize_file(ThreadIoContext *thread_ctx, static void finalize_file(ThreadIoContext *restrict thread_ctx,
WorkerContext *worker_ctx, FileReadContext *file) { WorkerContext *worker_ctx, FileReadContext *restrict file) {
FileEntry *fe = file->fe; FileEntry *restrict fe = file->fe;
os_file_close(file->file_handle); os_file_close(file->file_handle);
@@ -1906,26 +1888,26 @@ static void finalize_file(ThreadIoContext *thread_ctx,
size_kib); size_kib);
#endif #endif
char *dst = arena_push(&worker_ctx->arena, len, false); char *restrict dst = arena_push(&worker_ctx->arena, len, false);
memcpy(dst, stack_buf, len); memcpy(dst, stack_buf, len);
atomic_fetch_add(&g_files_hashed, 1); atomic_fetch_add(&g_files_hashed, 1);
} }
// -------------------- Hash files ----------------------- // -------------------- Hash files -----------------------
static void hash_ready_files(ThreadIoContext *thread_ctx, FileQueue *fq, static void hash_ready_files(ThreadIoContext *restrict thread_ctx, FileQueue *restrict fq,
WorkerContext *worker_ctx) { WorkerContext *worker_ctx) {
for (int i = 0; i < fq->count; i++) { for (int i = 0; i < fq->count; i++) {
FileReadContext *file = fq_peek_at(fq, i); FileReadContext *restrict file = fq_peek_at(fq, i);
if (!file || file->completed) if (!file || file->completed)
continue; continue;
// ---- HASH READY BUFFERS IN ORDER ---- // ---- HASH READY BUFFERS IN ORDER ----
while (file->head) { while (file->head) {
IoBuffer *buf = file->head; IoBuffer *restrict buf = file->head;
// CQE not received yet // CQE not received yet
if (buf->result == IO_PENDING) if (buf->result == IO_PENDING)
@@ -1980,12 +1962,12 @@ static void hash_ready_files(ThreadIoContext *thread_ctx, FileQueue *fq,
} }
// ------------------ Build pending reads ---------------------- // ------------------ Build pending reads ----------------------
static void build_pending_reads(ThreadIoContext *thread_ctx, FileQueue *fq, static void build_pending_reads(ThreadIoContext *restrict thread_ctx, FileQueue *restrict fq,
WorkerContext *worker_ctx) { WorkerContext *worker_ctx) {
MPMCQueue *file_queue = worker_ctx->file_queue; MPMCQueue *file_queue = worker_ctx->file_queue;
FileReadContext *file = fq_peek_tail(fq); FileReadContext *restrict file = fq_peek_tail(fq);
for (;;) { for (;;) {
@@ -1993,7 +1975,7 @@ static void build_pending_reads(ThreadIoContext *thread_ctx, FileQueue *fq,
if (file) { if (file) {
while (file->next_read_offset < file->file_size) { while (file->next_read_offset < file->file_size) {
IoBuffer *buf = get_free_buffer(thread_ctx); IoBuffer *restrict buf = get_free_buffer(thread_ctx);
if (!buf) if (!buf)
return; return;