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

src/backend/storage/aio/method_worker.c
1 065 строк29 KB
Michael Paquier
Emit a warning when io_min_workers exceeds io_max_workers
06 июл 2026, 05:37
06 июл 2026, 05:379d1188f
/*-------------------------------------------------------------------------
 *
 * method_worker.c
 *    AIO - perform AIO using worker processes
 *
 * IO workers consume IOs from a shared memory submission queue, run
 * traditional synchronous system calls, and perform the shared completion
 * handling immediately.  Client code submits most requests by pushing IOs
 * into the submission queue, and waits (if necessary) using condition
 * variables.  Some IOs cannot be performed in another process due to lack of
 * infrastructure for reopening the file, and must processed synchronously by
 * the client code when submitted.
 *
 * The pool of workers tries to stabilize at a size that can handle recently
 * seen variation in demand, within the configured limits.
 *
 * This method of AIO is available in all builds on all operating systems, and
 * is the default.
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/storage/aio/method_worker.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <limits.h>

#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "port/pg_bitutils.h"
#include "postmaster/auxprocess.h"
#include "postmaster/interrupt.h"
#include "storage/aio.h"
#include "storage/aio_internal.h"
#include "storage/aio_subsys.h"
#include "storage/io_worker.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "storage/shmem.h"
#include "tcop/tcopprot.h"
#include "utils/injection_point.h"
#include "utils/memdebug.h"
#include "utils/ps_status.h"
#include "utils/wait_event.h"

/*
 * Saturation for counters used to estimate wakeup:IO ratio.
 *
 * We maintain hist_wakeups for wakeups received and hist_ios for IOs
 * processed by each worker.  When either counter reaches this saturation
 * value, we divide both by two.  The result is an exponentially decaying
 * ratio of wakeups to IOs, with a very short memory.
 *
 * If a worker is itself experiencing useless wakeups, it assumes that
 * higher-numbered workers would experience even more, so it should end the
 * chain.
 */
#define PGAIO_WORKER_WAKEUP_RATIO_SATURATE 4

/* Debugging support: show current IO and wakeups:ios statistics in ps. */
/* #define PGAIO_WORKER_SHOW_PS_INFO */

typedef struct PgAioWorkerSubmissionQueue
{
	uint32		size;
	uint32		head;
	uint32		tail;
	int			sqes[FLEXIBLE_ARRAY_MEMBER];
} PgAioWorkerSubmissionQueue;

typedef struct PgAioWorkerSlot
{
	ProcNumber	proc_number;
} PgAioWorkerSlot;

/*
 * Sets of worker IDs are held in a simple bitmap, accessed through functions
 * that provide a more readable abstraction.  If we wanted to support more
 * workers than that, the contention on the single queue would surely get too
 * high, so we might want to consider multiple pools instead of widening this.
 */
typedef uint64 PgAioWorkerSet;

#define PGAIO_WORKERSET_BITS (sizeof(PgAioWorkerSet) * CHAR_BIT)

static_assert(PGAIO_WORKERSET_BITS >= MAX_IO_WORKERS, "too small");

typedef struct PgAioWorkerControl
{
	/* Seen by postmaster */
	bool		grow;
	bool		grow_signal_sent;

	/* Protected by AioWorkerSubmissionQueueLock. */
	PgAioWorkerSet idle_workerset;

	/* Protected by AioWorkerControlLock. */
	PgAioWorkerSet workerset;
	int			nworkers;

	/* Protected by AioWorkerControlLock. */
	PgAioWorkerSlot workers[FLEXIBLE_ARRAY_MEMBER];
} PgAioWorkerControl;


static void pgaio_worker_shmem_request(void *arg);
static void pgaio_worker_shmem_init(void *arg);

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_callbacks.request_fn = pgaio_worker_shmem_request,
	.shmem_callbacks.init_fn = pgaio_worker_shmem_init,

	.needs_synchronous_execution = pgaio_worker_needs_synchronous_execution,
	.submit = pgaio_worker_submit,
};


/* GUCs */
int			io_min_workers = 2;
int			io_max_workers = 8;
int			io_worker_idle_timeout = 60000;
int			io_worker_launch_interval = 100;


