src/backend/storage/aio/aio_init.c - master - githubmirror/postgres

src/backend/storage/aio/aio_init.c
255 строк7 KB
Peter Eisentraut
Fix for loop variables
11 июл 2026, 15:39
11 июл 2026, 15:39e615da8
/*-------------------------------------------------------------------------
 *
 * aio_init.c
 *    AIO - Subsystem Initialization
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *    src/backend/storage/aio/aio_init.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "miscadmin.h"
#include "storage/aio.h"
#include "storage/aio_internal.h"
#include "storage/aio_subsys.h"
#include "storage/bufmgr.h"
#include "storage/io_worker.h"
#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);

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)
{
	/*
	 * While AIO workers don't need their own AIO context, we can't currently
	 * guarantee that nothing gets assigned to an IO worker's ProcNumber if we
	 * just subtracted MAX_IO_WORKERS.
	 */
	return MaxBackends + NUM_AUXILIARY_PROCS;
}

static Size
AioBackendShmemSize(void)
{
	return mul_size(AioProcs(), sizeof(PgAioBackend));
}

static Size
AioHandleShmemSize(void)
{
	Size		sz;

	/* verify AioChooseMaxConcurrency() did its thing */
	Assert(io_max_concurrency > 0);

	/* io handles */
	sz = mul_size(AioProcs(),
				  mul_size(io_max_concurrency, sizeof(PgAioHandle)));

	return sz;
}

static Size
AioHandleIOVShmemSize(void)
{
	/* each IO handle can have up to io_max_combine_limit iovec objects */
	return mul_size(sizeof(struct iovec),
					mul_size(mul_size(io_max_combine_limit, AioProcs()),
							 io_max_concurrency));
}

static Size
AioHandleDataShmemSize(void)
{
	/* each buffer referenced by an iovec can have associated data */
	return mul_size(sizeof(uint64),
					mul_size(mul_size(io_max_combine_limit, AioProcs()),
							 io_max_concurrency));
}

/*
 * Choose a suitable value for io_max_concurrency.
 *
 * It's unlikely that we could have more IOs in flight than buffers that we
 * would be allowed to pin.
 *
 * On the upper end, apply a cap too - just because shared_buffers is large,
 * it doesn't make sense have millions of buffers undergo IO concurrently.
 */
static int
AioChooseMaxConcurrency(void)
{
	uint32		max_backends;
	int			max_proportional_pins;

	/* Similar logic to LimitAdditionalPins() */
	max_backends = MaxBackends + NUM_AUXILIARY_PROCS;
	max_proportional_pins = NBuffers / max_backends;

	max_proportional_pins = Max(max_proportional_pins, 1);

	/* apply upper limit */
	return Min(max_proportional_pins, 64);
}

/*
 * Register AIO subsystem's shared memory needs.
 */
static void
AioShmemRequest(void *arg)
{
	/*
	 * 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
	 * we must force the matter with PGC_S_OVERRIDE.
	 */
	if (io_max_concurrency == -1)
	{
		char		buf[32];

		snprintf(buf, sizeof(buf), "%d", AioChooseMaxConcurrency());
		SetConfigOption("io_max_concurrency", buf, PGC_POSTMASTER,
						PGC_S_DYNAMIC_DEFAULT);
		if (io_max_concurrency == -1)	/* failed to apply it? */
			SetConfigOption("io_max_concurrency", buf, PGC_POSTMASTER,
							PGC_S_OVERRIDE);
	}

	ShmemRequestStruct(.name = "AioCtl",
					   .size = sizeof(PgAioCtl),
					   .ptr = (void **) &pgaio_ctl,
		);

	ShmemRequestStruct(.name = "AioBackend",
					   .size = AioBackendShmemSize(),
					   .ptr = (void **) &AioBackendShmemPtr,
		);

	ShmemRequestStruct(.name = "AioHandle",
					   .size = AioHandleShmemSize(),
					   .ptr = (void **) &AioHandleShmemPtr,
		);

	ShmemRequestStruct(.name = "AioHandleIOV",
					   .size = AioHandleIOVShmemSize(),
					   .ptr = (void **) &AioHandleIOVShmemPtr,
		);

	ShmemRequestStruct(.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.opaque_arg);
}

/*
 * Initialize AIO shared memory during postmaster startup.
 */
static void
AioShmemInit(void *arg)
{
	uint32		io_handle_off = 0;
	uint32		iovec_off = 0;
	uint32		per_backend_iovecs = io_max_concurrency * io_max_combine_limit;

	pgaio_ctl->io_handle_count = AioProcs() * io_max_concurrency;
	pgaio_ctl->iovec_count = AioProcs() * per_backend_iovecs;

	pgaio_ctl->backend_state = AioBackendShmemPtr;
	pgaio_ctl->io_handles = AioHandleShmemPtr;
	pgaio_ctl->iovecs = AioHandleIOVShmemPtr;
	pgaio_ctl->handle_data = AioHandleDataShmemPtr;

	for (uint32 procno = 0; procno < AioProcs(); procno++)
	{
		PgAioBackend *bs = &pgaio_ctl->backend_state[procno];

		bs->io_handle_off = io_handle_off;
		io_handle_off += io_max_concurrency;

		dclist_init(&bs->idle_ios);
		memset(bs->staged_ios, 0, sizeof(PgAioHandle *) * PGAIO_SUBMIT_BATCH_SIZE);
		dclist_init(&bs->in_flight_ios);

		/* initialize per-backend IOs */
		for (int i = 0; i < io_max_concurrency; i++)
		{
			PgAioHandle *ioh = &pgaio_ctl->io_handles[bs->io_handle_off + i];

			ioh->generation = 1;
			ioh->owner_procno = procno;
			ioh->iovec_off = iovec_off;
			ioh->handle_data_len = 0;
			ioh->report_return = NULL;
			ioh->resowner = NULL;
			ioh->num_callbacks = 0;
			ioh->distilled_result.status = PGAIO_RS_UNKNOWN;
			ioh->flags = 0;

			ConditionVariableInit(&ioh->cv);

			dclist_push_tail(&bs->idle_ios, &ioh->node);
			iovec_off += io_max_combine_limit;
		}
	}

	if (pgaio_method_ops->shmem_callbacks.init_fn)
		pgaio_method_ops->shmem_callbacks.init_fn(pgaio_method_ops->shmem_callbacks.opaque_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.opaque_arg);
}

void
pgaio_init_backend(void)
{
	/* shouldn't be initialized twice */
	Assert(!pgaio_my_backend);

	if (MyBackendType == B_IO_WORKER)
		return;

	if (MyProc == NULL || MyProcNumber >= AioProcs())
		elog(ERROR, "aio requires a normal PGPROC");

	pgaio_my_backend = &pgaio_ctl->backend_state[MyProcNumber];

	if (pgaio_method_ops->init_backend)
		pgaio_method_ops->init_backend();

	before_shmem_exit(pgaio_shutdown, 0);
}