From 44b8c9e87cc95f73ce778fe988458c764b6866b6 Mon Sep 17 00:00:00 2001 From: Ryan Fleury Date: Mon, 4 Nov 2024 13:30:10 -0800 Subject: [PATCH] eliminate r/w lock promotion concept --- src/ctrl/ctrl_core.c | 58 ++++++++--------------------------- src/ctrl/ctrl_core.h | 3 +- src/dbgi_search/dbgi_search.c | 17 +++++----- src/dbgi_search/dbgi_search.h | 3 +- src/file_stream/file_stream.c | 52 ++++++++++--------------------- src/file_stream/file_stream.h | 3 +- src/os/core/os_core.h | 1 - 7 files changed, 45 insertions(+), 92 deletions(-) diff --git a/src/ctrl/ctrl_core.c b/src/ctrl/ctrl_core.c index c5b8edcd..460a24de 100644 --- a/src/ctrl/ctrl_core.c +++ b/src/ctrl/ctrl_core.c @@ -1253,7 +1253,8 @@ ctrl_init(void) for(U64 idx = 0; idx < ctrl_state->thread_reg_cache.stripes_count; idx += 1) { ctrl_state->thread_reg_cache.stripes[idx].arena = arena_alloc(); - ctrl_state->thread_reg_cache.stripes[idx].rw_mutex = os_rw_mutex_alloc(); + ctrl_state->thread_reg_cache.stripes[idx].r_mutex = os_rw_mutex_alloc(); + ctrl_state->thread_reg_cache.stripes[idx].w_mutex = os_rw_mutex_alloc(); } ctrl_state->module_image_info_cache.slots_count = 1024; ctrl_state->module_image_info_cache.slots = push_array(arena, CTRL_ModuleImageInfoCacheSlot, ctrl_state->module_image_info_cache.slots_count); @@ -1779,7 +1780,7 @@ ctrl_query_cached_reg_block_from_thread(Arena *arena, CTRL_EntityStore *store, C CTRL_ThreadRegCacheSlot *slot = &cache->slots[slot_idx]; CTRL_ThreadRegCacheStripe *stripe = &cache->stripes[stripe_idx]; void *result = push_array(arena, U8, reg_block_size); - OS_MutexScopeR(stripe->rw_mutex) + OS_MutexScopeR(stripe->r_mutex) { // rjf: find existing node CTRL_ThreadRegCacheNode *node = 0; @@ -1793,35 +1794,13 @@ ctrl_query_cached_reg_block_from_thread(Arena *arena, CTRL_EntityStore *store, C } // rjf: allocate existing node - if(!node) + if(!node) OS_MutexScopeW(stripe->w_mutex) { - OS_MutexScopeRWPromote(stripe->rw_mutex) - { - for(CTRL_ThreadRegCacheNode *n = slot->first; n != 0; n = n->next) - { - if(ctrl_handle_match(n->handle, handle)) - { - node = n; - break; - } - } - if(!node) - { - node = push_array(stripe->arena, CTRL_ThreadRegCacheNode, 1); - DLLPushBack(slot->first, slot->last, node); - node->handle = handle; - node->block_size = reg_block_size; - node->block = push_array(stripe->arena, U8, reg_block_size); - } - } - for(CTRL_ThreadRegCacheNode *n = slot->first; n != 0; n = n->next) - { - if(ctrl_handle_match(n->handle, handle)) - { - node = n; - break; - } - } + node = push_array(stripe->arena, CTRL_ThreadRegCacheNode, 1); + DLLPushBack(slot->first, slot->last, node); + node->handle = handle; + node->block_size = reg_block_size; + node->block = push_array(stripe->arena, U8, reg_block_size); } // rjf: copy from node @@ -1831,22 +1810,11 @@ ctrl_query_cached_reg_block_from_thread(Arena *arena, CTRL_EntityStore *store, C B32 need_stale = 1; if(node->reg_gen != current_reg_gen && dmn_thread_read_reg_block(handle.dmn_handle, result)) { - OS_MutexScopeRWPromote(stripe->rw_mutex) + OS_MutexScopeW(stripe->w_mutex) if(node != 0) { - for(CTRL_ThreadRegCacheNode *n = slot->first; n != 0; n = n->next) - { - if(ctrl_handle_match(n->handle, handle)) - { - node = n; - break; - } - } - if(node != 0) - { - need_stale = 0; - node->reg_gen = current_reg_gen; - MemoryCopy(node->block, result, reg_block_size); - } + need_stale = 0; + node->reg_gen = current_reg_gen; + MemoryCopy(node->block, result, reg_block_size); } } if(need_stale) diff --git a/src/ctrl/ctrl_core.h b/src/ctrl/ctrl_core.h index 47c0078d..6022256b 100644 --- a/src/ctrl/ctrl_core.h +++ b/src/ctrl/ctrl_core.h @@ -772,7 +772,8 @@ typedef struct CTRL_ThreadRegCacheStripe CTRL_ThreadRegCacheStripe; struct CTRL_ThreadRegCacheStripe { Arena *arena; - OS_Handle rw_mutex; + OS_Handle r_mutex; + OS_Handle w_mutex; }; typedef struct CTRL_ThreadRegCache CTRL_ThreadRegCache; diff --git a/src/dbgi_search/dbgi_search.c b/src/dbgi_search/dbgi_search.c index a322bd1e..81fa87f2 100644 --- a/src/dbgi_search/dbgi_search.c +++ b/src/dbgi_search/dbgi_search.c @@ -113,7 +113,8 @@ dis_init(void) for(U64 idx = 0; idx < dis_shared->stripes_count; idx += 1) { dis_shared->stripes[idx].arena = arena_alloc(); - dis_shared->stripes[idx].rw_mutex = os_rw_mutex_alloc(); + dis_shared->stripes[idx].r_mutex = os_rw_mutex_alloc(); + dis_shared->stripes[idx].w_mutex = os_rw_mutex_alloc(); dis_shared->stripes[idx].cv = os_condition_variable_alloc(); } dis_shared->thread_count = Min(os_get_system_info()->logical_processor_count, 2); @@ -208,7 +209,7 @@ dis_items_from_key_params_query(DIS_Scope *scope, U128 key, DIS_Params *params, DIS_Stripe *stripe = &dis_shared->stripes[stripe_idx]; //- rjf: query and/or request - OS_MutexScopeR(stripe->rw_mutex) for(;;) + OS_MutexScopeR(stripe->r_mutex) for(;;) { // rjf: map key -> node DIS_Node *node = 0; @@ -222,7 +223,7 @@ dis_items_from_key_params_query(DIS_Scope *scope, U128 key, DIS_Params *params, } // rjf: no node? -> allocate - if(node == 0) OS_MutexScopeRWPromote(stripe->rw_mutex) + if(node == 0) OS_MutexScopeW(stripe->w_mutex) { node = push_array(stripe->arena, DIS_Node, 1); SLLQueuePush(slot->first, slot->last, node); @@ -248,7 +249,7 @@ dis_items_from_key_params_query(DIS_Scope *scope, U128 key, DIS_Params *params, } // rjf: if stale -> request again - if(stale) OS_MutexScopeRWPromote(stripe->rw_mutex) + if(stale) OS_MutexScopeW(stripe->w_mutex) { if(node->gen <= node->submit_gen && node->submit_gen < node->gen + ArrayCount(node->buckets)-1) { @@ -272,7 +273,7 @@ dis_items_from_key_params_query(DIS_Scope *scope, U128 key, DIS_Params *params, } // rjf: no results, but have time to wait -> wait - os_condition_variable_wait_rw_r(stripe->cv, stripe->rw_mutex, endt_us); + os_condition_variable_wait_rw_r(stripe->cv, stripe->r_mutex, endt_us); } scratch_end(scratch); @@ -373,7 +374,7 @@ dis_search_thread__entry_point(void *p) String8 query = {0}; DIS_Params params = {RDI_SectionKind_Procedures}; U64 initial_submit_gen = 0; - OS_MutexScopeW(stripe->rw_mutex) + OS_MutexScopeR(stripe->r_mutex) OS_MutexScopeW(stripe->w_mutex) { for(DIS_Node *n = slot->first; n != 0; n = n->next) { @@ -478,7 +479,7 @@ dis_search_thread__entry_point(void *p) chunk->count += 1; items_list.total_count += 1; } - if(idx%100 == 99) OS_MutexScopeR(stripe->rw_mutex) + if(idx%100 == 99) OS_MutexScopeR(stripe->r_mutex) { for(DIS_Node *n = slot->first; n != 0; n = n->next) { @@ -520,7 +521,7 @@ dis_search_thread__entry_point(void *p) for(B32 done = 0; !done;) { B32 found = 0; - OS_MutexScopeW(stripe->rw_mutex) for(DIS_Node *n = slot->first; n != 0; n = n->next) + OS_MutexScopeR(stripe->r_mutex) OS_MutexScopeW(stripe->w_mutex) for(DIS_Node *n = slot->first; n != 0; n = n->next) { if(u128_match(n->key, key)) { diff --git a/src/dbgi_search/dbgi_search.h b/src/dbgi_search/dbgi_search.h index 3bcb9dee..4a2bb390 100644 --- a/src/dbgi_search/dbgi_search.h +++ b/src/dbgi_search/dbgi_search.h @@ -86,7 +86,8 @@ typedef struct DIS_Stripe DIS_Stripe; struct DIS_Stripe { Arena *arena; - OS_Handle rw_mutex; + OS_Handle r_mutex; + OS_Handle w_mutex; OS_Handle cv; }; diff --git a/src/file_stream/file_stream.c b/src/file_stream/file_stream.c index 3a128aa7..89232470 100644 --- a/src/file_stream/file_stream.c +++ b/src/file_stream/file_stream.c @@ -47,7 +47,8 @@ fs_init(void) { fs_shared->stripes[idx].arena = arena_alloc(); fs_shared->stripes[idx].cv = os_condition_variable_alloc(); - fs_shared->stripes[idx].rw_mutex = os_rw_mutex_alloc(); + fs_shared->stripes[idx].r_mutex = os_rw_mutex_alloc(); + fs_shared->stripes[idx].w_mutex = os_rw_mutex_alloc(); } fs_shared->u2s_ring_size = KB(64); fs_shared->u2s_ring_base = push_array_no_zero(arena, U8, fs_shared->u2s_ring_size); @@ -106,7 +107,7 @@ fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us) 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(;;) + OS_MutexScopeR(path_stripe->r_mutex) for(;;) { // rjf: path -> node FS_Node *node = 0; @@ -119,25 +120,14 @@ fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us) } } - // rjf: node does not exist? -> create & store (we must re-search after promoting r -> w) - if(node == 0) OS_MutexScopeRWPromote(path_stripe->rw_mutex) + // rjf: node does not exist? -> create & store + if(node == 0) OS_MutexScopeW(path_stripe->w_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); - } + 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 @@ -154,17 +144,9 @@ fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us) } } - // 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) + // rjf: range node does not exist? create & store + if(range_node == 0) OS_MutexScopeW(path_stripe->w_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); @@ -187,7 +169,7 @@ fs_hash_from_path_range(String8 path, Rng1U64 range, U64 endt_us) // 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); + os_condition_variable_wait_rw_r(path_stripe->cv, path_stripe->r_mutex, endt_us); } else { @@ -223,7 +205,7 @@ fs_timestamp_from_path(String8 path) 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]; - OS_MutexScopeR(stripe->rw_mutex) + OS_MutexScopeR(stripe->r_mutex) { for(FS_Node *n = slot->first; n != 0; n = n->next) { @@ -249,7 +231,7 @@ fs_size_from_path(String8 path) 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]; - OS_MutexScopeR(stripe->rw_mutex) + OS_MutexScopeR(stripe->r_mutex) { for(FS_Node *n = slot->first; n != 0; n = n->next) { @@ -374,7 +356,7 @@ ASYNC_WORK_DEF(fs_stream_work) } //- rjf: commit info to cache - ProfScope("commit to cache") OS_MutexScopeW(path_stripe->rw_mutex) + ProfScope("commit to cache") OS_MutexScopeR(path_stripe->r_mutex) OS_MutexScopeW(path_stripe->w_mutex) { FS_Node *node = 0; for(FS_Node *n = path_slot->first; n != 0; n = n->next) @@ -430,7 +412,7 @@ fs_detector_thread__entry_point(void *p) for(U64 stripe_idx = 0; stripe_idx < fs_shared->stripes_count; stripe_idx += 1) { FS_Stripe *stripe = &fs_shared->stripes[stripe_idx]; - OS_MutexScopeR(stripe->rw_mutex) for(U64 slot_in_stripe_idx = 0; slot_in_stripe_idx < slots_per_stripe; slot_in_stripe_idx += 1) + OS_MutexScopeR(stripe->r_mutex) for(U64 slot_in_stripe_idx = 0; slot_in_stripe_idx < slots_per_stripe; slot_in_stripe_idx += 1) { U64 slot_idx = stripe_idx*slots_per_stripe + slot_in_stripe_idx; FS_Slot *slot = &fs_shared->slots[slot_idx]; diff --git a/src/file_stream/file_stream.h b/src/file_stream/file_stream.h index e948c5a1..30178d9e 100644 --- a/src/file_stream/file_stream.h +++ b/src/file_stream/file_stream.h @@ -50,7 +50,8 @@ struct FS_Stripe { Arena *arena; OS_Handle cv; - OS_Handle rw_mutex; + OS_Handle r_mutex; + OS_Handle w_mutex; }; //////////////////////////////// diff --git a/src/os/core/os_core.h b/src/os/core/os_core.h index 79876e45..4d28542f 100644 --- a/src/os/core/os_core.h +++ b/src/os/core/os_core.h @@ -315,7 +315,6 @@ internal void os_semaphore_drop(OS_Handle semaphore); #define OS_MutexScope(mutex) DeferLoop(os_mutex_take(mutex), os_mutex_drop(mutex)) #define OS_MutexScopeR(mutex) DeferLoop(os_rw_mutex_take_r(mutex), os_rw_mutex_drop_r(mutex)) #define OS_MutexScopeW(mutex) DeferLoop(os_rw_mutex_take_w(mutex), os_rw_mutex_drop_w(mutex)) -#define OS_MutexScopeRWPromote(mutex) DeferLoop((os_rw_mutex_drop_r(mutex), os_rw_mutex_take_w(mutex)), (os_rw_mutex_drop_w(mutex), os_rw_mutex_take_r(mutex))) //////////////////////////////// //~ rjf: @os_hooks Dynamically-Loaded Libraries (Implemented Per-OS)