static int	io_worker_queue_size = 64;
static int	MyIoWorkerId = -1;
static PgAioWorkerSubmissionQueue *io_worker_submission_queue;
static PgAioWorkerControl *io_worker_control;


static void
pgaio_workerset_initialize(PgAioWorkerSet *set)
{
	*set = 0;
}

static bool
pgaio_workerset_is_empty(PgAioWorkerSet *set)
{
	return *set == 0;
}

static PgAioWorkerSet
pgaio_workerset_singleton(int worker)
{
	Assert(worker >= 0 && worker < MAX_IO_WORKERS);
	return UINT64_C(1) << worker;
}

static void
pgaio_workerset_all(PgAioWorkerSet *set)
{
	*set = UINT64_MAX >> (PGAIO_WORKERSET_BITS - MAX_IO_WORKERS);
}

static void
pgaio_workerset_subtract(PgAioWorkerSet *set1, const PgAioWorkerSet *set2)
{
	*set1 &= ~*set2;
}

static void
pgaio_workerset_insert(PgAioWorkerSet *set, int worker)
{
	Assert(worker >= 0 && worker < MAX_IO_WORKERS);
	*set |= pgaio_workerset_singleton(worker);
}

static void
pgaio_workerset_remove(PgAioWorkerSet *set, int worker)
{
	Assert(worker >= 0 && worker < MAX_IO_WORKERS);
	*set &= ~pgaio_workerset_singleton(worker);
}

static void
pgaio_workerset_remove_lte(PgAioWorkerSet *set, int worker)
{
	Assert(worker >= 0 && worker < MAX_IO_WORKERS);
	*set &= (~(PgAioWorkerSet) 0) << (worker + 1);
}

static int
pgaio_workerset_get_highest(PgAioWorkerSet *set)
{
	Assert(!pgaio_workerset_is_empty(set));
	return pg_leftmost_one_pos64(*set);
}

static int
pgaio_workerset_get_lowest(PgAioWorkerSet *set)
{
	Assert(!pgaio_workerset_is_empty(set));
	return pg_rightmost_one_pos64(*set);
}

static int
pgaio_workerset_pop_lowest(PgAioWorkerSet *set)
{
	int			worker = pgaio_workerset_get_lowest(set);

	pgaio_workerset_remove(set, worker);
	return worker;
}

#ifdef USE_ASSERT_CHECKING
static bool
pgaio_workerset_contains(PgAioWorkerSet *set, int worker)
{
	Assert(worker >= 0 && worker < MAX_IO_WORKERS);
	return (*set & pgaio_workerset_singleton(worker)) != 0;
}

static int
pgaio_workerset_count(PgAioWorkerSet *set)
{
	return pg_popcount64(*set);
}
#endif

static void
pgaio_worker_shmem_request(void *arg)
{
	size_t		size;
	int			queue_size;

	/* Round size up to next power of two so we can make a mask. */
	queue_size = pg_nextpower2_32(io_worker_queue_size);

	size = offsetof(PgAioWorkerSubmissionQueue, sqes) + sizeof(int) * queue_size;
	ShmemRequestStruct(.name = "AioWorkerSubmissionQueue",
					   .size = size,
					   .ptr = (void **) &io_worker_submission_queue,
		);

	size = offsetof(PgAioWorkerControl, workers) + sizeof(PgAioWorkerSlot) * MAX_IO_WORKERS;
	ShmemRequestStruct(.name = "AioWorkerControl",
					   .size = size,
					   .ptr = (void **) &io_worker_control,
		);
}

static void
pgaio_worker_shmem_init(void *arg)
{
	int			queue_size;

	/* Round size up like in pgaio_worker_shmem_request() */
	queue_size = pg_nextpower2_32(io_worker_queue_size);

	io_worker_submission_queue->size = queue_size;
	io_worker_submission_queue->head = 0;
	io_worker_submission_queue->tail = 0;
	io_worker_control->grow = false;
	pgaio_workerset_initialize(&io_worker_control->workerset);
	pgaio_workerset_initialize(&io_worker_control->idle_workerset);

	for (int i = 0; i < MAX_IO_WORKERS; ++i)
		io_worker_control->workers[i].proc_number = INVALID_PROC_NUMBER;
}

