extend file streaming cache to support range-based file stream queries; begin plugging in file stream queries as alternative to ctrl stream queries; further bugfixes & unification in eval system

src/file_stream/file_stream.c

@@ -1,6 +1,34 @@
 // Copyright (c) 2024 Epic Games Tools
 // Licensed under the MIT license (https://opensource.org/license/mit/)
 
+////////////////////////////////
+//~ rjf: Basic Helpers
+
+internal U64
+fs_little_hash_from_string(String8 string)
+{
+  U64 result = 5381;
+  for(U64 i = 0; i < string.size; i += 1)
+  {
+    result = ((result << 5) + result) + string.str[i];
+  }
+  return result;
+}
+
+internal U128
+fs_big_hash_from_string_range(String8 string, Rng1U64 range)
+{
+  Temp scratch = scratch_begin(0, 0);
+  U64 buffer_size = string.size + sizeof(U64)*2;
+  U8 *buffer = push_array_no_zero(scratch.arena, U8, buffer_size);
+  MemoryCopy(buffer, string.str, string.size);
+  MemoryCopy(buffer + string.size, &range.min, sizeof(range.min));
+  MemoryCopy(buffer + string.size + sizeof(range.min), &range.max, sizeof(range.max));
+  U128 hash = hs_hash_from_data(str8(buffer, buffer_size));
+  scratch_end(scratch);
+  return hash;
+}
+
 ////////////////////////////////
 //~ rjf: Top-Level API
 
@@ -47,21 +75,129 @@ fs_change_gen(void)
 //~ rjf: Cache Interaction
 
 internal U128
-fs_hash_from_path(String8 path, U64 endt_us)
+fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us)
 {
   Temp scratch = scratch_begin(0, 0);
-  U128 result = {0};
+  
+  //- rjf: unpack args
   path = path_normalized_from_string(scratch.arena, path);
-  U128 path_key = hs_hash_from_data(path);
+  U128 key = fs_big_hash_from_string_range(path, range);
+  
+  //- rjf: loop through key -> hash history; obtain most recent hash for this key
+  U128 result = {0};
   for(U64 rewind_idx = 0; rewind_idx < 2; rewind_idx += 1)
   {
-    result = hs_hash_from_key(path_key, rewind_idx);
+    result = hs_hash_from_key(key, rewind_idx);
+    
+    //- rjf: nonzero hash -> got valid results, return
     if(!u128_match(result, u128_zero()))
     {
       break;
     }
+    
+    //- rjf: zero hash, not rewound? -> send new stream request if needed
     else if(u128_match(result, u128_zero()) && rewind_idx == 0)
     {
+      // rjf: unpack path cache info
+      U64 path_little_hash = fs_little_hash_from_string(path);
+      U64 path_slot_idx = path_little_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: loop: request, check for results, return until we can't
+      OS_MutexScopeR(path_stripe->rw_mutex) for(;;)
+      {
+        // rjf: path -> node
+        FS_Node *node = 0;
+        for(FS_Node *n = path_slot->first; n != 0; n = n->next)
+        {
+          if(str8_match(path, n->path, 0))
+          {
+            node = n;
+            break;
+          }
+        }
+        
+        // rjf: node does not exist? -> create & store (we must re-search after promoting r -> w)
+        if(node == 0) OS_MutexScopeRWPromote(path_stripe->rw_mutex)
+        {
+          for(FS_Node *n = path_slot->first; n != 0; n = n->next)
+          {
+            if(str8_match(path, n->path, 0))
+            {
+              node = n;
+              break;
+            }
+          }
+          if(node == 0)
+          {
+            node = push_array(path_stripe->arena, FS_Node, 1);
+            SLLQueuePush(path_slot->first, path_slot->last, node);
+            node->path = push_str8_copy(path_stripe->arena, path);
+            node->slots_count = 64;
+            node->slots = push_array(path_stripe->arena, FS_RangeSlot, node->slots_count);
+          }
+        }
+        
+        // rjf: range -> node
+        U64 range_hash = fs_little_hash_from_string(str8_struct(&range));
+        U64 range_slot_idx = range_hash%node->slots_count;
+        FS_RangeSlot *range_slot = &node->slots[range_slot_idx];
+        FS_RangeNode *range_node = 0;
+        for(FS_RangeNode *n = range_slot->first; n != 0; n = n->next)
+        {
+          if(MemoryMatchStruct(&n->range, &range))
+          {
+            range_node = n;
+            break;
+          }
+        }
+        
+        // rjf: range node does not exist? create & store (we must re-search after promoting r -> w)
+        if(range_node == 0) OS_MutexScopeRWPromote(path_stripe->rw_mutex)
+        {
+          for(FS_RangeNode *n = range_slot->first; n != 0; n = n->next)
+          {
+            if(MemoryMatchStruct(&n->range, &range))
+            {
+              range_node = n;
+              break;
+            }
+          }
+          if(range_node == 0)
+          {
+            range_node = push_array(path_stripe->arena, FS_RangeNode, 1);
+            SLLQueuePush(range_slot->first, range_slot->last, range_node);
+            range_node->range = range;
+          }
+        }
+        
+        // rjf: try to send stream request
+        if(!ins_atomic_u32_eval_cond_assign(&range_node->is_working, 1, 0) &&
+           !fs_u2s_enqueue_req(range, path, endt_us))
+        {
+          ins_atomic_u32_eval_assign(&range_node->is_working, 0);
+        }
+        
+        // rjf: try to reobtain results
+        result = hs_hash_from_key(key, 0);
+        
+        // rjf: have time to wait? -> wait on this stripe; otherwise exit
+        if(u128_match(result, u128_zero()) && os_now_microseconds() <= endt_us)
+        {
+          os_condition_variable_wait_rw_r(path_stripe->cv, path_stripe->rw_mutex, endt_us);
+        }
+        else
+        {
+          break;
+        }
+      }
+      
+      //-
+      // TODO(rjf): OLD vvvvvvvvvvvvvvvv
+      //-
+#if 0
       U64 slot_idx = path_key.u64[0]%fs_shared->slots_count;
       U64 stripe_idx = slot_idx%fs_shared->stripes_count;
       FS_Slot *slot = &fs_shared->slots[slot_idx];
@@ -84,7 +220,7 @@ fs_hash_from_path(String8 path, U64 endt_us)
           node->path = push_str8_copy(stripe->arena, path);
         }
         if(!ins_atomic_u32_eval_cond_assign(&node->is_working, 1, 0) &&
-           !fs_u2s_enqueue_path(path, endt_us))
+           !fs_u2s_enqueue_req(range, path, endt_us))
         {
           ins_atomic_u32_eval_assign(&node->is_working, 0);
         }
@@ -98,17 +234,22 @@ fs_hash_from_path(String8 path, U64 endt_us)
           break;
         }
       }
