first pass at setting up base layer async thread path

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, },

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
+  }
+}

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

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
 

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