database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-26 01:22:12 +03:00
Code Activity

aio: Regularize IO worker internal naming.

Browse Source 
Adopt PgAioXXX convention for pgaio module type names.  Rename a
function that didn't use a pgaio_worker_ submodule prefix.  Rename the
internal submit function's arguments to match the indirectly relevant
function pointer declaration and nearby examples.  Rename the array of
handle IDs in PgAioSubmissionQueue to sqes, a term of art seen in the
systems it emulates, also clarifying that they're not IO handle
pointers as the old name might imply.

No change in behavior, just type, variable and function name cleanup.

Back-patch to 18.

Discussion: https://postgr.es/m/CA%2BhUKG%2BwbaZZ9Nwc_bTopm4f-7vDmCwLk80uKDHj9mq%2BUp0E%2Bg%40mail.gmail.com
This commit is contained in:
Thomas Munro
2025-07-12 13:43:27 +12:00
parent 40e105042a
commit 01d618bcd7
2 changed files with 30 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
src/backend/storage/aio/method_worker.c
View File

@ -52,26 +52,26 @@
#define IO_WORKER_WAKEUP_FANOUT 2 #define IO_WORKER_WAKEUP_FANOUT 2
typedef struct AioWorkerSubmissionQueue typedef struct PgAioWorkerSubmissionQueue
{ {
uint32 size; uint32 size;
uint32 mask; uint32 mask;
uint32 head; uint32 head;
uint32 tail; uint32 tail;
uint32 ios[FLEXIBLE_ARRAY_MEMBER]; uint32 sqes[FLEXIBLE_ARRAY_MEMBER];
} AioWorkerSubmissionQueue; } PgAioWorkerSubmissionQueue;
typedef struct AioWorkerSlot typedef struct PgAioWorkerSlot
{ {
Latch *latch; Latch *latch;
bool in_use; bool in_use;
} AioWorkerSlot; } PgAioWorkerSlot;
typedef struct AioWorkerControl typedef struct PgAioWorkerControl
{ {
uint64 idle_worker_mask; uint64 idle_worker_mask;
AioWorkerSlot workers[FLEXIBLE_ARRAY_MEMBER]; PgAioWorkerSlot workers[FLEXIBLE_ARRAY_MEMBER];
} AioWorkerControl; } PgAioWorkerControl;
static size_t pgaio_worker_shmem_size(void); static size_t pgaio_worker_shmem_size(void);
@ -96,8 +96,8 @@ int io_workers = 3;
static int io_worker_queue_size = 64; static int io_worker_queue_size = 64;
static int MyIoWorkerId; static int MyIoWorkerId;
static AioWorkerSubmissionQueue *io_worker_submission_queue; static PgAioWorkerSubmissionQueue *io_worker_submission_queue;
static AioWorkerControl *io_worker_control; static PgAioWorkerControl *io_worker_control;
static size_t static size_t
@ -106,15 +106,15 @@ pgaio_worker_queue_shmem_size(int *queue_size)
/* Round size up to next power of two so we can make a mask. */ /* Round size up to next power of two so we can make a mask. */
*queue_size = pg_nextpower2_32(io_worker_queue_size); *queue_size = pg_nextpower2_32(io_worker_queue_size);
return offsetof(AioWorkerSubmissionQueue, ios) + return offsetof(PgAioWorkerSubmissionQueue, sqes) +
sizeof(uint32) * *queue_size; sizeof(uint32) * *queue_size;
} }
static size_t static size_t
pgaio_worker_control_shmem_size(void) pgaio_worker_control_shmem_size(void)
{ {
return offsetof(AioWorkerControl, workers) + return offsetof(PgAioWorkerControl, workers) +
sizeof(AioWorkerSlot) * MAX_IO_WORKERS; sizeof(PgAioWorkerSlot) * MAX_IO_WORKERS;
} }
static size_t static size_t
@ -162,7 +162,7 @@ pgaio_worker_shmem_init(bool first_time)
} }
static int static int
pgaio_choose_idle_worker(void) pgaio_worker_choose_idle(void)
{ {
int worker; int worker;
@ -180,7 +180,7 @@ pgaio_choose_idle_worker(void)
static bool static bool
pgaio_worker_submission_queue_insert(PgAioHandle *ioh) pgaio_worker_submission_queue_insert(PgAioHandle *ioh)
{ {
AioWorkerSubmissionQueue *queue; PgAioWorkerSubmissionQueue *queue;
uint32 new_head; uint32 new_head;
queue = io_worker_submission_queue; queue = io_worker_submission_queue;
@ -192,7 +192,7 @@ pgaio_worker_submission_queue_insert(PgAioHandle *ioh)
return false; /* full */ return false; /* full */
} }
queue->ios[queue->head] = pgaio_io_get_id(ioh); queue->sqes[queue->head] = pgaio_io_get_id(ioh);
queue->head = new_head; queue->head = new_head;
return true; return true;
@ -201,14 +201,14 @@ pgaio_worker_submission_queue_insert(PgAioHandle *ioh)
static uint32 static uint32
pgaio_worker_submission_queue_consume(void) pgaio_worker_submission_queue_consume(void)
{ {
AioWorkerSubmissionQueue *queue; PgAioWorkerSubmissionQueue *queue;
uint32 result; uint32 result;
queue = io_worker_submission_queue; queue = io_worker_submission_queue;
if (queue->tail == queue->head) if (queue->tail == queue->head)
return UINT32_MAX; /* empty */ return UINT32_MAX; /* empty */
result = queue->ios[queue->tail]; result = queue->sqes[queue->tail];
queue->tail = (queue->tail + 1) & (queue->size - 1); queue->tail = (queue->tail + 1) & (queue->size - 1);
return result; return result;
@ -241,37 +241,37 @@ pgaio_worker_needs_synchronous_execution(PgAioHandle *ioh)
} }
static void static void
pgaio_worker_submit_internal(int nios, PgAioHandle *ios[]) pgaio_worker_submit_internal(int num_staged_ios, PgAioHandle **staged_ios)
{ {
PgAioHandle *synchronous_ios[PGAIO_SUBMIT_BATCH_SIZE]; PgAioHandle *synchronous_ios[PGAIO_SUBMIT_BATCH_SIZE];
int nsync = 0; int nsync = 0;
Latch *wakeup = NULL; Latch *wakeup = NULL;
int worker; int worker;
Assert(nios <= PGAIO_SUBMIT_BATCH_SIZE); Assert(num_staged_ios <= PGAIO_SUBMIT_BATCH_SIZE);
LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE); LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
for (int i = 0; i < nios; ++i) for (int i = 0; i < num_staged_ios; ++i)
{ {
Assert(!pgaio_worker_needs_synchronous_execution(ios[i])); Assert(!pgaio_worker_needs_synchronous_execution(staged_ios[i]));
if (!pgaio_worker_submission_queue_insert(ios[i])) if (!pgaio_worker_submission_queue_insert(staged_ios[i]))
{ {
/* /*
* We'll do it synchronously, but only after we've sent as many as * We'll do it synchronously, but only after we've sent as many as
* we can to workers, to maximize concurrency. * we can to workers, to maximize concurrency.
*/ */
synchronous_ios[nsync++] = ios[i]; synchronous_ios[nsync++] = staged_ios[i];
continue; continue;
} }
if (wakeup == NULL) if (wakeup == NULL)
{ {
/* Choose an idle worker to wake up if we haven't already. */ /* Choose an idle worker to wake up if we haven't already. */
worker = pgaio_choose_idle_worker(); worker = pgaio_worker_choose_idle();
if (worker >= 0) if (worker >= 0)
wakeup = io_worker_control->workers[worker].latch; wakeup = io_worker_control->workers[worker].latch;
pgaio_debug_io(DEBUG4, ios[i], pgaio_debug_io(DEBUG4, staged_ios[i],
"choosing worker %d", "choosing worker %d",
worker); worker);
} }
@ -490,7 +490,7 @@ IoWorkerMain(const void *startup_data, size_t startup_data_len)
IO_WORKER_WAKEUP_FANOUT); IO_WORKER_WAKEUP_FANOUT);
for (int i = 0; i < nwakeups; ++i) for (int i = 0; i < nwakeups; ++i)
{ {
if ((worker = pgaio_choose_idle_worker()) < 0) if ((worker = pgaio_worker_choose_idle()) < 0)
break; break;
latches[nlatches++] = io_worker_control->workers[worker].latch; latches[nlatches++] = io_worker_control->workers[worker].latch;
} }

6
src/tools/pgindent/typedefs.list
View File

@ -55,9 +55,6 @@ AggStrategy
AggTransInfo AggTransInfo
Aggref Aggref
AggregateInstrumentation AggregateInstrumentation
AioWorkerControl
AioWorkerSlot
AioWorkerSubmissionQueue
AlenState AlenState
Alias Alias
AllocBlock AllocBlock
@ -2184,6 +2181,9 @@ PgAioTargetInfo
PgAioUringCaps PgAioUringCaps
PgAioUringContext PgAioUringContext
PgAioWaitRef PgAioWaitRef
PgAioWorkerControl
PgAioWorkerSlot
PgAioWorkerSubmissionQueue
PgArchData PgArchData
PgBackendGSSStatus PgBackendGSSStatus
PgBackendSSLStatus PgBackendSSLStatus