PostgreSQL Source Code git master
method_worker.c
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * method_worker.c
4 * AIO - perform AIO using worker processes
5 *
6 * IO workers consume IOs from a shared memory submission queue, run
7 * traditional synchronous system calls, and perform the shared completion
8 * handling immediately. Client code submits most requests by pushing IOs
9 * into the submission queue, and waits (if necessary) using condition
10 * variables. Some IOs cannot be performed in another process due to lack of
11 * infrastructure for reopening the file, and must processed synchronously by
12 * the client code when submitted.
13 *
14 * So that the submitter can make just one system call when submitting a batch
15 * of IOs, wakeups "fan out"; each woken IO worker can wake two more. XXX This
16 * could be improved by using futexes instead of latches to wake N waiters.
17 *
18 * This method of AIO is available in all builds on all operating systems, and
19 * is the default.
20 *
21 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
22 * Portions Copyright (c) 1994, Regents of the University of California
23 *
24 * IDENTIFICATION
25 * src/backend/storage/aio/method_worker.c
26 *
27 *-------------------------------------------------------------------------
28 */
29
30#include "postgres.h"
31
32#include "libpq/pqsignal.h"
33#include "miscadmin.h"
34#include "port/pg_bitutils.h"
37#include "storage/aio.h"
39#include "storage/aio_subsys.h"
40#include "storage/io_worker.h"
41#include "storage/ipc.h"
42#include "storage/latch.h"
43#include "storage/proc.h"
44#include "tcop/tcopprot.h"
46#include "utils/memdebug.h"
47#include "utils/ps_status.h"
48#include "utils/wait_event.h"
49
50
51/* How many workers should each worker wake up if needed? */
52#define IO_WORKER_WAKEUP_FANOUT 2
53
54
56{
63
64typedef struct AioWorkerSlot
65{
67 bool in_use;
69
70typedef struct AioWorkerControl
71{
75
76
77static size_t pgaio_worker_shmem_size(void);
78static void pgaio_worker_shmem_init(bool first_time);
79
81static int pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios);
82
83
86 .shmem_init = pgaio_worker_shmem_init,
87
88 .needs_synchronous_execution = pgaio_worker_needs_synchronous_execution,
89 .submit = pgaio_worker_submit,
90};
91
92
93/* GUCs */
94int io_workers = 3;
95
96
97static int io_worker_queue_size = 64;
98static int MyIoWorkerId;
101
102
103static size_t
105{
106 /* Round size up to next power of two so we can make a mask. */
108
109 return offsetof(AioWorkerSubmissionQueue, ios) +
110 sizeof(uint32) * *queue_size;
111}
112
113static size_t
115{
116 return offsetof(AioWorkerControl, workers) +
118}
119
120static size_t
122{
123 size_t sz;
124 int queue_size;
125
126 sz = pgaio_worker_queue_shmem_size(&queue_size);
128
129 return sz;
130}
131
132static void
134{
135 bool found;
136 int queue_size;
137
139 ShmemInitStruct("AioWorkerSubmissionQueue",
141 &found);
142 if (!found)
143 {
144 io_worker_submission_queue->size = queue_size;
147 }
148
150 ShmemInitStruct("AioWorkerControl",
152 &found);
153 if (!found)
154 {
156 for (int i = 0; i < MAX_IO_WORKERS; ++i)
157 {
160 }
161 }
162}
163
164static int
166{
167 int worker;
168
170 return -1;
171
172 /* Find the lowest bit position, and clear it. */
174 io_worker_control->idle_worker_mask &= ~(UINT64_C(1) << worker);
175
176 return worker;
177}
178
179static bool
181{
183 uint32 new_head;
184
186 new_head = (queue->head + 1) & (queue->size - 1);
187 if (new_head == queue->tail)
188 {
189 pgaio_debug(DEBUG3, "io queue is full, at %u elements",
191 return false; /* full */
192 }
193
194 queue->ios[queue->head] = pgaio_io_get_id(ioh);
195 queue->head = new_head;
196
197 return true;
198}
199
200static uint32
202{
204 uint32 result;
205
207 if (queue->tail == queue->head)
208 return UINT32_MAX; /* empty */
209
210 result = queue->ios[queue->tail];
211 queue->tail = (queue->tail + 1) & (queue->size - 1);
212
213 return result;
214}
215
216static uint32
218{
219 uint32 head;
220 uint32 tail;
221
224
225 if (tail > head)
227
228 Assert(head >= tail);
229
230 return head - tail;
231}
232
233static bool
235{
236 return
239 || !pgaio_io_can_reopen(ioh);
240}
241
242static void
244{
245 PgAioHandle *synchronous_ios[PGAIO_SUBMIT_BATCH_SIZE];
246 int nsync = 0;
247 Latch *wakeup = NULL;
248 int worker;
249
251
252 LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
253 for (int i = 0; i < nios; ++i)
254 {
257 {
258 /*
259 * We'll do it synchronously, but only after we've sent as many as
260 * we can to workers, to maximize concurrency.
261 */
262 synchronous_ios[nsync++] = ios[i];
263 continue;
264 }
265
266 if (wakeup == NULL)
267 {
268 /* Choose an idle worker to wake up if we haven't already. */
269 worker = pgaio_choose_idle_worker();
270 if (worker >= 0)
272
273 pgaio_debug_io(DEBUG4, ios[i],
274 "choosing worker %d",
275 worker);
276 }
277 }
278 LWLockRelease(AioWorkerSubmissionQueueLock);
279
280 if (wakeup)
282
283 /* Run whatever is left synchronously. */
284 if (nsync > 0)
285 {
286 for (int i = 0; i < nsync; ++i)
287 {
288 pgaio_io_perform_synchronously(synchronous_ios[i]);
289 }
290 }
291}
292
293static int
294pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios)
295{
296 for (int i = 0; i < num_staged_ios; i++)
297 {
298 PgAioHandle *ioh = staged_ios[i];
299
301 }
302
303 pgaio_worker_submit_internal(num_staged_ios, staged_ios);
304
305 return num_staged_ios;
306}
307
308/*
309 * on_shmem_exit() callback that releases the worker's slot in
310 * io_worker_control.
311 */
312static void
314{
315 LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
318
321 LWLockRelease(AioWorkerSubmissionQueueLock);
322}
323
324/*
325 * Register the worker in shared memory, assign MyIoWorkerId and register a
326 * shutdown callback to release registration.
327 */
328static void
330{
331 MyIoWorkerId = -1;
332
333 /*
334 * XXX: This could do with more fine-grained locking. But it's also not
335 * very common for the number of workers to change at the moment...
336 */
337 LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
338
339 for (int i = 0; i < MAX_IO_WORKERS; ++i)
340 {
342 {
345 MyIoWorkerId = i;
346 break;
347 }
348 else
350 }
351
352 if (MyIoWorkerId == -1)
353 elog(ERROR, "couldn't find a free worker slot");
354
357 LWLockRelease(AioWorkerSubmissionQueueLock);
358
360}
361
362static void
364{
365 ProcNumber owner;
366 PGPROC *owner_proc;
367 int32 owner_pid;
368 PgAioHandle *ioh = arg;
369
370 if (!ioh)
371 return;
372
375
376 owner = ioh->owner_procno;
377 owner_proc = GetPGProcByNumber(owner);
378 owner_pid = owner_proc->pid;
379
380 errcontext("I/O worker executing I/O on behalf of process %d", owner_pid);
381}
382
383void
384IoWorkerMain(const void *startup_data, size_t startup_data_len)
385{
386 sigjmp_buf local_sigjmp_buf;
387 PgAioHandle *volatile error_ioh = NULL;
388 ErrorContextCallback errcallback = {0};
389 volatile int error_errno = 0;
390 char cmd[128];
391
394
396 pqsignal(SIGINT, die); /* to allow manually triggering worker restart */
397
398 /*
399 * Ignore SIGTERM, will get explicit shutdown via SIGUSR2 later in the
400 * shutdown sequence, similar to checkpointer.
401 */
402 pqsignal(SIGTERM, SIG_IGN);
403 /* SIGQUIT handler was already set up by InitPostmasterChild */
404 pqsignal(SIGALRM, SIG_IGN);
405 pqsignal(SIGPIPE, SIG_IGN);
408
409 /* also registers a shutdown callback to unregister */
411
412 sprintf(cmd, "%d", MyIoWorkerId);
413 set_ps_display(cmd);
414
416 errcallback.previous = error_context_stack;
417 error_context_stack = &errcallback;
418
419 /* see PostgresMain() */
420 if (sigsetjmp(local_sigjmp_buf, 1) != 0)
421 {
422 error_context_stack = NULL;
424
426
427 /*
428 * In the - very unlikely - case that the IO failed in a way that
429 * raises an error we need to mark the IO as failed.
430 *
431 * Need to do just enough error recovery so that we can mark the IO as
432 * failed and then exit (postmaster will start a new worker).
433 */
435
436 if (error_ioh != NULL)
437 {
438 /* should never fail without setting error_errno */
439 Assert(error_errno != 0);
440
441 errno = error_errno;
442
444 pgaio_io_process_completion(error_ioh, -error_errno);
446 }
447
448 proc_exit(1);
449 }
450
451 /* We can now handle ereport(ERROR) */
452 PG_exception_stack = &local_sigjmp_buf;
453
454 sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
455
457 {
458 uint32 io_index;
460 int nlatches = 0;
461 int nwakeups = 0;
462 int worker;
463
464 /* Try to get a job to do. */
465 LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
466 if ((io_index = pgaio_worker_submission_queue_consume()) == UINT32_MAX)
467 {
468 /*
469 * Nothing to do. Mark self idle.
470 *
471 * XXX: Invent some kind of back pressure to reduce useless
472 * wakeups?
473 */
475 }
476 else
477 {
478 /* Got one. Clear idle flag. */
479 io_worker_control->idle_worker_mask &= ~(UINT64_C(1) << MyIoWorkerId);
480
481 /* See if we can wake up some peers. */
484 for (int i = 0; i < nwakeups; ++i)
485 {
486 if ((worker = pgaio_choose_idle_worker()) < 0)
487 break;
488 latches[nlatches++] = io_worker_control->workers[worker].latch;
489 }
490 }
491 LWLockRelease(AioWorkerSubmissionQueueLock);
492
493 for (int i = 0; i < nlatches; ++i)
494 SetLatch(latches[i]);
495
496 if (io_index != UINT32_MAX)
497 {
498 PgAioHandle *ioh = NULL;
499
500 ioh = &pgaio_ctl->io_handles[io_index];
501 error_ioh = ioh;
502 errcallback.arg = ioh;
503
505 "worker %d processing IO",
507
508 /*
509 * Prevent interrupts between pgaio_io_reopen() and
510 * pgaio_io_perform_synchronously() that otherwise could lead to
511 * the FD getting closed in that window.
512 */
514
515 /*
516 * It's very unlikely, but possible, that reopen fails. E.g. due
517 * to memory allocations failing or file permissions changing or
518 * such. In that case we need to fail the IO.
519 *
520 * There's not really a good errno we can report here.
521 */
522 error_errno = ENOENT;
523 pgaio_io_reopen(ioh);
524
525 /*
526 * To be able to exercise the reopen-fails path, allow injection
527 * points to trigger a failure at this point.
528 */
529 INJECTION_POINT("aio-worker-after-reopen", ioh);
530
531 error_errno = 0;
532 error_ioh = NULL;
533
534 /*
535 * As part of IO completion the buffer will be marked as NOACCESS,
536 * until the buffer is pinned again - which never happens in io
537 * workers. Therefore the next time there is IO for the same
538 * buffer, the memory will be considered inaccessible. To avoid
539 * that, explicitly allow access to the memory before reading data
540 * into it.
541 */
542#ifdef USE_VALGRIND
543 {
544 struct iovec *iov;
545 uint16 iov_length = pgaio_io_get_iovec_length(ioh, &iov);
546
547 for (int i = 0; i < iov_length; i++)
548 VALGRIND_MAKE_MEM_UNDEFINED(iov[i].iov_base, iov[i].iov_len);
549 }
550#endif
551
552 /*
553 * We don't expect this to ever fail with ERROR or FATAL, no need
554 * to keep error_ioh set to the IO.
555 * pgaio_io_perform_synchronously() contains a critical section to
556 * ensure we don't accidentally fail.
557 */
559
561 errcallback.arg = NULL;
562 }
563 else
564 {
566 WAIT_EVENT_IO_WORKER_MAIN);
568 }
569
571 }
572
573 error_context_stack = errcallback.previous;
574 proc_exit(0);
575}
576
577bool
579{
580 return io_method == IOMETHOD_WORKER;
581}
void pgaio_io_process_completion(PgAioHandle *ioh, int result)
Definition: aio.c:486
int io_method
Definition: aio.c:74
int pgaio_io_get_id(PgAioHandle *ioh)
Definition: aio.c:316
PgAioCtl * pgaio_ctl
Definition: aio.c:78
void pgaio_io_prepare_submit(PgAioHandle *ioh)
Definition: aio.c:468
@ IOMETHOD_WORKER
Definition: aio.h:35
@ PGAIO_HF_REFERENCES_LOCAL
Definition: aio.h:60
#define pgaio_debug(elevel, msg,...)
Definition: aio_internal.h:376
#define pgaio_debug_io(elevel, ioh, msg,...)
Definition: aio_internal.h:389
#define PGAIO_SUBMIT_BATCH_SIZE
Definition: aio_internal.h:28
void pgaio_io_perform_synchronously(PgAioHandle *ioh)
Definition: aio_io.c:116
int pgaio_io_get_iovec_length(PgAioHandle *ioh, struct iovec **iov)
Definition: aio_io.c:219
void pgaio_io_reopen(PgAioHandle *ioh)
Definition: aio_target.c:116
bool pgaio_io_can_reopen(PgAioHandle *ioh)
Definition: aio_target.c:103
void AuxiliaryProcessMainCommon(void)
Definition: auxprocess.c:39
sigset_t UnBlockSig
Definition: pqsignal.c:22
#define Min(x, y)
Definition: c.h:975
#define FLEXIBLE_ARRAY_MEMBER
Definition: c.h:434
int32_t int32
Definition: c.h:498
uint64_t uint64
Definition: c.h:503
uint16_t uint16
Definition: c.h:501
uint32_t uint32
Definition: c.h:502
void EmitErrorReport(void)
Definition: elog.c:1692
ErrorContextCallback * error_context_stack
Definition: elog.c:95
sigjmp_buf * PG_exception_stack
Definition: elog.c:97
#define errcontext
Definition: elog.h:197
#define DEBUG3
Definition: elog.h:28
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:225
#define DEBUG4
Definition: elog.h:27
ProcNumber MyProcNumber
Definition: globals.c:91
bool IsUnderPostmaster
Definition: globals.c:121
struct Latch * MyLatch
Definition: globals.c:64
Assert(PointerIsAligned(start, uint64))
#define INJECTION_POINT(name, arg)
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:109
volatile sig_atomic_t ShutdownRequestPending
Definition: interrupt.c:28
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:65
void on_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:365
void proc_exit(int code)
Definition: ipc.c:104
int i
Definition: isn.c:77
void SetLatch(Latch *latch)
Definition: latch.c:288
void ResetLatch(Latch *latch)
Definition: latch.c:372
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:172
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1182
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1902
void LWLockReleaseAll(void)
Definition: lwlock.c:1953
@ LW_EXCLUSIVE
Definition: lwlock.h:114
#define VALGRIND_MAKE_MEM_UNDEFINED(addr, size)
Definition: memdebug.h:28
static size_t pgaio_worker_control_shmem_size(void)
static uint32 pgaio_worker_submission_queue_depth(void)
struct AioWorkerControl AioWorkerControl
static void pgaio_worker_error_callback(void *arg)
static bool pgaio_worker_needs_synchronous_execution(PgAioHandle *ioh)
static int pgaio_choose_idle_worker(void)
static int pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios)
#define IO_WORKER_WAKEUP_FANOUT
Definition: method_worker.c:52
static size_t pgaio_worker_shmem_size(void)
static size_t pgaio_worker_queue_shmem_size(int *queue_size)
static AioWorkerSubmissionQueue * io_worker_submission_queue
Definition: method_worker.c:99
static int io_worker_queue_size
Definition: method_worker.c:97
static void pgaio_worker_register(void)
static int MyIoWorkerId
Definition: method_worker.c:98
struct AioWorkerSubmissionQueue AioWorkerSubmissionQueue
const IoMethodOps pgaio_worker_ops
Definition: method_worker.c:84
static void pgaio_worker_die(int code, Datum arg)
struct AioWorkerSlot AioWorkerSlot
static uint32 pgaio_worker_submission_queue_consume(void)
static void pgaio_worker_submit_internal(int nios, PgAioHandle *ios[])
static bool pgaio_worker_submission_queue_insert(PgAioHandle *ioh)
bool pgaio_workers_enabled(void)
void IoWorkerMain(const void *startup_data, size_t startup_data_len)
static AioWorkerControl * io_worker_control
static void pgaio_worker_shmem_init(bool first_time)
int io_workers
Definition: method_worker.c:94
#define RESUME_INTERRUPTS()
Definition: miscadmin.h:136
#define START_CRIT_SECTION()
Definition: miscadmin.h:150
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:123
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:134
@ B_IO_WORKER
Definition: miscadmin.h:364
#define END_CRIT_SECTION()
Definition: miscadmin.h:152
BackendType MyBackendType
Definition: miscinit.c:64
void * arg
static int pg_rightmost_one_pos64(uint64 word)
Definition: pg_bitutils.h:145
static uint32 pg_nextpower2_32(uint32 num)
Definition: pg_bitutils.h:189
#define die(msg)
#define pqsignal
Definition: port.h:531
#define sprintf
Definition: port.h:241
uintptr_t Datum
Definition: postgres.h:69
#define MAX_IO_WORKERS
Definition: proc.h:454
#define GetPGProcByNumber(n)
Definition: proc.h:432
int ProcNumber
Definition: procnumber.h:24
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:674
static void set_ps_display(const char *activity)
Definition: ps_status.h:40
Size add_size(Size s1, Size s2)
Definition: shmem.c:493
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:387
uint64 idle_worker_mask
Definition: method_worker.c:72
AioWorkerSlot workers[FLEXIBLE_ARRAY_MEMBER]
Definition: method_worker.c:73
uint32 ios[FLEXIBLE_ARRAY_MEMBER]
Definition: method_worker.c:61
struct ErrorContextCallback * previous
Definition: elog.h:296
void(* callback)(void *arg)
Definition: elog.h:297
size_t(* shmem_size)(void)
Definition: aio_internal.h:271
Definition: latch.h:114
Definition: proc.h:171
int pid
Definition: proc.h:191
PgAioHandle * io_handles
Definition: aio_internal.h:246
int32 owner_procno
Definition: aio_internal.h:125
#define WL_EXIT_ON_PM_DEATH
Definition: waiteventset.h:39
#define WL_LATCH_SET
Definition: waiteventset.h:34
static TimestampTz wakeup[NUM_WALRCV_WAKEUPS]
Definition: walreceiver.c:130
#define SIGHUP
Definition: win32_port.h:158
#define SIGPIPE
Definition: win32_port.h:163
#define SIGUSR1
Definition: win32_port.h:170
#define SIGALRM
Definition: win32_port.h:164
#define SIGUSR2
Definition: win32_port.h:171