From 99c989a3c368532a1f364119b65c2540990b3a7b Mon Sep 17 00:00:00 2001 From: Ryan Fleury Date: Wed, 17 Sep 2025 10:06:13 -0700 Subject: [PATCH] first pass at setting up base layer async thread path --- project.4coder | 4 +- src/base/base_entry_point.c | 18 ++++++ src/base/base_entry_point.h | 1 + src/file_stream/file_stream.c | 113 ++++++++++++++++++++++++++++++++++ src/file_stream/file_stream.h | 28 +++++++++ 5 files changed, 162 insertions(+), 2 deletions(-) diff --git a/project.4coder b/project.4coder index 1bb5aad9..9dc98bf1 100644 --- a/project.4coder +++ b/project.4coder @@ -46,8 +46,8 @@ load_paths = commands = { //- rjf: [raddbg] - // .f1 = { .win = "raddbg_stable --ipc kill_all && build raddbg telemetry", .linux = "", .out = "*compilation*", .footer_panel = true, .save_dirty_files = true, .cursor_at_end = false, }, - .f1 = { .win = "raddbg_stable --ipc kill_all && build radbin debug telemetry", .linux = "", .out = "*compilation*", .footer_panel = true, .save_dirty_files = true, .cursor_at_end = false, }, + .f1 = { .win = "raddbg_stable --ipc kill_all && build raddbg telemetry", .linux = "", .out = "*compilation*", .footer_panel = true, .save_dirty_files = true, .cursor_at_end = false, }, + // .f1 = { .win = "raddbg_stable --ipc kill_all && build radbin debug telemetry", .linux = "", .out = "*compilation*", .footer_panel = true, .save_dirty_files = true, .cursor_at_end = false, }, //- rjf: [raddbg wsl] // .f1 = { .win = "wsl ./build.sh raddbg", .linux = "", .out = "*compilation*", .footer_panel = true, .save_dirty_files = true, .cursor_at_end = false, }, diff --git a/src/base/base_entry_point.c b/src/base/base_entry_point.c index 6c9c17f2..36b303f7 100644 --- a/src/base/base_entry_point.c +++ b/src/base/base_entry_point.c @@ -2,6 +2,8 @@ // Licensed under the MIT license (https://opensource.org/license/mit/) global U64 global_update_tick_idx = 0; +global CondVar async_tick_cond_var = {0}; +global Mutex async_tick_mutex = {0}; internal void main_thread_base_entry_point(int arguments_count, char **arguments) @@ -9,6 +11,10 @@ main_thread_base_entry_point(int arguments_count, char **arguments) Temp scratch = scratch_begin(0, 0); ThreadNameF("[main thread]"); + //- rjf: set up async thread group info + async_tick_cond_var = cond_var_alloc(); + async_tick_mutex = mutex_alloc(); + //- rjf: set up telemetry #if PROFILE_TELEMETRY local_persist char tm_data[MB(64)]; @@ -133,3 +139,15 @@ update(void) #endif return result; } + +internal void +async_thread_entry_point(void *params) +{ + MutexScope(async_tick_mutex) for(;;) + { + cond_var_wait(async_tick_cond_var, async_tick_mutex, max_U64); +#if defined(FILE_STREAM_H) + fs_async_tick(); +#endif + } +} diff --git a/src/base/base_entry_point.h b/src/base/base_entry_point.h index 346df5df..91cffd99 100644 --- a/src/base/base_entry_point.h +++ b/src/base/base_entry_point.h @@ -8,5 +8,6 @@ internal void main_thread_base_entry_point(int argc, char **argv); internal void supplement_thread_base_entry_point(void (*entry_point)(void *params), void *params); internal U64 update_tick_idx(void); internal B32 update(void); +internal void async_thread_entry_point(void *params); #endif // BASE_ENTRY_POINT_H diff --git a/src/file_stream/file_stream.c b/src/file_stream/file_stream.c index 0fec0e2c..e322ee83 100644 --- a/src/file_stream/file_stream.c +++ b/src/file_stream/file_stream.c @@ -244,6 +244,119 @@ fs_properties_from_path(String8 path) return result; } +//////////////////////////////// +//~ rjf: Asynchronous Tick + +internal void +fs_async_tick(void) +{ + Temp scratch = scratch_begin(0, 0); + + //- rjf: gather all requests + local_persist FS_Request *reqs = 0; + local_persist U64 reqs_count = 0; + if(lane_idx() == 0) MutexScope(fs_shared->req_mutex) + { + reqs_count = fs_shared->req_count; + reqs = push_array(scratch.arena, FS_Request, reqs_count); + U64 idx = 0; + for EachNode(r, FS_RequestNode, fs_shared->first_req) + { + MemoryCopyStruct(&reqs[idx], &r->v); + reqs[idx].path = str8_copy(scratch.arena, reqs[idx].path); + idx += 1; + } + arena_clear(fs_shared->req_arena); + fs_shared->first_req = fs_shared->last_req = 0; + fs_shared->req_count = 0; + } + lane_sync(); + + //- rjf: do requests + Rng1U64 range = lane_range(reqs_count); + for EachInRange(req_idx, range) + { + //- rjf: unpack + FS_Request *r = &reqs[req_idx]; + HS_Key key = r->key; + String8 path = r->path; + Rng1U64 range = r->range; + U64 path_hash = fs_little_hash_from_string(path); + U64 path_slot_idx = path_hash%fs_shared->slots_count; + U64 path_stripe_idx = path_slot_idx%fs_shared->stripes_count; + FS_Slot *path_slot = &fs_shared->slots[path_slot_idx]; + FS_Stripe *path_stripe = &fs_shared->stripes[path_stripe_idx]; + + //- rjf: load + ProfBegin("load \"%.*s\"", str8_varg(path)); + FileProperties pre_props = os_properties_from_file_path(path); + U64 range_size = dim_1u64(range); + U64 read_size = Min(pre_props.size - range.min, range_size); + OS_Handle file = os_file_open(OS_AccessFlag_Read|OS_AccessFlag_ShareRead|OS_AccessFlag_ShareWrite, path); + B32 file_handle_is_valid = !os_handle_match(os_handle_zero(), file); + U64 data_arena_size = read_size+ARENA_HEADER_SIZE; + data_arena_size += KB(4)-1; + data_arena_size -= data_arena_size%KB(4); + ProfBegin("allocate"); + Arena *data_arena = arena_alloc(.reserve_size = data_arena_size, .commit_size = data_arena_size); + ProfEnd(); + ProfBegin("read"); + String8 data = os_string_from_file_range(data_arena, file, r1u64(range.min, range.min+read_size)); + ProfEnd(); + os_file_close(file); + FileProperties post_props = os_properties_from_file_path(path); + + //- rjf: abort if modification timestamps or sizes differ - we did not successfully read the file + B32 read_good = (pre_props.modified == post_props.modified && + pre_props.size == post_props.size && + read_size == data.size && + (file_handle_is_valid || pre_props.flags & FilePropertyFlag_IsFolder)); + if(!read_good) + { + ProfScope("abort") + { + arena_release(data_arena); + MemoryZeroStruct(&data); + data_arena = 0; + } + } + + //- rjf: submit + else + { + ProfScope("submit") + { + hs_submit_data(key, &data_arena, data); + } + } + + //- rjf: commit info to cache + ProfScope("commit to cache") OS_MutexScopeW(path_stripe->rw_mutex) + { + FS_Node *node = 0; + for(FS_Node *n = path_slot->first; n != 0; n = n->next) + { + if(str8_match(n->path, path, 0)) + { + node = n; + break; + } + } + if(node != 0 && read_good) + { + if(node->props.modified != 0) + { + ins_atomic_u64_inc_eval(&fs_shared->change_gen); + } + node->props = post_props; + } + } + cond_var_broadcast(path_stripe->cv); + } + + scratch_end(scratch); +} + //////////////////////////////// //~ rjf: Streamer Threads diff --git a/src/file_stream/file_stream.h b/src/file_stream/file_stream.h index b3b4aac2..dae0393b 100644 --- a/src/file_stream/file_stream.h +++ b/src/file_stream/file_stream.h @@ -57,6 +57,22 @@ struct FS_Stripe //////////////////////////////// //~ rjf: Shared State Bundle +typedef struct FS_Request FS_Request; +struct FS_Request +{ + FS_Request *next; + HS_Key key; + String8 path; + Rng1U64 range; +}; + +typedef struct FS_RequestNode FS_RequestNode; +struct FS_RequestNode +{ + FS_RequestNode *next; + FS_Request v; +}; + typedef struct FS_Shared FS_Shared; struct FS_Shared { @@ -69,6 +85,13 @@ struct FS_Shared FS_Slot *slots; FS_Stripe *stripes; + // rjf: requests + Mutex req_mutex; + Arena *req_arena; + FS_RequestNode *first_req; + FS_RequestNode *last_req; + U64 req_count; + // rjf: user -> streamer ring buffer U64 u2s_ring_size; U8 *u2s_ring_base; @@ -109,6 +132,11 @@ internal HS_Key fs_key_from_path_range(String8 path, Rng1U64 range, U64 endt_us) internal U128 fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us); internal FileProperties fs_properties_from_path(String8 path); +//////////////////////////////// +//~ rjf: Asynchronous Tick + +internal void fs_async_tick(void); + //////////////////////////////// //~ rjf: Streaming Work