+#endif
     }
   }
+  
   scratch_end(scratch);
   return result;
 }
 
 internal U128
-fs_key_from_path(String8 path)
+fs_key_from_path_range(String8 path, Rng1U64 range)
 {
-  U128 key = hs_hash_from_data(path);
-  fs_hash_from_path(path, 0);
+  Temp scratch = scratch_begin(0, 0);
+  String8 path_normalized = path_normalized_from_string(scratch.arena, path);
+  U128 key = fs_big_hash_from_string_range(path_normalized, range);
+  fs_hash_from_path_range(path_normalized, range, 0);
+  scratch_end(scratch);
   return key;
 }
 
@@ -118,8 +259,8 @@ fs_timestamp_from_path(String8 path)
   Temp scratch = scratch_begin(0, 0);
   U64 result = 0;
   path = path_normalized_from_string(scratch.arena, path);
-  U128 path_key = hs_hash_from_data(path);
-  U64 slot_idx = path_key.u64[0]%fs_shared->slots_count;
+  U64 path_hash = fs_little_hash_from_string(path);
+  U64 slot_idx = path_hash%fs_shared->slots_count;
   U64 stripe_idx = slot_idx%fs_shared->stripes_count;
   FS_Slot *slot = &fs_shared->slots[slot_idx];
   FS_Stripe *stripe = &fs_shared->stripes[stripe_idx];
@@ -144,8 +285,8 @@ fs_size_from_path(String8 path)
   Temp scratch = scratch_begin(0, 0);
   U64 result = 0;
   path = path_normalized_from_string(scratch.arena, path);
-  U128 path_key = hs_hash_from_data(path);
-  U64 slot_idx = path_key.u64[0]%fs_shared->slots_count;
+  U64 path_hash = fs_little_hash_from_string(path);
+  U64 slot_idx = path_hash%fs_shared->slots_count;
   U64 stripe_idx = slot_idx%fs_shared->stripes_count;
   FS_Slot *slot = &fs_shared->slots[slot_idx];
   FS_Stripe *stripe = &fs_shared->stripes[stripe_idx];
@@ -168,7 +309,7 @@ fs_size_from_path(String8 path)
 //~ rjf: Streamer Threads
 
 internal B32
