ed/raddebugger
1
0
Fork 0
mirror of https://github.com/Ed94/raddebugger.git synced 2026-07-27 01:40:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

async priorities, strip out incorrect ring position calculations

Browse Source
This commit is contained in:
Ryan Fleury
2024-11-13 18:23:18 -08:00
parent 5a79518dcb
commit 4c0ebc851c
10 changed files with 116 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/async/async.c
+47 -17
View File
@@ -10,10 +10,16 @@ async_init(void)
Arena *arena = arena_alloc();
async_shared = push_array(arena, ASYNC_Shared, 1);
async_shared->arena = arena;
async_shared->u2w_ring_size = MB(8);
async_shared->u2w_ring_base = push_array_no_zero(arena, U8, async_shared->u2w_ring_size);
async_shared->u2w_ring_mutex = os_mutex_alloc();
async_shared->u2w_ring_cv = os_condition_variable_alloc();
for EachEnumVal(ASYNC_Priority, p)
{
ASYNC_Ring *ring = &async_shared->rings[p];
ring->ring_size = MB(8);
ring->ring_base = push_array_no_zero(arena, U8, ring->ring_size);
ring->ring_mutex = os_mutex_alloc();
ring->ring_cv = os_condition_variable_alloc();
}
async_shared->ring_mutex = os_mutex_alloc();
async_shared->ring_cv = os_condition_variable_alloc();
async_shared->work_threads_count = Max(1, os_get_system_info()->logical_processor_count-1);
async_shared->work_threads = push_array(arena, OS_Handle, async_shared->work_threads_count);
for EachIndex(idx, async_shared->work_threads_count)
@@ -37,6 +43,9 @@ async_thread_count(void)
internal B32
async_push_work_(ASYNC_WorkFunctionType *work_function, ASYNC_WorkParams *params)
{
// rjf: choose ring
ASYNC_Ring *ring = &async_shared->rings[params->priority];
// rjf: build work package
ASYNC_Work work = {0};
work.work_function = work_function;
@@ -50,7 +59,7 @@ async_push_work_(ASYNC_WorkFunctionType *work_function, ASYNC_WorkParams *params
// thread, and skip ring buffer if so.
B32 queued_in_ring_buffer = 0;
B32 need_to_execute_on_this_thread = 0;
OS_MutexScope(async_shared->u2w_ring_mutex) for(;;)
OS_MutexScope(ring->ring_mutex) for(;;)
{
U64 num_available_work_threads = (async_shared->work_threads_count - ins_atomic_u64_eval(&async_shared->work_threads_live_count));
if(num_available_work_threads == 0 && async_work_thread_depth > 0)
@@ -58,8 +67,8 @@ async_push_work_(ASYNC_WorkFunctionType *work_function, ASYNC_WorkParams *params
need_to_execute_on_this_thread = 1;
break;
}
U64 unconsumed_size = async_shared->u2w_ring_write_pos - async_shared->u2w_ring_read_pos;
U64 available_size = async_shared->u2w_ring_size - unconsumed_size;
U64 unconsumed_size = ring->ring_write_pos - ring->ring_read_pos;
U64 available_size = ring->ring_size - unconsumed_size;
if(available_size >= sizeof(work))
{
queued_in_ring_buffer = 1;
@@ -67,20 +76,21 @@ async_push_work_(ASYNC_WorkFunctionType *work_function, ASYNC_WorkParams *params
{
os_semaphore_take(params->semaphore, max_U64);
}
async_shared->u2w_ring_write_pos += ring_write_struct(async_shared->u2w_ring_base, async_shared->u2w_ring_size, async_shared->u2w_ring_write_pos, &work);
ring->ring_write_pos += ring_write_struct(ring->ring_base, ring->ring_size, ring->ring_write_pos, &work);
break;
}
if(os_now_microseconds() >= params->endt_us)
{
break;
}
os_condition_variable_wait(async_shared->u2w_ring_cv, async_shared->u2w_ring_mutex, params->endt_us);
os_condition_variable_wait(ring->ring_cv, ring->ring_mutex, params->endt_us);
}
// rjf: broadcast ring buffer cv if we wrote successfully
if(queued_in_ring_buffer)
{
os_condition_variable_broadcast(async_shared->u2w_ring_cv);
os_condition_variable_broadcast(ring->ring_cv);
os_condition_variable_broadcast(async_shared->ring_cv);
}
// rjf: if we did not queue successfully, and we have determined that
@@ -145,18 +155,38 @@ async_task_join(ASYNC_Task *task)
internal ASYNC_Work
async_pop_work(void)
{
ProfBeginFunction();
ASYNC_Work work = {0};
OS_MutexScope(async_shared->u2w_ring_mutex) for(;;)
B32 done = 0;
ASYNC_Priority taken_priority = ASYNC_Priority_Low;
OS_MutexScope(async_shared->ring_mutex) for(;!done;)
{
U64 unconsumed_size = async_shared->u2w_ring_write_pos - async_shared->u2w_ring_read_pos;
if(unconsumed_size >= sizeof(work))
for(ASYNC_Priority priority = ASYNC_Priority_High;; priority = (ASYNC_Priority)(priority - 1))
{
async_shared->u2w_ring_read_pos += ring_read_struct(async_shared->u2w_ring_base, async_shared->u2w_ring_size, async_shared->u2w_ring_read_pos, &work);
break;
ASYNC_Ring *ring = &async_shared->rings[priority];
OS_MutexScope(ring->ring_mutex)
{
U64 unconsumed_size = ring->ring_write_pos - ring->ring_read_pos;
if(unconsumed_size >= sizeof(work))
{
ring->ring_read_pos += ring_read_struct(ring->ring_base, ring->ring_size, ring->ring_read_pos, &work);
done = 1;
taken_priority = priority;
}
}
if(priority == ASYNC_Priority_Low)
{
break;
}
}
if(!done)
{
os_condition_variable_wait(async_shared->ring_cv, async_shared->ring_mutex, max_U64);
}
os_condition_variable_wait(async_shared->u2w_ring_cv, async_shared->u2w_ring_mutex, max_U64);
}
os_condition_variable_broadcast(async_shared->u2w_ring_cv);
os_condition_variable_broadcast(async_shared->ring_cv);
os_condition_variable_broadcast(async_shared->rings[taken_priority].ring_cv);
ProfEnd();
return work;
}
src/async/async.h
+25 -8
View File
@@ -14,6 +14,14 @@ typedef ASYNC_WORK_SIG(ASYNC_WorkFunctionType);
////////////////////////////////
//~ rjf: Work Types
typedef enum ASYNC_Priority
{
ASYNC_Priority_Low,
ASYNC_Priority_High,
ASYNC_Priority_COUNT
}
ASYNC_Priority;
typedef struct ASYNC_WorkParams ASYNC_WorkParams;
struct ASYNC_WorkParams
{
@@ -22,6 +30,7 @@ struct ASYNC_WorkParams
OS_Handle semaphore;
U64 *completion_counter;
U64 endt_us;
ASYNC_Priority priority;
};
typedef struct ASYNC_Work ASYNC_Work;
@@ -62,18 +71,26 @@ struct ASYNC_TaskList
////////////////////////////////
//~ rjf: Shared State Bundle
typedef struct ASYNC_Ring ASYNC_Ring;
struct ASYNC_Ring
{
U64 ring_size;
U8 *ring_base;
U64 ring_write_pos;
U64 ring_read_pos;
OS_Handle ring_mutex;
OS_Handle ring_cv;
};
typedef struct ASYNC_Shared ASYNC_Shared;
struct ASYNC_Shared
{
Arena *arena;
// rjf: user -> work thread ring buffer
U64 u2w_ring_size;
U8 *u2w_ring_base;
U64 u2w_ring_write_pos;
U64 u2w_ring_read_pos;
OS_Handle u2w_ring_mutex;
OS_Handle u2w_ring_cv;
// rjf: user -> work thread ring buffers
ASYNC_Ring rings[ASYNC_Priority_COUNT];
OS_Handle ring_mutex;
OS_Handle ring_cv;
// rjf: work threads
OS_Handle *work_threads;
@@ -102,7 +119,7 @@ internal U64 async_thread_count(void);
//~ rjf: Work Kickoffs
internal B32 async_push_work_(ASYNC_WorkFunctionType *work_function, ASYNC_WorkParams *params);
#define async_push_work(function, ...) async_push_work_((function), &(ASYNC_WorkParams){.endt_us = max_U64, __VA_ARGS__})
#define async_push_work(function, ...) async_push_work_((function), &(ASYNC_WorkParams){.endt_us = max_U64, .priority = ASYNC_Priority_High, __VA_ARGS__})
////////////////////////////////
//~ rjf: Task-Based Work Helper