/*
 * Tell postmaster that we think a new worker is needed.
 */
static void
pgaio_worker_request_grow(void)
{
	/*
	 * Suppress useless signaling if we already know that we're at the
	 * maximum.  This uses an unlocked read of nworkers, but that's OK for
	 * this heuristic purpose.
	 */
	if (io_worker_control->nworkers >= io_max_workers)
		return;

	/* Already requested? */
	if (io_worker_control->grow)
		return;

	io_worker_control->grow = true;
	pg_memory_barrier();

	/*
	 * If the postmaster has already been signaled, don't do it again until
	 * the postmaster clears this flag.  There is no point in repeated signals
	 * if grow is being set and cleared repeatedly while the postmaster is
	 * waiting for io_worker_launch_interval, which it applies even to
	 * canceled requests.
	 */
	if (io_worker_control->grow_signal_sent)
		return;

	io_worker_control->grow_signal_sent = true;
	pg_memory_barrier();
	SendPostmasterSignal(PMSIGNAL_IO_WORKER_GROW);
}

/*
 * Cancel any request for a new worker, after observing an empty queue.
 */
static void
pgaio_worker_cancel_grow(void)
{
	if (!io_worker_control->grow)
		return;

	io_worker_control->grow = false;
	pg_memory_barrier();
}

/*
 * Called by the postmaster to check if a new worker has been requested (but
 * possibly canceled since).
 */
bool
pgaio_worker_pm_test_grow_signal_sent(void)
{
	pg_memory_barrier();
	return io_worker_control && io_worker_control->grow_signal_sent;
}

/*
 * Called by the postmaster to check if a new worker has been requested and
 * not canceled since.
 */
bool
pgaio_worker_pm_test_grow(void)
{
	pg_memory_barrier();
	return io_worker_control && io_worker_control->grow;
}

/*
 * Called by the postmaster to clear the request for a new worker.
 */
void
pgaio_worker_pm_clear_grow_signal_sent(void)
{
	if (!io_worker_control)
		return;

	io_worker_control->grow = false;
	io_worker_control->grow_signal_sent = false;
	pg_memory_barrier();
}

static int
pgaio_worker_choose_idle(int only_workers_above)
{
	PgAioWorkerSet workerset;
	int			worker;

	Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));

	workerset = io_worker_control->idle_workerset;
	if (only_workers_above >= 0)
		pgaio_workerset_remove_lte(&workerset, only_workers_above);
	if (pgaio_workerset_is_empty(&workerset))
		return -1;

	/* Find the lowest numbered idle worker and mark it not idle. */
	worker = pgaio_workerset_get_lowest(&workerset);
	pgaio_workerset_remove(&io_worker_control->idle_workerset, worker);

	return worker;
}

/*
 * Try to wake a worker by setting its latch, to tell it there are IOs to
 * process in the submission queue.
 */
static void
pgaio_worker_wake(int worker)
{
	ProcNumber	proc_number;

	/*
	 * If the selected worker is concurrently exiting, then pgaio_worker_die()
	 * had not yet removed it as of when we saw it in idle_workerset.  That's
	 * OK, because it will wake all remaining workers to close wakeup-vs-exit
	 * races: *someone* will see the queued IO.  If there are no workers
	 * running, the postmaster will start a new one.
	 */
	proc_number = io_worker_control->workers[worker].proc_number;
	if (proc_number != INVALID_PROC_NUMBER)
		SetLatch(&GetPGProcByNumber(proc_number)->procLatch);
}

/*
 * Try to wake a set of workers.  Used on pool change, to close races
 * described in the callers.
 */
static void
pgaio_workerset_wake(PgAioWorkerSet workerset)
{
	while (!pgaio_workerset_is_empty(&workerset))
		pgaio_worker_wake(pgaio_workerset_pop_lowest(&workerset));
}