-fs_u2s_enqueue_path(String8 path, U64 endt_us)
+fs_u2s_enqueue_req(Rng1U64 range, String8 path, U64 endt_us)
 {
   B32 result = 0;
   path.size = Min(path.size, fs_shared->u2s_ring_size);
@@ -179,6 +320,8 @@ fs_u2s_enqueue_path(String8 path, U64 endt_us)
     if(available_size >= sizeof(U64) + path.size)
     {
       result = 1;
+      fs_shared->u2s_ring_write_pos += ring_write_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_write_pos, &range.min);
+      fs_shared->u2s_ring_write_pos += ring_write_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_write_pos, &range.max);
       fs_shared->u2s_ring_write_pos += ring_write_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_write_pos, &path.size);
       fs_shared->u2s_ring_write_pos += ring_write(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_write_pos, path.str, path.size);
       fs_shared->u2s_ring_write_pos += 7;
@@ -194,18 +337,19 @@ fs_u2s_enqueue_path(String8 path, U64 endt_us)
   return result;
 }
 
-internal String8
-fs_u2s_dequeue_path(Arena *arena)
+internal void
+fs_u2s_dequeue_req(Arena *arena, Rng1U64 *range_out, String8 *path_out)
 {
-  String8 path = {0};
   OS_MutexScope(fs_shared->u2s_ring_mutex) for(;;)
   {
     U64 unconsumed_size = fs_shared->u2s_ring_write_pos - fs_shared->u2s_ring_read_pos;
     if(unconsumed_size >= sizeof(U64))
     {
-      fs_shared->u2s_ring_read_pos += ring_read_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_read_pos, &path.size);
-      path.str = push_array(arena, U8, path.size);
-      fs_shared->u2s_ring_read_pos += ring_read(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_read_pos, path.str, path.size);
+      fs_shared->u2s_ring_read_pos += ring_read_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_read_pos, &range_out->min);
+      fs_shared->u2s_ring_read_pos += ring_read_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_read_pos, &range_out->max);
+      fs_shared->u2s_ring_read_pos += ring_read_struct(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_read_pos, &path_out->size);
+      path_out->str = push_array(arena, U8, path_out->size);
+      fs_shared->u2s_ring_read_pos += ring_read(fs_shared->u2s_ring_base, fs_shared->u2s_ring_size, fs_shared->u2s_ring_read_pos, path_out->str, path_out->size);
       fs_shared->u2s_ring_read_pos += 7;
       fs_shared->u2s_ring_read_pos -= fs_shared->u2s_ring_read_pos%8;
       break;
@@ -213,7 +357,6 @@ fs_u2s_dequeue_path(Arena *arena)
     os_condition_variable_wait(fs_shared->u2s_ring_cv, fs_shared->u2s_ring_mutex, max_U64);
   }
   os_condition_variable_broadcast(fs_shared->u2s_ring_cv);
-  return path;
 }
 
 internal void
@@ -224,26 +367,33 @@ fs_streamer_thread__entry_point(void *p)
   {
     Temp scratch = scratch_begin(0, 0);
     
-    //- rjf: unpack path
-    String8 path = fs_u2s_dequeue_path(scratch.arena);
-    U128 key = hs_hash_from_data(path);
-    U64 slot_idx = key.u64[0]%fs_shared->slots_count;
-    U64 stripe_idx = slot_idx%fs_shared->stripes_count;
-    FS_Slot *slot = &fs_shared->slots[slot_idx];
-    FS_Stripe *stripe = &fs_shared->stripes[stripe_idx];
+    //- rjf: get next request
+    Rng1U64 range = {0};
+    String8 path = {0};
+    fs_u2s_dequeue_req(scratch.arena, &range, &path);
+    
+    //- rjf: unpack request
+    U128 key = fs_big_hash_from_string_range(path, 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_size);
     OS_Handle file = os_file_open(OS_AccessFlag_Read|OS_AccessFlag_ShareRead|OS_AccessFlag_ShareWrite, path);
-    U64 data_arena_size = pre_props.size+ARENA_HEADER_SIZE;
+    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(0, pre_props.size));
+    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);
@@ -269,10 +419,10 @@ fs_streamer_thread__entry_point(void *p)
     }
     
     //- rjf: commit info to cache
