eliminate txt layer parsing threads; use async layer

src/text_cache/text_cache.c

@@ -1611,12 +1611,6 @@ txt_init(void)
   txt_shared->u2p_ring_base = push_array_no_zero(arena, U8, txt_shared->u2p_ring_size);
   txt_shared->u2p_ring_cv = os_condition_variable_alloc();
   txt_shared->u2p_ring_mutex = os_mutex_alloc();
-  txt_shared->parse_thread_count = Clamp(1, os_get_system_info()->logical_processor_count-1, 4);
-  txt_shared->parse_threads = push_array(arena, OS_Handle, txt_shared->parse_thread_count);
-  for(U64 idx = 0; idx < txt_shared->parse_thread_count; idx += 1)
-  {
-    txt_shared->parse_threads[idx] = os_thread_launch(txt_parse_thread__entry_point, (void *)idx, 0);
-  }
   txt_shared->evictor_thread = os_thread_launch(txt_evictor_thread__entry_point, 0, 0);
 }
 
@@ -1782,6 +1776,7 @@ txt_text_info_from_hash_lang(TXT_Scope *scope, U128 hash, TXT_LangKind lang)
     if(node_is_new)
     {
       txt_u2p_enqueue_req(hash, lang, max_U64);
+      async_push_work(txt_parse_work);
     }
   }
   return info;
@@ -2151,202 +2146,199 @@ txt_u2p_dequeue_req(U128 *hash_out, TXT_LangKind *lang_out)
   os_condition_variable_broadcast(txt_shared->u2p_ring_cv);
 }
 