static bool
pgaio_worker_submission_queue_insert(PgAioHandle *ioh)
{
	PgAioWorkerSubmissionQueue *queue;
	uint32		new_head;

	Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));

	queue = io_worker_submission_queue;
	new_head = (queue->head + 1) & (queue->size - 1);
	if (new_head == queue->tail)
	{
		pgaio_debug(DEBUG3, "io queue is full, at %u elements",
					io_worker_submission_queue->size);
		return false;			/* full */
	}

	queue->sqes[queue->head] = pgaio_io_get_id(ioh);
	queue->head = new_head;

	return true;
}

static int
pgaio_worker_submission_queue_consume(void)
{
	PgAioWorkerSubmissionQueue *queue;
	int			result;

	Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));

	queue = io_worker_submission_queue;
	if (queue->tail == queue->head)
		return -1;				/* empty */

	result = queue->sqes[queue->tail];
	queue->tail = (queue->tail + 1) & (queue->size - 1);

	return result;
}

static uint32
pgaio_worker_submission_queue_depth(void)
{
	uint32		head;
	uint32		tail;

	Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));

	head = io_worker_submission_queue->head;
	tail = io_worker_submission_queue->tail;

	if (tail > head)
		head += io_worker_submission_queue->size;

	Assert(head >= tail);

	return head - tail;
}

static bool
pgaio_worker_needs_synchronous_execution(PgAioHandle *ioh)
{
	return
		!IsUnderPostmaster
		|| ioh->flags & PGAIO_HF_REFERENCES_LOCAL
		|| !pgaio_io_can_reopen(ioh);
}

static int
pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios)
{
	PgAioHandle **synchronous_ios = NULL;
	int			nsync = 0;
	int			worker = -1;

	Assert(num_staged_ios <= PGAIO_SUBMIT_BATCH_SIZE);

	for (int i = 0; i < num_staged_ios; i++)
		pgaio_io_prepare_submit(staged_ios[i]);

	if (LWLockConditionalAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE))
	{
		for (int i = 0; i < num_staged_ios; ++i)
		{
			Assert(!pgaio_worker_needs_synchronous_execution(staged_ios[i]));
			if (!pgaio_worker_submission_queue_insert(staged_ios[i]))
			{
				/*
				 * Do the rest synchronously. If the queue is full, give up
				 * and do the rest synchronously. We're holding an exclusive
				 * lock on the queue so nothing can consume entries.
				 */
				synchronous_ios = &staged_ios[i];
				nsync = (num_staged_ios - i);

				break;
			}
		}
		/* Choose one worker to wake for this batch. */
		worker = pgaio_worker_choose_idle(-1);
		LWLockRelease(AioWorkerSubmissionQueueLock);

		/* Wake up chosen worker.  It will wake peers if necessary. */
		if (worker != -1)
			pgaio_worker_wake(worker);
	}
	else
	{
		/* do everything synchronously, no wakeup needed */
		synchronous_ios = staged_ios;
		nsync = num_staged_ios;
	}

	/* Run whatever is left synchronously. */
	if (nsync > 0)
	{
		for (int i = 0; i < nsync; ++i)
		{
			pgaio_io_perform_synchronously(synchronous_ios[i]);
		}
	}

	return num_staged_ios;
}

/*
 * on_shmem_exit() callback that releases the worker's slot in
 * io_worker_control.
 */
static void
pgaio_worker_die(int code, Datum arg)
{
	PgAioWorkerSet notify_set;

	LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
	pgaio_workerset_remove(&io_worker_control->idle_workerset, MyIoWorkerId);
	LWLockRelease(AioWorkerSubmissionQueueLock);

	LWLockAcquire(AioWorkerControlLock, LW_EXCLUSIVE);
	Assert(io_worker_control->workers[MyIoWorkerId].proc_number == MyProcNumber);
	io_worker_control->workers[MyIoWorkerId].proc_number = INVALID_PROC_NUMBER;
	Assert(pgaio_workerset_contains(&io_worker_control->workerset, MyIoWorkerId));
	pgaio_workerset_remove(&io_worker_control->workerset, MyIoWorkerId);
	notify_set = io_worker_control->workerset;
	Assert(io_worker_control->nworkers > 0);
	io_worker_control->nworkers--;
	Assert(pgaio_workerset_count(&io_worker_control->workerset) ==
		   io_worker_control->nworkers);
	LWLockRelease(AioWorkerControlLock);

	/*
	 * Notify other workers on pool change.  This allows the new highest
	 * worker to know that it is now the one that can time out, and closes a
	 * wakeup-loss race described in pgaio_worker_wake().
	 */
	pgaio_workerset_wake(notify_set);
}

