From 69e66d04061a3804181015c543e3ac19a77793b6 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Sat, 21 Mar 2026 12:43:16 +0200 Subject: [PATCH 11/14] Convert AIO to the new interface This replaces the "shmem_size" and "shmem_init" callbacks in the IO methods table with the same ShmemCallback struct that we now use in other subsystems --- src/backend/access/transam/clog.c | 20 ++-- src/backend/access/transam/commit_ts.c | 24 ++--- src/backend/access/transam/multixact.c | 42 ++++---- src/backend/access/transam/slru.c | 8 +- src/backend/access/transam/subtrans.c | 20 ++-- src/backend/commands/async.c | 30 +++--- src/backend/storage/aio/aio_init.c | 121 ++++++++++++++-------- src/backend/storage/aio/method_io_uring.c | 42 +++++--- src/backend/storage/aio/method_worker.c | 86 ++++++++------- src/backend/storage/ipc/ipci.c | 2 - src/backend/storage/lmgr/predicate.c | 89 ++++++++-------- src/include/access/slru.h | 6 +- src/include/storage/aio_internal.h | 16 +-- src/include/storage/aio_subsys.h | 4 - src/include/storage/subsystemlist.h | 3 + src/test/modules/test_slru/test_slru.c | 40 +++---- 16 files changed, 296 insertions(+), 257 deletions(-) diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c index 87f7f5707de..95f160879e0 100644 --- a/src/backend/access/transam/clog.c +++ b/src/backend/access/transam/clog.c @@ -810,19 +810,19 @@ CLOGShmemRequest(void *arg) } Assert(transaction_buffers != 0); SimpleLruRequest(&XactSlruDesc, - .name = "transaction", - .Dir = "pg_xact", - .long_segment_names = false, + .name = "transaction", + .Dir = "pg_xact", + .long_segment_names = false, - .nslots = CLOGShmemBuffers(), - .nlsns = CLOG_LSNS_PER_PAGE, + .nslots = CLOGShmemBuffers(), + .nlsns = CLOG_LSNS_PER_PAGE, - .sync_handler = SYNC_HANDLER_CLOG, - .PagePrecedes = CLOGPagePrecedes, - .errdetail_for_io_error = clog_errdetail_for_io_error, + .sync_handler = SYNC_HANDLER_CLOG, + .PagePrecedes = CLOGPagePrecedes, + .errdetail_for_io_error = clog_errdetail_for_io_error, - .buffer_tranche_id = LWTRANCHE_XACT_BUFFER, - .bank_tranche_id = LWTRANCHE_XACT_SLRU, + .buffer_tranche_id = LWTRANCHE_XACT_BUFFER, + .bank_tranche_id = LWTRANCHE_XACT_SLRU, ); } diff --git a/src/backend/access/transam/commit_ts.c b/src/backend/access/transam/commit_ts.c index 236d8fb4baa..675dac9e40f 100644 --- a/src/backend/access/transam/commit_ts.c +++ b/src/backend/access/transam/commit_ts.c @@ -551,24 +551,24 @@ CommitTsShmemRequest(void *arg) } Assert(commit_timestamp_buffers != 0); SimpleLruRequest(&CommitTsSlruDesc, - .name = "commit_timestamp", - .Dir = "pg_commit_ts", - .long_segment_names = false, + .name = "commit_timestamp", + .Dir = "pg_commit_ts", + .long_segment_names = false, - .nslots = CommitTsShmemBuffers(), + .nslots = CommitTsShmemBuffers(), - .PagePrecedes = CommitTsPagePrecedes, - .errdetail_for_io_error = commit_ts_errdetail_for_io_error, + .PagePrecedes = CommitTsPagePrecedes, + .errdetail_for_io_error = commit_ts_errdetail_for_io_error, - .sync_handler = SYNC_HANDLER_COMMIT_TS, - .buffer_tranche_id = LWTRANCHE_COMMITTS_BUFFER, - .bank_tranche_id = LWTRANCHE_COMMITTS_SLRU, + .sync_handler = SYNC_HANDLER_COMMIT_TS, + .buffer_tranche_id = LWTRANCHE_COMMITTS_BUFFER, + .bank_tranche_id = LWTRANCHE_COMMITTS_SLRU, ); ShmemRequestStruct(&CommitTsShmemDesc, - .name = "CommitTs shared", - .size = sizeof(CommitTimestampShared), - .ptr = (void **) &commitTsShared, + .name = "CommitTs shared", + .size = sizeof(CommitTimestampShared), + .ptr = (void **) &commitTsShared, ); } diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c index 940ac5a78d6..88e46d6868d 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -1779,39 +1779,39 @@ MultiXactShmemRequest(void *arg) size = add_size(size, mul_size(sizeof(MultiXactId), NumVisibleSlots)); ShmemRequestStruct(&MultiXactShmemDesc, - .name = "Shared MultiXact State", - .size = size, - .ptr = (void **) &MultiXactState, + .name = "Shared MultiXact State", + .size = size, + .ptr = (void **) &MultiXactState, ); SimpleLruRequest(&MultiXactOffsetSlruDesc, - .name = "multixact_offset", - .Dir = "pg_multixact/offsets", - .long_segment_names = false, + .name = "multixact_offset", + .Dir = "pg_multixact/offsets", + .long_segment_names = false, - .nslots = multixact_offset_buffers, + .nslots = multixact_offset_buffers, - .sync_handler = SYNC_HANDLER_MULTIXACT_OFFSET, - .PagePrecedes = MultiXactOffsetPagePrecedes, - .errdetail_for_io_error = MultiXactOffsetIoErrorDetail, + .sync_handler = SYNC_HANDLER_MULTIXACT_OFFSET, + .PagePrecedes = MultiXactOffsetPagePrecedes, + .errdetail_for_io_error = MultiXactOffsetIoErrorDetail, - .buffer_tranche_id = LWTRANCHE_MULTIXACTOFFSET_BUFFER, - .bank_tranche_id = LWTRANCHE_MULTIXACTOFFSET_SLRU, + .buffer_tranche_id = LWTRANCHE_MULTIXACTOFFSET_BUFFER, + .bank_tranche_id = LWTRANCHE_MULTIXACTOFFSET_SLRU, ); SimpleLruRequest(&MultiXactMemberSlruDesc, - .name = "multixact_member", - .Dir = "pg_multixact/members", - .long_segment_names = true, + .name = "multixact_member", + .Dir = "pg_multixact/members", + .long_segment_names = true, - .nslots = multixact_member_buffers, + .nslots = multixact_member_buffers, - .sync_handler = SYNC_HANDLER_MULTIXACT_MEMBER, - .PagePrecedes = MultiXactMemberPagePrecedes, - .errdetail_for_io_error = MultiXactMemberIoErrorDetail, + .sync_handler = SYNC_HANDLER_MULTIXACT_MEMBER, + .PagePrecedes = MultiXactMemberPagePrecedes, + .errdetail_for_io_error = MultiXactMemberIoErrorDetail, - .buffer_tranche_id = LWTRANCHE_MULTIXACTMEMBER_BUFFER, - .bank_tranche_id = LWTRANCHE_MULTIXACTMEMBER_SLRU, + .buffer_tranche_id = LWTRANCHE_MULTIXACTMEMBER_BUFFER, + .bank_tranche_id = LWTRANCHE_MULTIXACTMEMBER_SLRU, ); /* diff --git a/src/backend/access/transam/slru.c b/src/backend/access/transam/slru.c index 3fe60c5804b..6d9dda6b29b 100644 --- a/src/backend/access/transam/slru.c +++ b/src/backend/access/transam/slru.c @@ -242,9 +242,9 @@ SimpleLruAutotuneBuffers(int divisor, int max) * Register a simple LRU cache in shared memory. */ void -SimpleLruRequestWithOpts(SlruDesc *desc, const SlruOpts *options) +SimpleLruRequestWithOpts(SlruDesc *desc, const SlruOpts * options) { - SlruOpts *options_copy; + SlruOpts *options_copy; Assert(options->name != NULL); Assert(options->nslots > 0); @@ -265,7 +265,7 @@ SimpleLruRequestWithOpts(SlruDesc *desc, const SlruOpts *options) void shmem_slru_init(ShmemStructDesc *base_desc, ShmemStructOpts *base_options) { - SlruOpts *options = (SlruOpts *) base_options; + SlruOpts *options = (SlruOpts *) base_options; SlruDesc *desc = (SlruDesc *) base_desc; char namebuf[NAMEDATALEN]; SlruShared shared; @@ -356,7 +356,7 @@ shmem_slru_init(ShmemStructDesc *base_desc, ShmemStructOpts *base_options) void shmem_slru_attach(ShmemStructDesc *base_desc, ShmemStructOpts *base_options) { - SlruOpts *options = (SlruOpts *) base_options; + SlruOpts *options = (SlruOpts *) base_options; SlruDesc *desc = (SlruDesc *) base_desc; int nslots = options->nslots; int nbanks = nslots / SLRU_BANK_SIZE; diff --git a/src/backend/access/transam/subtrans.c b/src/backend/access/transam/subtrans.c index ca273fb4680..5f68c3f0cca 100644 --- a/src/backend/access/transam/subtrans.c +++ b/src/backend/access/transam/subtrans.c @@ -244,19 +244,19 @@ SUBTRANSShmemRequest(void *arg) Assert(subtransaction_buffers != 0); SimpleLruRequest(&SubTransSlruDesc, - .name = "subtransaction", - .Dir = "pg_subtrans", - .long_segment_names = false, + .name = "subtransaction", + .Dir = "pg_subtrans", + .long_segment_names = false, - .nslots = SUBTRANSShmemBuffers(), + .nslots = SUBTRANSShmemBuffers(), - .sync_handler = SYNC_HANDLER_NONE, - .PagePrecedes = SubTransPagePrecedes, - .errdetail_for_io_error = subtrans_errdetail_for_io_error, + .sync_handler = SYNC_HANDLER_NONE, + .PagePrecedes = SubTransPagePrecedes, + .errdetail_for_io_error = subtrans_errdetail_for_io_error, - .buffer_tranche_id = LWTRANCHE_SUBTRANS_BUFFER, - .bank_tranche_id = LWTRANCHE_SUBTRANS_SLRU, - ); + .buffer_tranche_id = LWTRANCHE_SUBTRANS_BUFFER, + .bank_tranche_id = LWTRANCHE_SUBTRANS_SLRU, + ); } static void diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index 9cd27695787..b8345295f0d 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -804,27 +804,27 @@ AsyncShmemRequest(void *arg) size = add_size(size, offsetof(AsyncQueueControl, backend)); ShmemRequestStruct(&AsyncQueueControlShmemDesc, - .name = "Async Queue Control", - .size = size, - .ptr = (void **) &asyncQueueControl, - ); + .name = "Async Queue Control", + .size = size, + .ptr = (void **) &asyncQueueControl, + ); SimpleLruRequest(&NotifySlruDesc, - .name = "notify", - .Dir = "pg_notify", + .name = "notify", + .Dir = "pg_notify", - /* long segment names are used in order to avoid wraparound */ - .long_segment_names = true, + /* long segment names are used in order to avoid wraparound */ + .long_segment_names = true, - .nslots = notify_buffers, + .nslots = notify_buffers, - .sync_handler = SYNC_HANDLER_NONE, - .PagePrecedes = asyncQueuePagePrecedes, - .errdetail_for_io_error = asyncQueueErrdetailForIoError, + .sync_handler = SYNC_HANDLER_NONE, + .PagePrecedes = asyncQueuePagePrecedes, + .errdetail_for_io_error = asyncQueueErrdetailForIoError, - .buffer_tranche_id = LWTRANCHE_NOTIFY_BUFFER, - .bank_tranche_id = LWTRANCHE_NOTIFY_SLRU, - ); + .buffer_tranche_id = LWTRANCHE_NOTIFY_BUFFER, + .bank_tranche_id = LWTRANCHE_NOTIFY_SLRU, + ); } static void diff --git a/src/backend/storage/aio/aio_init.c b/src/backend/storage/aio/aio_init.c index d3c68d8b04c..9efe53912ec 100644 --- a/src/backend/storage/aio/aio_init.c +++ b/src/backend/storage/aio/aio_init.c @@ -23,16 +23,24 @@ #include "storage/ipc.h" #include "storage/proc.h" #include "storage/shmem.h" +#include "storage/subsystems.h" #include "utils/guc.h" +static void AioShmemRequest(void *arg); +static void AioShmemInit(void *arg); +static void AioShmemAttach(void *arg); -static Size -AioCtlShmemSize(void) -{ - /* pgaio_ctl itself */ - return sizeof(PgAioCtl); -} +const ShmemCallbacks AioShmemCallbacks = { + .request_fn = AioShmemRequest, + .init_fn = AioShmemInit, + .attach_fn = AioShmemAttach, +}; + +static PgAioBackend *AioBackendShmemPtr; +static PgAioHandle *AioHandleShmemPtr; +static struct iovec *AioHandleIOVShmemPtr; +static uint64 *AioHandleDataShmemPtr; static uint32 AioProcs(void) @@ -109,12 +117,21 @@ AioChooseMaxConcurrency(void) return Min(max_proportional_pins, 64); } -Size -AioShmemSize(void) +/* + * Register shared memory area for AIO subsystem. + */ +static void +AioShmemRequest(void *arg) { - Size sz = 0; + static ShmemStructDesc AioCtlShmemDesc; + static ShmemStructDesc AioBackendShmemDesc; + static ShmemStructDesc AioHandleShmemDesc; + static ShmemStructDesc AioHandleIOVShmemDesc; + static ShmemStructDesc AioHandleDataShmemDesc; /* + * Resolve io_max_concurrency if not already done + * * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT. * However, if the DBA explicitly set io_max_concurrency = -1 in the * config file, then PGC_S_DYNAMIC_DEFAULT will fail to override that and @@ -132,48 +149,57 @@ AioShmemSize(void) PGC_S_OVERRIDE); } - sz = add_size(sz, AioCtlShmemSize()); - sz = add_size(sz, AioBackendShmemSize()); - sz = add_size(sz, AioHandleShmemSize()); - sz = add_size(sz, AioHandleIOVShmemSize()); - sz = add_size(sz, AioHandleDataShmemSize()); - - /* Reserve space for method specific resources. */ - if (pgaio_method_ops->shmem_size) - sz = add_size(sz, pgaio_method_ops->shmem_size()); - - return sz; + ShmemRequestStruct(&AioCtlShmemDesc, + .name = "AioCtl", + .size = sizeof(PgAioCtl), + .ptr = (void **) &pgaio_ctl, + ); + + ShmemRequestStruct(&AioBackendShmemDesc, + .name = "AioBackend", + .size = AioBackendShmemSize(), + .ptr = (void **) &AioBackendShmemPtr, + ); + + ShmemRequestStruct(&AioHandleShmemDesc, + .name = "AioHandle", + .size = AioHandleShmemSize(), + .ptr = (void **) &AioHandleShmemPtr, + ); + + ShmemRequestStruct(&AioHandleIOVShmemDesc, + .name = "AioHandleIOV", + .size = AioHandleIOVShmemSize(), + .ptr = (void **) &AioHandleIOVShmemPtr, + ); + + ShmemRequestStruct(&AioHandleDataShmemDesc, + .name = "AioHandleData", + .size = AioHandleDataShmemSize(), + .ptr = (void **) &AioHandleDataShmemPtr, + ); + + if (pgaio_method_ops->shmem_callbacks.request_fn) + pgaio_method_ops->shmem_callbacks.request_fn(pgaio_method_ops->shmem_callbacks.request_fn_arg); } -void -AioShmemInit(void) +/* + * Initialize AIO shared memory during postmaster startup. + */ +static void +AioShmemInit(void *arg) { - bool found; uint32 io_handle_off = 0; uint32 iovec_off = 0; uint32 per_backend_iovecs = io_max_concurrency * io_max_combine_limit; - pgaio_ctl = (PgAioCtl *) - ShmemInitStruct("AioCtl", AioCtlShmemSize(), &found); - - if (found) - goto out; - - memset(pgaio_ctl, 0, AioCtlShmemSize()); - pgaio_ctl->io_handle_count = AioProcs() * io_max_concurrency; pgaio_ctl->iovec_count = AioProcs() * per_backend_iovecs; - pgaio_ctl->backend_state = (PgAioBackend *) - ShmemInitStruct("AioBackend", AioBackendShmemSize(), &found); - - pgaio_ctl->io_handles = (PgAioHandle *) - ShmemInitStruct("AioHandle", AioHandleShmemSize(), &found); - - pgaio_ctl->iovecs = (struct iovec *) - ShmemInitStruct("AioHandleIOV", AioHandleIOVShmemSize(), &found); - pgaio_ctl->handle_data = (uint64 *) - ShmemInitStruct("AioHandleData", AioHandleDataShmemSize(), &found); + pgaio_ctl->backend_state = AioBackendShmemPtr; + pgaio_ctl->io_handles = AioHandleShmemPtr; + pgaio_ctl->iovecs = AioHandleIOVShmemPtr; + pgaio_ctl->handle_data = AioHandleDataShmemPtr; for (int procno = 0; procno < AioProcs(); procno++) { @@ -208,10 +234,15 @@ AioShmemInit(void) } } -out: - /* Initialize IO method specific resources. */ - if (pgaio_method_ops->shmem_init) - pgaio_method_ops->shmem_init(!found); + if (pgaio_method_ops->shmem_callbacks.init_fn) + pgaio_method_ops->shmem_callbacks.init_fn(pgaio_method_ops->shmem_callbacks.init_fn_arg); +} + +static void +AioShmemAttach(void *arg) +{ + if (pgaio_method_ops->shmem_callbacks.attach_fn) + pgaio_method_ops->shmem_callbacks.attach_fn(pgaio_method_ops->shmem_callbacks.attach_fn_arg); } void diff --git a/src/backend/storage/aio/method_io_uring.c b/src/backend/storage/aio/method_io_uring.c index 39984df31b4..82fbbefc0b6 100644 --- a/src/backend/storage/aio/method_io_uring.c +++ b/src/backend/storage/aio/method_io_uring.c @@ -49,8 +49,8 @@ /* Entry points for IoMethodOps. */ -static size_t pgaio_uring_shmem_size(void); -static void pgaio_uring_shmem_init(bool first_time); +static void pgaio_uring_shmem_request(void *arg); +static void pgaio_uring_shmem_init(void *arg); static void pgaio_uring_init_backend(void); static int pgaio_uring_submit(uint16 num_staged_ios, PgAioHandle **staged_ios); static void pgaio_uring_wait_one(PgAioHandle *ioh, uint64 ref_generation); @@ -59,7 +59,6 @@ static void pgaio_uring_check_one(PgAioHandle *ioh, uint64 ref_generation); /* helper functions */ static void pgaio_uring_sq_from_io(PgAioHandle *ioh, struct io_uring_sqe *sqe); - const IoMethodOps pgaio_uring_ops = { /* * While io_uring mostly is OK with FDs getting closed while the IO is in @@ -70,8 +69,8 @@ const IoMethodOps pgaio_uring_ops = { */ .wait_on_fd_before_close = true, - .shmem_size = pgaio_uring_shmem_size, - .shmem_init = pgaio_uring_shmem_init, + .shmem_callbacks.request_fn = pgaio_uring_shmem_request, + .shmem_callbacks.init_fn = pgaio_uring_shmem_init, .init_backend = pgaio_uring_init_backend, .submit = pgaio_uring_submit, @@ -267,23 +266,34 @@ pgaio_uring_shmem_size(void) { size_t sz; + sz = pgaio_uring_context_shmem_size(); + sz = add_size(sz, pgaio_uring_ring_shmem_size()); + + return sz; +} + +static void +pgaio_uring_shmem_request(void *arg) +{ + static ShmemStructDesc AioUringShmemDesc; + /* * Kernel and liburing support for various features influences how much * shmem we need, perform the necessary checks. */ pgaio_uring_check_capabilities(); - sz = pgaio_uring_context_shmem_size(); - sz = add_size(sz, pgaio_uring_ring_shmem_size()); - - return sz; + ShmemRequestStruct(&AioUringShmemDesc, + .name = "AioUringContext", + .size = pgaio_uring_shmem_size(), + .ptr = (void **) &pgaio_uring_contexts, + ); } static void -pgaio_uring_shmem_init(bool first_time) +pgaio_uring_shmem_init(void *arg) { int TotalProcs = pgaio_uring_procs(); - bool found; char *shmem; size_t ring_mem_remain = 0; char *ring_mem_next = 0; @@ -291,13 +301,11 @@ pgaio_uring_shmem_init(bool first_time) /* * We allocate memory for all PgAioUringContext instances and, if * supported, the memory required for each of the io_uring instances, in - * one ShmemInitStruct(). + * one combined allocation. + * + * pgaio_uring_contexts is already set to the base of the allocation. */ - shmem = ShmemInitStruct("AioUringContext", pgaio_uring_shmem_size(), &found); - if (found) - return; - - pgaio_uring_contexts = (PgAioUringContext *) shmem; + shmem = (char *) pgaio_uring_contexts; shmem += pgaio_uring_context_shmem_size(); /* if supported, handle memory alignment / sizing for io_uring memory */ diff --git a/src/backend/storage/aio/method_worker.c b/src/backend/storage/aio/method_worker.c index efe38e9f113..7697ae34d66 100644 --- a/src/backend/storage/aio/method_worker.c +++ b/src/backend/storage/aio/method_worker.c @@ -41,6 +41,7 @@ #include "storage/ipc.h" #include "storage/latch.h" #include "storage/proc.h" +#include "storage/shmem.h" #include "tcop/tcopprot.h" #include "utils/injection_point.h" #include "utils/memdebug.h" @@ -73,16 +74,20 @@ typedef struct PgAioWorkerControl } PgAioWorkerControl; -static size_t pgaio_worker_shmem_size(void); -static void pgaio_worker_shmem_init(bool first_time); +static void pgaio_worker_shmem_request(void *arg); +static void pgaio_worker_shmem_init(void *arg); +static void pgaio_worker_shmem_attach(void *arg); + +static PgAioWorkerSubmissionQueue *io_worker_submission_queue; static bool pgaio_worker_needs_synchronous_execution(PgAioHandle *ioh); static int pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios); const IoMethodOps pgaio_worker_ops = { - .shmem_size = pgaio_worker_shmem_size, - .shmem_init = pgaio_worker_shmem_init, + .shmem_callbacks.request_fn = pgaio_worker_shmem_request, + .shmem_callbacks.init_fn = pgaio_worker_shmem_init, + .shmem_callbacks.attach_fn = pgaio_worker_shmem_attach, .needs_synchronous_execution = pgaio_worker_needs_synchronous_execution, .submit = pgaio_worker_submit, @@ -95,7 +100,6 @@ int io_workers = 3; static int io_worker_queue_size = 64; static int MyIoWorkerId; -static PgAioWorkerSubmissionQueue *io_worker_submission_queue; static PgAioWorkerControl *io_worker_control; @@ -116,50 +120,62 @@ pgaio_worker_control_shmem_size(void) sizeof(PgAioWorkerSlot) * MAX_IO_WORKERS; } -static size_t -pgaio_worker_shmem_size(void) +/* + * Set secondary AIO worker pointer from the combined allocation. + */ +static void +pgaio_worker_set_secondary_ptr(void) { - size_t sz; int queue_size; + Size queue_sz = pgaio_worker_queue_shmem_size(&queue_size); - sz = pgaio_worker_queue_shmem_size(&queue_size); - sz = add_size(sz, pgaio_worker_control_shmem_size()); - - return sz; + io_worker_control = (PgAioWorkerControl *) + ((char *) io_worker_submission_queue + MAXALIGN(queue_sz)); } static void -pgaio_worker_shmem_init(bool first_time) +pgaio_worker_shmem_init(void *arg) { - bool found; int queue_size; - io_worker_submission_queue = - ShmemInitStruct("AioWorkerSubmissionQueue", - pgaio_worker_queue_shmem_size(&queue_size), - &found); - if (!found) - { - io_worker_submission_queue->size = queue_size; - io_worker_submission_queue->head = 0; - io_worker_submission_queue->tail = 0; - } + pgaio_worker_queue_shmem_size(&queue_size); + io_worker_submission_queue->size = queue_size; + io_worker_submission_queue->head = 0; + io_worker_submission_queue->tail = 0; - io_worker_control = - ShmemInitStruct("AioWorkerControl", - pgaio_worker_control_shmem_size(), - &found); - if (!found) + pgaio_worker_set_secondary_ptr(); + + io_worker_control->idle_worker_mask = 0; + for (int i = 0; i < MAX_IO_WORKERS; ++i) { - io_worker_control->idle_worker_mask = 0; - for (int i = 0; i < MAX_IO_WORKERS; ++i) - { - io_worker_control->workers[i].latch = NULL; - io_worker_control->workers[i].in_use = false; - } + io_worker_control->workers[i].latch = NULL; + io_worker_control->workers[i].in_use = false; } } +static void +pgaio_worker_shmem_attach(void *arg) +{ + pgaio_worker_set_secondary_ptr(); +} + +static void +pgaio_worker_shmem_request(void *arg) +{ + static ShmemStructDesc AioWorkerShmemDesc; + size_t size; + int queue_size; + + size = MAXALIGN(pgaio_worker_queue_shmem_size(&queue_size)) + + pgaio_worker_control_shmem_size(); + + ShmemRequestStruct(&AioWorkerShmemDesc, + .name = "AioWorkerSubmissionQueue", + .size = size, + .ptr = (void **) &io_worker_submission_queue, + ); +} + static int pgaio_worker_choose_idle(void) { diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index faaf9c471f2..bd67f81cea3 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -122,7 +122,6 @@ CalculateShmemSize(void) size = add_size(size, WaitEventCustomShmemSize()); size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); - size = add_size(size, AioShmemSize()); size = add_size(size, WaitLSNShmemSize()); size = add_size(size, LogicalDecodingCtlShmemSize()); size = add_size(size, DataChecksumsShmemSize()); @@ -301,7 +300,6 @@ CreateOrAttachShmemStructs(void) StatsShmemInit(); WaitEventCustomShmemInit(); InjectionPointShmemInit(); - AioShmemInit(); WaitLSNShmemInit(); LogicalDecodingCtlShmemInit(); } diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c index 02dbbf30950..f1d38338c83 100644 --- a/src/backend/storage/lmgr/predicate.c +++ b/src/backend/storage/lmgr/predicate.c @@ -1149,15 +1149,14 @@ PredicateLockShmemRequest(void *arg) * per-predicate-lock-target information. */ ShmemRequestHash(&PredicateLockTargetHashDesc, - .name = "PREDICATELOCKTARGET hash", - .nelems = max_predicate_lock_targets, - - .ptr = &PredicateLockTargetHash, - .hash_info.keysize = sizeof(PREDICATELOCKTARGETTAG), - .hash_info.entrysize = sizeof(PREDICATELOCKTARGET), - .hash_info.num_partitions = NUM_PREDICATELOCK_PARTITIONS, - .hash_flags = HASH_ELEM | HASH_BLOBS | HASH_PARTITION | HASH_FIXED_SIZE, - ); + .name = "PREDICATELOCKTARGET hash", + .nelems = max_predicate_lock_targets, + .ptr = &PredicateLockTargetHash, + .hash_info.keysize = sizeof(PREDICATELOCKTARGETTAG), + .hash_info.entrysize = sizeof(PREDICATELOCKTARGET), + .hash_info.num_partitions = NUM_PREDICATELOCK_PARTITIONS, + .hash_flags = HASH_ELEM | HASH_BLOBS | HASH_PARTITION | HASH_FIXED_SIZE, + ); /* * Allocate hash table for PREDICATELOCK structs. This stores per @@ -1168,16 +1167,14 @@ PredicateLockShmemRequest(void *arg) max_predicate_locks = max_predicate_lock_targets * 2; ShmemRequestHash(&PredicateLockHashDesc, - .name = "PREDICATELOCK hash", - - .nelems = max_predicate_locks, - - .ptr = &PredicateLockHash, - .hash_info.keysize = sizeof(PREDICATELOCKTAG), - .hash_info.entrysize = sizeof(PREDICATELOCK), - .hash_info.hash = predicatelock_hash, - .hash_info.num_partitions = NUM_PREDICATELOCK_PARTITIONS, - .hash_flags = HASH_ELEM | HASH_FUNCTION | HASH_PARTITION | HASH_FIXED_SIZE, + .name = "PREDICATELOCK hash", + .nelems = max_predicate_locks, + .ptr = &PredicateLockHash, + .hash_info.keysize = sizeof(PREDICATELOCKTAG), + .hash_info.entrysize = sizeof(PREDICATELOCK), + .hash_info.hash = predicatelock_hash, + .hash_info.num_partitions = NUM_PREDICATELOCK_PARTITIONS, + .hash_flags = HASH_ELEM | HASH_FUNCTION | HASH_PARTITION | HASH_FIXED_SIZE, ); /* @@ -1195,11 +1192,11 @@ PredicateLockShmemRequest(void *arg) * predicate locking. */ ShmemRequestStruct(&PredXactListShmemDesc, - .name = "PredXactList", - .size = add_size(PredXactListDataSize, - (mul_size((Size) max_serializable_xacts, - sizeof(SERIALIZABLEXACT)))), - .ptr = (void **) &PredXact, + .name = "PredXactList", + .size = add_size(PredXactListDataSize, + (mul_size((Size) max_serializable_xacts, + sizeof(SERIALIZABLEXACT)))), + .ptr = (void **) &PredXact, ); /* @@ -1229,45 +1226,45 @@ PredicateLockShmemRequest(void *arg) max_rw_conflicts = max_serializable_xacts * 5; ShmemRequestStruct(&RWConflictPoolShmemDesc, - .name = "RWConflictPool", - .size = RWConflictPoolHeaderDataSize + mul_size((Size) max_rw_conflicts, - RWConflictDataSize), - .ptr = (void **) &RWConflictPool, - ); + .name = "RWConflictPool", + .size = RWConflictPoolHeaderDataSize + mul_size((Size) max_rw_conflicts, + RWConflictDataSize), + .ptr = (void **) &RWConflictPool, + ); ShmemRequestStruct(&FinishedSerializableShmemDesc, - .name = "FinishedSerializableTransactions", - .size = sizeof(dlist_head), - .ptr = (void **) &FinishedSerializableTransactions, - ); + .name = "FinishedSerializableTransactions", + .size = sizeof(dlist_head), + .ptr = (void **) &FinishedSerializableTransactions, + ); /* * Initialize the SLRU storage for old committed serializable * transactions. */ SimpleLruRequest(&SerialSlruDesc, - .name = "serializable", - .Dir = "pg_serial", - .long_segment_names = false, + .name = "serializable", + .Dir = "pg_serial", + .long_segment_names = false, - .nslots = serializable_buffers, + .nslots = serializable_buffers, - .sync_handler = SYNC_HANDLER_NONE, - .PagePrecedes = SerialPagePrecedesLogically, - .errdetail_for_io_error = serial_errdetail_for_io_error, + .sync_handler = SYNC_HANDLER_NONE, + .PagePrecedes = SerialPagePrecedesLogically, + .errdetail_for_io_error = serial_errdetail_for_io_error, - .buffer_tranche_id = LWTRANCHE_SERIAL_BUFFER, - .bank_tranche_id = LWTRANCHE_SERIAL_SLRU, + .buffer_tranche_id = LWTRANCHE_SERIAL_BUFFER, + .bank_tranche_id = LWTRANCHE_SERIAL_SLRU, ); #ifdef USE_ASSERT_CHECKING SerialPagePrecedesLogicallyUnitTests(); #endif ShmemRequestStruct(&SerialControlShmemDesc, - .name = "SerialControlData", - .size = sizeof(SerialControlData), - .ptr = (void **) &serialControl, - ); + .name = "SerialControlData", + .size = sizeof(SerialControlData), + .ptr = (void **) &serialControl, + ); } static void diff --git a/src/include/access/slru.h b/src/include/access/slru.h index 820c7986854..1dbb0b62525 100644 --- a/src/include/access/slru.h +++ b/src/include/access/slru.h @@ -169,7 +169,7 @@ typedef struct SlruOpts */ int buffer_tranche_id; int bank_tranche_id; -} SlruOpts; +} SlruOpts; /* * SlruDesc is an unshared structure that points to the active information @@ -179,7 +179,7 @@ typedef struct SlruDesc { ShmemStructDesc base; - SlruOpts options; + SlruOpts options; SlruShared shared; @@ -203,7 +203,7 @@ SimpleLruGetBankLock(SlruDesc *ctl, int64 pageno) return &(ctl->shared->bank_locks[bankno].lock); } -extern void SimpleLruRequestWithOpts(SlruDesc *desc, const SlruOpts *options); +extern void SimpleLruRequestWithOpts(SlruDesc *desc, const SlruOpts * options); #define SimpleLruRequest(desc, ...) \ SimpleLruRequestWithOpts(desc, &(SlruOpts){__VA_ARGS__}) diff --git a/src/include/storage/aio_internal.h b/src/include/storage/aio_internal.h index 33e1e2dc048..9ca4087aa7f 100644 --- a/src/include/storage/aio_internal.h +++ b/src/include/storage/aio_internal.h @@ -20,6 +20,8 @@ #include "port/pg_iovec.h" #include "storage/aio.h" #include "storage/condition_variable.h" +#include "storage/ipc.h" +#include "storage/shmem.h" /* @@ -267,20 +269,8 @@ typedef struct IoMethodOps */ bool wait_on_fd_before_close; - /* global initialization */ - - /* - * Amount of additional shared memory to reserve for the io_method. Called - * just like a normal ipci.c style *Size() function. Optional. - */ - size_t (*shmem_size) (void); - - /* - * Initialize shared memory. First time is true if AIO's shared memory was - * just initialized, false otherwise. Optional. - */ - void (*shmem_init) (bool first_time); + ShmemCallbacks shmem_callbacks; /* * Per-backend initialization. Optional. diff --git a/src/include/storage/aio_subsys.h b/src/include/storage/aio_subsys.h index 276cb3e31c4..dd54869351f 100644 --- a/src/include/storage/aio_subsys.h +++ b/src/include/storage/aio_subsys.h @@ -20,12 +20,8 @@ /* aio_init.c */ -extern Size AioShmemSize(void); -extern void AioShmemInit(void); - extern void pgaio_init_backend(void); - /* aio.c */ extern void pgaio_error_cleanup(void); extern void AtEOXact_Aio(bool is_commit); diff --git a/src/include/storage/subsystemlist.h b/src/include/storage/subsystemlist.h index c199f18a27a..b438794d46d 100644 --- a/src/include/storage/subsystemlist.h +++ b/src/include/storage/subsystemlist.h @@ -53,3 +53,6 @@ PG_SHMEM_SUBSYSTEM(ProcSignalShmemCallbacks) /* other modules that need some shared memory space */ PG_SHMEM_SUBSYSTEM(AsyncShmemCallbacks) + +/* AIO subsystem. This delegates to the method-specific callbacks */ +PG_SHMEM_SUBSYSTEM(AioShmemCallbacks) diff --git a/src/test/modules/test_slru/test_slru.c b/src/test/modules/test_slru/test_slru.c index 3c2a143b4d5..6bd1bec72c5 100644 --- a/src/test/modules/test_slru/test_slru.c +++ b/src/test/modules/test_slru/test_slru.c @@ -234,24 +234,24 @@ static void test_slru_shmem_request(void *arg) { SimpleLruRequest(&TestSlruDesc, - .name = "TestSLRU", - .Dir = TestSlruDir, - - /* - * Short segments names are well tested elsewhere so in this test we are - * focusing on long names. - */ - .long_segment_names = true, - - .nslots = NUM_TEST_BUFFERS, - .nlsns = 0, - - .sync_handler = SYNC_HANDLER_NONE, - .PagePrecedes = test_slru_page_precedes_logically, - .errdetail_for_io_error = test_slru_errdetail_for_io_error, - - /* let slru.c assign these */ - .buffer_tranche_id = 0, - .bank_tranche_id = 0, - ); + .name = "TestSLRU", + .Dir = TestSlruDir, + + /* + * Short segments names are well tested elsewhere so in this test we are + * focusing on long names. + */ + .long_segment_names = true, + + .nslots = NUM_TEST_BUFFERS, + .nlsns = 0, + + .sync_handler = SYNC_HANDLER_NONE, + .PagePrecedes = test_slru_page_precedes_logically, + .errdetail_for_io_error = test_slru_errdetail_for_io_error, + + /* let slru.c assign these */ + .buffer_tranche_id = 0, + .bank_tranche_id = 0, + ); } -- 2.47.3