-internal void
+ASYNC_WORK_DEF(txt_parse_work)
-txt_parse_thread__entry_point(void *p)
 {
-  for(;;)
+  //- rjf: get next key
+  U128 hash = {0};
+  TXT_LangKind lang = TXT_LangKind_Null;
+  txt_u2p_dequeue_req(&hash, &lang);
+  HS_Scope *scope = hs_scope_open();
+  
+  //- rjf: unpack hash
+  U64 slot_idx = hash.u64[1]%txt_shared->slots_count;
+  U64 stripe_idx = slot_idx%txt_shared->stripes_count;
+  TXT_Slot *slot = &txt_shared->slots[slot_idx];
+  TXT_Stripe *stripe = &txt_shared->stripes[stripe_idx];
+  
+  //- rjf: take task
+  B32 got_task = 0;
+  OS_MutexScopeR(stripe->rw_mutex)
   {
-    //- rjf: get next key
+    for(TXT_Node *n = slot->first; n != 0; n = n->next)
-    U128 hash = {0};
+    {
-    TXT_LangKind lang = TXT_LangKind_Null;
+      if(u128_match(n->hash, hash) && n->lang == lang)
-    txt_u2p_dequeue_req(&hash, &lang);
+      {
-    HS_Scope *scope = hs_scope_open();
+        got_task = !ins_atomic_u32_eval_cond_assign(&n->is_working, 1, 0);
+        break;
+      }
+    }
+  }
   
-    //- rjf: unpack hash
+  //- rjf: hash -> data
-    U64 slot_idx = hash.u64[1]%txt_shared->slots_count;
+  String8 data = {0};
-    U64 stripe_idx = slot_idx%txt_shared->stripes_count;
+  if(got_task)
-    TXT_Slot *slot = &txt_shared->slots[slot_idx];
+  {
-    TXT_Stripe *stripe = &txt_shared->stripes[stripe_idx];
+    data = hs_data_from_hash(scope, hash);
+  }
   
-    //- rjf: take task
+  //- rjf: data -> text info
-    B32 got_task = 0;
+  Arena *info_arena = 0;
+  TXT_TextInfo info = {0};
+  if(got_task && !u128_match(hash, u128_zero()))
+  {
+    info_arena = arena_alloc();
+    
+    //- rjf: grab pointers to working counters
+    U64 *bytes_processed_ptr = 0;
+    U64 *bytes_to_process_ptr = 0;
     OS_MutexScopeR(stripe->rw_mutex)
     {
       for(TXT_Node *n = slot->first; n != 0; n = n->next)
       {
         if(u128_match(n->hash, hash) && n->lang == lang)
         {
-          got_task = !ins_atomic_u32_eval_cond_assign(&n->is_working, 1, 0);
+          bytes_processed_ptr = &n->info.bytes_processed;
-          break;
+          bytes_to_process_ptr = &n->info.bytes_to_process;
         }
       }
     }
     
-    //- rjf: hash -> data
+    //- rjf: set # of bytes to process
-    String8 data = {0};
+    if(bytes_to_process_ptr)
-    if(got_task)
     {
-      data = hs_data_from_hash(scope, hash);
+      //                                               (line ending calc)     (line counting)    (line measuring)   (lexing)
+      ins_atomic_u64_eval_assign(bytes_to_process_ptr, Min(data.size, 1024) + data.size        + data.size        + data.size*(lang != TXT_LangKind_Null));
     }
     
-    //- rjf: data -> text info
+    //- rjf: detect line end kind
-    Arena *info_arena = 0;
+    TXT_LineEndKind line_end_kind = TXT_LineEndKind_Null;
-    TXT_TextInfo info = {0};
-    if(got_task && !u128_match(hash, u128_zero()))
     {
-      info_arena = arena_alloc();
+      U64 lf_count = 0;
-      
+      U64 cr_count = 0;
-      //- rjf: grab pointers to working counters
+      for(U64 idx = 0; idx < data.size && idx < 1024; idx += 1)
-      U64 *bytes_processed_ptr = 0;
-      U64 *bytes_to_process_ptr = 0;
-      OS_MutexScopeR(stripe->rw_mutex)
       {
-        for(TXT_Node *n = slot->first; n != 0; n = n->next)
+        if(data.str[idx] == '\r')
         {
-          if(u128_match(n->hash, hash) && n->lang == lang)
+          cr_count += 1;
-          {
+        }
-            bytes_processed_ptr = &n->info.bytes_processed;
+        if(data.str[idx] == '\n')
-            bytes_to_process_ptr = &n->info.bytes_to_process;
+        {
-          }
+          lf_count += 1;
         }
       }
-      
+      if(cr_count >= lf_count/2 && lf_count >= 1)
-      //- rjf: set # of bytes to process
-      if(bytes_to_process_ptr)
       {
-        //                                               (line ending calc)     (line counting)    (line measuring)   (lexing)
+        line_end_kind = TXT_LineEndKind_CRLF;
-        ins_atomic_u64_eval_assign(bytes_to_process_ptr, Min(data.size, 1024) + data.size        + data.size        + data.size*(lang != TXT_LangKind_Null));
       }
-      
+      else if(lf_count >= 1)
-      //- rjf: detect line end kind
-      TXT_LineEndKind line_end_kind = TXT_LineEndKind_Null;
       {
-        U64 lf_count = 0;
+        line_end_kind = TXT_LineEndKind_LF;
-        U64 cr_count = 0;
-        for(U64 idx = 0; idx < data.size && idx < 1024; idx += 1)
-        {
-          if(data.str[idx] == '\r')
-          {
-            cr_count += 1;
-          }
-          if(data.str[idx] == '\n')
-          {
-            lf_count += 1;
-          }
-        }
-        if(cr_count >= lf_count/2 && lf_count >= 1)
-        {
-          line_end_kind = TXT_LineEndKind_CRLF;
-        }
-        else if(lf_count >= 1)
-        {
-          line_end_kind = TXT_LineEndKind_LF;
-        }
-        info.line_end_kind = line_end_kind;
-      }
-      
-      //- rjf: bump progress
-      if(bytes_processed_ptr)
-      {
-        ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024));
-      }
-      
-      //- rjf: count # of lines
-      U64 line_count = 1;
-      U64 byte_process_start_idx = 0;
-      for(U64 idx = 0; idx < data.size; idx += 1)
-      {
-        if(data.str[idx] == '\n' || data.str[idx] == '\r')
-        {
-          line_count += 1;
-          if(data.str[idx] == '\r')
-          {
-            idx += 1;
-          }
-        }
-        if(idx && idx%1000 == 0)
-        {
-          ins_atomic_u64_add_eval(bytes_processed_ptr, 1000);
-        }
-      }
-      
-      //- rjf: bump progress
-      if(bytes_processed_ptr)
-      {
-        ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024) + data.size);
-      }
-      
-      //- rjf: allocate & store line ranges
-      info.lines_count = line_count;
-      info.lines_ranges = push_array_no_zero(info_arena, Rng1U64, info.lines_count);
-      U64 line_idx = 0;
-      U64 line_start_idx = 0;
-      for(U64 idx = 0; idx <= data.size; idx += 1)
-      {
-        if(idx == data.size || data.str[idx] == '\n' || data.str[idx] == '\r')
-        {
-          Rng1U64 line_range = r1u64(line_start_idx, idx);
-          U64 line_size = dim_1u64(line_range);
-          info.lines_ranges[line_idx] = line_range;
-          info.lines_max_size = Max(info.lines_max_size, line_size);
-          line_idx += 1;
-          line_start_idx = idx+1;
-          if(idx < data.size && data.str[idx] == '\r')
-          {
-            line_start_idx += 1;
-            idx += 1;
-          }
-        }
-        if(idx && idx%1000 == 0)
-        {
-          ins_atomic_u64_add_eval(bytes_processed_ptr, 1000);
-        }
-      }
-      
-      //- rjf: bump progress
-      if(bytes_processed_ptr)
-      {
-        ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024) + data.size + data.size);
-      }
-      
-      //- rjf: lang -> lex function
-      TXT_LangLexFunctionType *lex_function = txt_lex_function_from_lang_kind(lang);
-      
-      //- rjf: lex function * data -> tokens
-      TXT_TokenArray tokens = {0};
-      if(lex_function != 0)
-      {
-        tokens = lex_function(info_arena, bytes_processed_ptr, data);
-      }
-      info.tokens = tokens;
-      
-      //- rjf: bump progress
-      if(bytes_processed_ptr)
-      {
-        ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024) + data.size + data.size + data.size*(lex_function != 0));
       }