/*
 * Register the worker in shared memory, assign MyIoWorkerId and register a
 * shutdown callback to release registration.
 */
static void
pgaio_worker_register(void)
{
	PgAioWorkerSet free_workerset;
	PgAioWorkerSet old_workerset;

	MyIoWorkerId = -1;

	LWLockAcquire(AioWorkerControlLock, LW_EXCLUSIVE);
	/* Find lowest unused worker ID. */
	pgaio_workerset_all(&free_workerset);
	pgaio_workerset_subtract(&free_workerset, &io_worker_control->workerset);
	if (!pgaio_workerset_is_empty(&free_workerset))
		MyIoWorkerId = pgaio_workerset_get_lowest(&free_workerset);
	if (MyIoWorkerId == -1)
		elog(ERROR, "couldn't find a free worker ID");

	Assert(io_worker_control->workers[MyIoWorkerId].proc_number ==
		   INVALID_PROC_NUMBER);
	io_worker_control->workers[MyIoWorkerId].proc_number = MyProcNumber;

	old_workerset = io_worker_control->workerset;
	Assert(!pgaio_workerset_contains(&old_workerset, MyIoWorkerId));
	pgaio_workerset_insert(&io_worker_control->workerset, MyIoWorkerId);
	io_worker_control->nworkers++;
	Assert(io_worker_control->nworkers <= MAX_IO_WORKERS);
	Assert(pgaio_workerset_count(&io_worker_control->workerset) ==
		   io_worker_control->nworkers);
	LWLockRelease(AioWorkerControlLock);

	/*
	 * Notify other workers on pool change.  If we were the highest worker,
	 * this allows the new highest worker to know that it can time out.
	 */
	pgaio_workerset_wake(old_workerset);

	on_shmem_exit(pgaio_worker_die, 0);
}

static void
pgaio_worker_error_callback(void *arg)
{
	ProcNumber	owner;
	PGPROC	   *owner_proc;
	int32		owner_pid;
	PgAioHandle *ioh = arg;

	if (!ioh)
		return;

	Assert(ioh->owner_procno != MyProcNumber);
	Assert(MyBackendType == B_IO_WORKER);

	owner = ioh->owner_procno;
	owner_proc = GetPGProcByNumber(owner);
	owner_pid = owner_proc->pid;

	errcontext("I/O worker executing I/O on behalf of process %d", owner_pid);
}

/*
 * Check if this backend is allowed to time out, and thus should use a
 * non-infinite sleep time.  Only the highest-numbered worker is allowed to
 * time out, and only if the pool is above io_min_workers.  Serializing
 * timeouts keeps IDs in a range 0..N without gaps, and avoids undershooting
 * io_min_workers.
 *
 * The result is only instantaneously true and may be temporarily inconsistent
 * in different workers around transitions, but all workers are woken up on
 * pool size or GUC changes making the result eventually consistent.
 */
static bool
pgaio_worker_can_timeout(void)
{
	PgAioWorkerSet workerset;

	if (MyIoWorkerId < io_min_workers)
		return false;

	/* Serialize against pool size changes. */
	LWLockAcquire(AioWorkerControlLock, LW_SHARED);
	workerset = io_worker_control->workerset;
	LWLockRelease(AioWorkerControlLock);

	if (MyIoWorkerId != pgaio_workerset_get_highest(&workerset))
		return false;

	return true;
}

/*
 * Emit a WARNING if io_min_workers > io_max_workers, since the worker
 * pool will never exceed io_max_workers regardless of the minimum setting.
 */
static void
check_io_worker_gucs(void)
{
	/* Only do the check in one worker, to limit noise */
	if (MyIoWorkerId != 0)
		return;

	if (io_min_workers > io_max_workers)
		ereport(WARNING,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("\"%s\" (%d) should be less than or equal to \"%s\" (%d)",
						"io_min_workers", io_min_workers,
						"io_max_workers", io_max_workers),
				 errdetail("The I/O worker pool will not exceed \"%s\" (%d) workers.",
						   "io_max_workers", io_max_workers)));
}