-    ProfScope("commit to cache") OS_MutexScopeW(stripe->rw_mutex)
+    ProfScope("commit to cache") OS_MutexScopeW(path_stripe->rw_mutex)
     {
       FS_Node *node = 0;
-      for(FS_Node *n = slot->first; n != 0; n = n->next)
+      for(FS_Node *n = path_slot->first; n != 0; n = n->next)
       {
         if(str8_match(n->path, path, 0))
         {
@@ -291,10 +441,22 @@ fs_streamer_thread__entry_point(void *p)
           node->timestamp = post_props.modified;
           node->size = post_props.size;
         }
-        ins_atomic_u32_eval_assign(&node->is_working, 0);
+        U64 range_hash = fs_little_hash_from_string(str8_struct(&range));
+        U64 range_slot_idx = range_hash%node->slots_count;
+        FS_RangeSlot *range_slot = &node->slots[range_slot_idx];
+        FS_RangeNode *range_node = 0;
+        for(FS_RangeNode *n = range_slot->first; n != 0; n = n->next)
+        {
+          if(MemoryMatchStruct(&n->range, &range))
+          {
+            range_node = n;
+            break;
+          }
+        }
+        ins_atomic_u32_eval_assign(&range_node->is_working, 0);
       }
     }
-    os_condition_variable_broadcast(stripe->cv);
+    os_condition_variable_broadcast(path_stripe->cv);
     
     ProfEnd();
     scratch_end(scratch);
@@ -323,10 +485,18 @@ fs_detector_thread__entry_point(void *p)
           FileProperties props = os_properties_from_file_path(n->path);
           if(props.modified != n->timestamp)
           {
-            if(!ins_atomic_u32_eval_cond_assign(&n->is_working, 1, 0) &&
-               !fs_u2s_enqueue_path(n->path, os_now_microseconds()+100000))
+            for(U64 range_slot_idx = 0; range_slot_idx < n->slots_count; range_slot_idx += 1)
             {
-              ins_atomic_u32_eval_assign(&n->is_working, 0);
+              for(FS_RangeNode *range_n = n->slots[range_slot_idx].first;
+                  range_n != 0;
+                  range_n = range_n->next)
+              {
+                if(!ins_atomic_u32_eval_cond_assign(&range_n->is_working, 1, 0) &&
+                   !fs_u2s_enqueue_req(range_n->range, n->path, os_now_microseconds()+100000))
+                {
+                  ins_atomic_u32_eval_assign(&range_n->is_working, 0);
+                }
+              }
             }
           }
         }

src/file_stream/file_stream.h

@@ -7,14 +7,34 @@
 ////////////////////////////////
 //~ rjf: Per-Path Info Cache Types
 
+typedef struct FS_RangeNode FS_RangeNode;
+struct FS_RangeNode
+{
+  FS_RangeNode *next;
+  Rng1U64 range;
+  B32 is_working;
+};
+
+typedef struct FS_RangeSlot FS_RangeSlot;
+struct FS_RangeSlot
+{
+  FS_RangeNode *first;
+  FS_RangeNode *last;
+};
+
 typedef struct FS_Node FS_Node;
 struct FS_Node
 {
   FS_Node *next;
+  
+  // rjf: file metadata
   String8 path;
   U64 timestamp;
   U64 size;
-  B32 is_working;
+  
+  // rjf: sub-table of per-requested-file-range info
+  U64 slots_count;
+  FS_RangeSlot *slots;
 };
 
 typedef struct FS_Slot FS_Slot;
@@ -68,6 +88,12 @@ struct FS_Shared
 
 global FS_Shared *fs_shared = 0;
 
+////////////////////////////////
+//~ rjf: Basic Helpers
+
+internal U64 fs_little_hash_from_string(String8 string);
+internal U128 fs_big_hash_from_string_range(String8 string, Rng1U64 range);
+
 ////////////////////////////////
 //~ rjf: Top-Level API
 
@@ -81,16 +107,17 @@ internal U64 fs_change_gen(void);
 ////////////////////////////////
 //~ rjf: Cache Interaction
 
-internal U128 fs_hash_from_path(String8 path, U64 endt_us);
-internal U128 fs_key_from_path(String8 path);
+internal U128 fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us);
+internal U128 fs_key_from_path_range(String8 path, Rng1U64 range);
+
 internal U64 fs_timestamp_from_path(String8 path);
 internal U64 fs_size_from_path(String8 path);
 
 ////////////////////////////////
 //~ rjf: Streamer Threads
 
-internal B32 fs_u2s_enqueue_path(String8 path, U64 endt_us);
-internal String8 fs_u2s_dequeue_path(Arena *arena);
+internal B32 fs_u2s_enqueue_req(Rng1U64 range, String8 path, U64 endt_us);
+internal void fs_u2s_dequeue_req(Arena *arena, Rng1U64 *range_out, String8 *path_out);
 
 internal void fs_streamer_thread__entry_point(void *p);