+      info.line_end_kind = line_end_kind;
     }
     
-    //- rjf: commit results to cache
+    //- rjf: bump progress
-    if(got_task) OS_MutexScopeW(stripe->rw_mutex)
+    if(bytes_processed_ptr)
     {
-      for(TXT_Node *n = slot->first; n != 0; n = n->next)
+      ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024));
+    }
+    
+    //- rjf: count # of lines
+    U64 line_count = 1;
+    U64 byte_process_start_idx = 0;
+    for(U64 idx = 0; idx < data.size; idx += 1)
+    {
+      if(data.str[idx] == '\n' || data.str[idx] == '\r')
       {
-        if(u128_match(n->hash, hash) && n->lang == lang)
+        line_count += 1;
+        if(data.str[idx] == '\r')
         {
-          n->arena = info_arena;
+          idx += 1;
-          info.bytes_processed = n->info.bytes_processed;
-          info.bytes_to_process = n->info.bytes_to_process;
-          MemoryCopyStruct(&n->info, &info);
-          ins_atomic_u32_eval_assign(&n->is_working, 0);
-          ins_atomic_u64_inc_eval(&n->load_count);
-          break;
         }
       }
+      if(idx && idx%1000 == 0)
+      {
+        ins_atomic_u64_add_eval(bytes_processed_ptr, 1000);
+      }
     }
     
-    hs_scope_close(scope);
+    //- rjf: bump progress
+    if(bytes_processed_ptr)
+    {
+      ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024) + data.size);
+    }
+    
+    //- rjf: allocate & store line ranges
+    info.lines_count = line_count;
+    info.lines_ranges = push_array_no_zero(info_arena, Rng1U64, info.lines_count);
+    U64 line_idx = 0;
+    U64 line_start_idx = 0;
+    for(U64 idx = 0; idx <= data.size; idx += 1)
+    {
+      if(idx == data.size || data.str[idx] == '\n' || data.str[idx] == '\r')
+      {
+        Rng1U64 line_range = r1u64(line_start_idx, idx);
+        U64 line_size = dim_1u64(line_range);
+        info.lines_ranges[line_idx] = line_range;
+        info.lines_max_size = Max(info.lines_max_size, line_size);
+        line_idx += 1;
+        line_start_idx = idx+1;
+        if(idx < data.size && data.str[idx] == '\r')
+        {
+          line_start_idx += 1;
+          idx += 1;
+        }
+      }
+      if(idx && idx%1000 == 0)
+      {
+        ins_atomic_u64_add_eval(bytes_processed_ptr, 1000);
+      }
+    }
+    
+    //- rjf: bump progress
+    if(bytes_processed_ptr)
+    {
+      ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024) + data.size + data.size);
+    }
+    
+    //- rjf: lang -> lex function
+    TXT_LangLexFunctionType *lex_function = txt_lex_function_from_lang_kind(lang);
+    
+    //- rjf: lex function * data -> tokens
+    TXT_TokenArray tokens = {0};
+    if(lex_function != 0)
+    {
+      tokens = lex_function(info_arena, bytes_processed_ptr, data);
+    }
+    info.tokens = tokens;
+    
+    //- rjf: bump progress
+    if(bytes_processed_ptr)
+    {
+      ins_atomic_u64_eval_assign(bytes_processed_ptr, Min(data.size, 1024) + data.size + data.size + data.size*(lex_function != 0));
+    }
   }
+  
+  //- rjf: commit results to cache
+  if(got_task) OS_MutexScopeW(stripe->rw_mutex)
+  {
+    for(TXT_Node *n = slot->first; n != 0; n = n->next)
+    {
+      if(u128_match(n->hash, hash) && n->lang == lang)
+      {
+        n->arena = info_arena;
+        info.bytes_processed = n->info.bytes_processed;
+        info.bytes_to_process = n->info.bytes_to_process;
+        MemoryCopyStruct(&n->info, &info);
+        ins_atomic_u32_eval_assign(&n->is_working, 0);
+        ins_atomic_u64_inc_eval(&n->load_count);
+        break;
+      }
+    }
+  }
+  
+  hs_scope_close(scope);
+  return 0;
 }
 
 ////////////////////////////////

src/text_cache/text_cache.h

@@ -218,10 +218,6 @@ struct TXT_Shared
   OS_Handle u2p_ring_cv;
   OS_Handle u2p_ring_mutex;
   
-  // rjf: parse threads
-  U64 parse_thread_count;
-  OS_Handle *parse_threads;
-  
   // rjf: evictor thread
   OS_Handle evictor_thread;
 };
@@ -303,7 +299,7 @@ internal TXT_LineTokensSlice txt_line_tokens_slice_from_info_data_line_range(Are
 
 internal B32 txt_u2p_enqueue_req(U128 hash, TXT_LangKind lang, U64 endt_us);
 internal void txt_u2p_dequeue_req(U128 *hash_out, TXT_LangKind *lang_out);
-internal void txt_parse_thread__entry_point(void *p);
+ASYNC_WORK_DEF(txt_parse_work);
 
 ////////////////////////////////
 //~ rjf: Evictor Threads