void
IoWorkerMain(const void *startup_data, size_t startup_data_len)
{
	sigjmp_buf	local_sigjmp_buf;
	TimestampTz idle_timeout_abs = 0;
	int			timeout_guc_used = 0;
	PgAioHandle *volatile error_ioh = NULL;
	ErrorContextCallback errcallback = {0};
	volatile int error_errno = 0;
	char		cmd[128];
	int			hist_ios = 0;
	int			hist_wakeups = 0;

	AuxiliaryProcessMainCommon();

	pqsignal(SIGHUP, SignalHandlerForConfigReload);
	pqsignal(SIGINT, die);		/* to allow manually triggering worker restart */

	/*
	 * Ignore SIGTERM, will get explicit shutdown via SIGUSR2 later in the
	 * shutdown sequence, similar to checkpointer.
	 */
	pqsignal(SIGTERM, PG_SIG_IGN);
	/* SIGQUIT handler was already set up by InitPostmasterChild */
	pqsignal(SIGALRM, PG_SIG_IGN);
	pqsignal(SIGPIPE, PG_SIG_IGN);
	pqsignal(SIGUSR1, procsignal_sigusr1_handler);
	pqsignal(SIGUSR2, SignalHandlerForShutdownRequest);

	/* also registers a shutdown callback to unregister */
	pgaio_worker_register();

	check_io_worker_gucs();

	sprintf(cmd, "%d", MyIoWorkerId);
	set_ps_display(cmd);

	errcallback.callback = pgaio_worker_error_callback;
	errcallback.previous = error_context_stack;
	error_context_stack = &errcallback;

	/* see PostgresMain() */
	if (sigsetjmp(local_sigjmp_buf, 1) != 0)
	{
		error_context_stack = NULL;
		HOLD_INTERRUPTS();

		EmitErrorReport();

		/*
		 * In the - very unlikely - case that the IO failed in a way that
		 * raises an error we need to mark the IO as failed.
		 *
		 * Need to do just enough error recovery so that we can mark the IO as
		 * failed and then exit (postmaster will start a new worker).
		 */
		LWLockReleaseAll();

		if (error_ioh != NULL)
		{
			/* should never fail without setting error_errno */
			Assert(error_errno != 0);

			errno = error_errno;

			START_CRIT_SECTION();
			pgaio_io_process_completion(error_ioh, -error_errno);
			END_CRIT_SECTION();
		}

		proc_exit(1);
	}

	/* We can now handle ereport(ERROR) */
	PG_exception_stack = &local_sigjmp_buf;

	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);

	while (!ShutdownRequestPending)
	{
		uint32		io_index;
		int			worker = -1;
		int			queue_depth = 0;
		bool		maybe_grow = false;

		/*
		 * Try to get a job to do.
		 *
		 * The lwlock acquisition also provides the necessary memory barrier
		 * to ensure that we don't see an outdated data in the handle.
		 */
		LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
		if ((io_index = pgaio_worker_submission_queue_consume()) == -1)
		{
			/* Nothing to do.  Mark self idle. */
			pgaio_workerset_insert(&io_worker_control->idle_workerset,
								   MyIoWorkerId);
		}
		else
		{
			/* Got one.  Clear idle flag. */
			pgaio_workerset_remove(&io_worker_control->idle_workerset,
								   MyIoWorkerId);

			/*
			 * See if we should wake up a higher numbered peer.  Only do that
			 * if this worker is not receiving spurious wakeups itself.  The
			 * intention is create a frontier beyond which idle workers stay
			 * asleep.
			 *
			 * This heuristic tries to discover the useful wakeup propagation
			 * chain length when IOs are very fast and workers wake up to find
			 * that all IOs have already been taken.
			 *
			 * If we chose not to wake a worker when we ideally should have,
			 * then the ratio will soon change to correct that.
			 */
			if (hist_wakeups <= hist_ios)
			{
				queue_depth = pgaio_worker_submission_queue_depth();
				if (queue_depth > 0)
				{
					/* Choose a worker higher than me to wake. */
					worker = pgaio_worker_choose_idle(MyIoWorkerId);
					if (worker == -1)
						maybe_grow = true;
				}
			}
		}
		LWLockRelease(AioWorkerSubmissionQueueLock);

		/* Propagate wakeups. */
		if (worker != -1)
		{
			pgaio_worker_wake(worker);
		}
		else if (maybe_grow)
		{
			/*
			 * We know there was at least one more item in the queue, and we
			 * failed to find a higher-numbered idle worker to wake.  Now we
			 * decide if we should try to start one more worker.
			 *
			 * We do this with a simple heuristic: is the queue depth greater
			 * than the current number of workers?
			 *
			 * Consider the following situations:
			 *
			 * 1. The queue depth is constantly increasing, because IOs are
			 * arriving faster than they can possibly be serviced.  It doesn't
			 * matter much which threshold we choose, as we will surely hit
			 * it.  Crossing the current worker count is a useful signal
			 * because it's clearly too deep to avoid queuing latency already,
			 * but still leaves a small window of opportunity to improve the
			 * situation before the queue overflows.
			 *
			 * 2. The worker pool is keeping up, no latency is being
			 * introduced and an extra worker would be a waste of resources.
			 * Queue depth distributions tend to be heavily skewed, with long
			 * tails of low probability spikes (due to submission clustering,
			 * scheduling, jitter, stalls, noisy neighbors, etc).  We want a
			 * number that is very unlikely to be triggered by an outlier, and
			 * we bet that an exponential or similar distribution whose
			 * outliers never reach this threshold must be almost entirely
			 * concentrated at the low end.  If we do see a spike as big as
			 * the worker count, we take it as a signal that the distribution
			 * is surely too wide.
			 *
			 * On its own, this is an extremely crude signal.  When combined
			 * with the wakeup propagation test that precedes it (but on its
			 * own tends to overshoot) and io_worker_launch_interval, the
			 * result is that we gradually test each pool size until we find
			 * one that doesn't trigger further expansion, and then hold it
			 * for at least io_worker_idle_timeout.
			 *
			 * XXX Perhaps ideas from queueing theory or control theory could
			 * do a better job of this.
			 */

			/* Read nworkers without lock for this heuristic purpose. */
			if (queue_depth > io_worker_control->nworkers)
				pgaio_worker_request_grow();
		}

		if (io_index != -1)
		{
			PgAioHandle *ioh = NULL;

			/* Cancel timeout and update wakeup:work ratio. */
			idle_timeout_abs = 0;
			if (++hist_ios == PGAIO_WORKER_WAKEUP_RATIO_SATURATE)
			{
				hist_wakeups /= 2;
				hist_ios /= 2;
			}

			ioh = &pgaio_ctl->io_handles[io_index];
			error_ioh = ioh;
			errcallback.arg = ioh;

			pgaio_debug_io(DEBUG4, ioh,
						   "worker %d processing IO",
						   MyIoWorkerId);

			/*
			 * Prevent interrupts between pgaio_io_reopen() and
			 * pgaio_io_perform_synchronously() that otherwise could lead to
			 * the FD getting closed in that window.
			 */
			HOLD_INTERRUPTS();

			/*
			 * It's very unlikely, but possible, that reopen fails. E.g. due
			 * to memory allocations failing or file permissions changing or
			 * such.  In that case we need to fail the IO.
			 *
			 * There's not really a good errno we can report here.
			 */
			error_errno = ENOENT;
			pgaio_io_reopen(ioh);

			/*
			 * To be able to exercise the reopen-fails path, allow injection
			 * points to trigger a failure at this point.
			 */
			INJECTION_POINT("aio-worker-after-reopen", ioh);

			error_errno = 0;
			error_ioh = NULL;

			/*
			 * As part of IO completion the buffer will be marked as NOACCESS,
			 * until the buffer is pinned again - which never happens in io
			 * workers. Therefore the next time there is IO for the same
			 * buffer, the memory will be considered inaccessible. To avoid
			 * that, explicitly allow access to the memory before reading data
			 * into it.
			 */
#ifdef USE_VALGRIND
			{
				struct iovec *iov;
				uint16		iov_length = pgaio_io_get_iovec_length(ioh, &iov);

				for (int i = 0; i < iov_length; i++)
					VALGRIND_MAKE_MEM_UNDEFINED(iov[i].iov_base, iov[i].iov_len);
			}
#endif

#ifdef PGAIO_WORKER_SHOW_PS_INFO
			{
				char	   *description = pgaio_io_get_target_description(ioh);

				sprintf(cmd, "%d: [%s] %s",
						MyIoWorkerId,
						pgaio_io_get_op_name(ioh),
						description);
				pfree(description);
				set_ps_display(cmd);
			}
#endif

			/*
			 * We don't expect this to ever fail with ERROR or FATAL, no need
			 * to keep error_ioh set to the IO.
			 * pgaio_io_perform_synchronously() contains a critical section to
			 * ensure we don't accidentally fail.
			 */
			pgaio_io_perform_synchronously(ioh);

			RESUME_INTERRUPTS();
			errcallback.arg = NULL;
		}
		else
		{
			int			timeout_ms;

			/* Cancel new worker request if pending. */
			pgaio_worker_cancel_grow();

			/* Compute the remaining allowed idle time. */
			if (io_worker_idle_timeout == -1)
			{
				/* Never time out. */
				timeout_ms = -1;
			}
			else
			{
				TimestampTz now = GetCurrentTimestamp();

				/* If the GUC changes, reset timer. */
				if (idle_timeout_abs != 0 &&
					io_worker_idle_timeout != timeout_guc_used)
					idle_timeout_abs = 0;

				/* Only the highest-numbered worker can time out. */
				if (pgaio_worker_can_timeout())
				{
					if (idle_timeout_abs == 0)
					{
						/*
						 * I have just been promoted to the timeout worker, or
						 * the GUC changed.  Compute new absolute time from
						 * now.
						 */
						idle_timeout_abs =
							TimestampTzPlusMilliseconds(now,
														io_worker_idle_timeout);
						timeout_guc_used = io_worker_idle_timeout;
					}
					timeout_ms =
						TimestampDifferenceMilliseconds(now, idle_timeout_abs);
				}
				else
				{
					/* No timeout for me. */
					idle_timeout_abs = 0;
					timeout_ms = -1;
				}
			}

#ifdef PGAIO_WORKER_SHOW_PS_INFO
			sprintf(cmd, "%d: idle, wakeups:ios = %d:%d",
					MyIoWorkerId, hist_wakeups, hist_ios);
			set_ps_display(cmd);
#endif

			if (WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH | WL_TIMEOUT,
						  timeout_ms,
						  WAIT_EVENT_IO_WORKER_MAIN) == WL_TIMEOUT)
			{
				/* WL_TIMEOUT */
				if (pgaio_worker_can_timeout())
					if (GetCurrentTimestamp() >= idle_timeout_abs)
						break;
			}
			else
			{
				/* WL_LATCH_SET */
				if (++hist_wakeups == PGAIO_WORKER_WAKEUP_RATIO_SATURATE)
				{
					hist_wakeups /= 2;
					hist_ios /= 2;
				}
			}
			ResetLatch(MyLatch);
		}

		CHECK_FOR_INTERRUPTS();

		if (ConfigReloadPending)
		{
			int			io_max_workers_prev = io_max_workers;
			int			io_min_workers_prev = io_min_workers;

			ConfigReloadPending = false;
			ProcessConfigFile(PGC_SIGHUP);

			/*
			 * Emit a WARNING if io_min_workers > io_max_workers.  If no bound
			 * has changed, skip this to avoid too many log messages.
			 */
			if (io_min_workers_prev != io_min_workers ||
				io_max_workers_prev != io_max_workers)
				check_io_worker_gucs();

			/* If io_max_workers has been decreased, exit highest first. */
			if (MyIoWorkerId >= io_max_workers)
				break;
		}
	}

	error_context_stack = errcallback.previous;
	proc_exit(0);
}

bool
pgaio_workers_enabled(void)
{
	return io_method == IOMETHOD_WORKER;
}