Branch data Line data Source code
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 : : * The pool of workers tries to stabilize at a size that can handle recently
15 : : * seen variation in demand, within the configured limits.
16 : : *
17 : : * This method of AIO is available in all builds on all operating systems, and
18 : : * is the default.
19 : : *
20 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
21 : : * Portions Copyright (c) 1994, Regents of the University of California
22 : : *
23 : : * IDENTIFICATION
24 : : * src/backend/storage/aio/method_worker.c
25 : : *
26 : : *-------------------------------------------------------------------------
27 : : */
28 : :
29 : : #include "postgres.h"
30 : :
31 : : #include <limits.h>
32 : :
33 : : #include "libpq/pqsignal.h"
34 : : #include "miscadmin.h"
35 : : #include "port/pg_bitutils.h"
36 : : #include "postmaster/auxprocess.h"
37 : : #include "postmaster/interrupt.h"
38 : : #include "storage/aio.h"
39 : : #include "storage/aio_internal.h"
40 : : #include "storage/aio_subsys.h"
41 : : #include "storage/io_worker.h"
42 : : #include "storage/ipc.h"
43 : : #include "storage/latch.h"
44 : : #include "storage/lwlock.h"
45 : : #include "storage/pmsignal.h"
46 : : #include "storage/proc.h"
47 : : #include "storage/shmem.h"
48 : : #include "tcop/tcopprot.h"
49 : : #include "utils/injection_point.h"
50 : : #include "utils/memdebug.h"
51 : : #include "utils/ps_status.h"
52 : : #include "utils/wait_event.h"
53 : :
54 : : /*
55 : : * Saturation for counters used to estimate wakeup:IO ratio.
56 : : *
57 : : * We maintain hist_wakeups for wakeups received and hist_ios for IOs
58 : : * processed by each worker. When either counter reaches this saturation
59 : : * value, we divide both by two. The result is an exponentially decaying
60 : : * ratio of wakeups to IOs, with a very short memory.
61 : : *
62 : : * If a worker is itself experiencing useless wakeups, it assumes that
63 : : * higher-numbered workers would experience even more, so it should end the
64 : : * chain.
65 : : */
66 : : #define PGAIO_WORKER_WAKEUP_RATIO_SATURATE 4
67 : :
68 : : /* Debugging support: show current IO and wakeups:ios statistics in ps. */
69 : : /* #define PGAIO_WORKER_SHOW_PS_INFO */
70 : :
71 : : typedef struct PgAioWorkerSubmissionQueue
72 : : {
73 : : uint32 size;
74 : : uint32 head;
75 : : uint32 tail;
76 : : int sqes[FLEXIBLE_ARRAY_MEMBER];
77 : : } PgAioWorkerSubmissionQueue;
78 : :
79 : : typedef struct PgAioWorkerSlot
80 : : {
81 : : ProcNumber proc_number;
82 : : } PgAioWorkerSlot;
83 : :
84 : : /*
85 : : * Sets of worker IDs are held in a simple bitmap, accessed through functions
86 : : * that provide a more readable abstraction. If we wanted to support more
87 : : * workers than that, the contention on the single queue would surely get too
88 : : * high, so we might want to consider multiple pools instead of widening this.
89 : : */
90 : : typedef uint64 PgAioWorkerSet;
91 : :
92 : : #define PGAIO_WORKERSET_BITS (sizeof(PgAioWorkerSet) * CHAR_BIT)
93 : :
94 : : static_assert(PGAIO_WORKERSET_BITS >= MAX_IO_WORKERS, "too small");
95 : :
96 : : typedef struct PgAioWorkerControl
97 : : {
98 : : /* Seen by postmaster */
99 : : bool grow;
100 : : bool grow_signal_sent;
101 : :
102 : : /* Protected by AioWorkerSubmissionQueueLock. */
103 : : PgAioWorkerSet idle_workerset;
104 : :
105 : : /* Protected by AioWorkerControlLock. */
106 : : PgAioWorkerSet workerset;
107 : : int nworkers;
108 : :
109 : : /* Protected by AioWorkerControlLock. */
110 : : PgAioWorkerSlot workers[FLEXIBLE_ARRAY_MEMBER];
111 : : } PgAioWorkerControl;
112 : :
113 : :
114 : : static void pgaio_worker_shmem_request(void *arg);
115 : : static void pgaio_worker_shmem_init(void *arg);
116 : :
117 : : static bool pgaio_worker_needs_synchronous_execution(PgAioHandle *ioh);
118 : : static int pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios);
119 : :
120 : :
121 : : const IoMethodOps pgaio_worker_ops = {
122 : : .shmem_callbacks.request_fn = pgaio_worker_shmem_request,
123 : : .shmem_callbacks.init_fn = pgaio_worker_shmem_init,
124 : :
125 : : .needs_synchronous_execution = pgaio_worker_needs_synchronous_execution,
126 : : .submit = pgaio_worker_submit,
127 : : };
128 : :
129 : :
130 : : /* GUCs */
131 : : int io_min_workers = 2;
132 : : int io_max_workers = 8;
133 : : int io_worker_idle_timeout = 60000;
134 : : int io_worker_launch_interval = 100;
135 : :
136 : :
137 : : static int io_worker_queue_size = 64;
138 : : static int MyIoWorkerId = -1;
139 : : static PgAioWorkerSubmissionQueue *io_worker_submission_queue;
140 : : static PgAioWorkerControl *io_worker_control;
141 : :
142 : :
143 : : static void
144 : 2500 : pgaio_workerset_initialize(PgAioWorkerSet *set)
145 : : {
146 : 2500 : *set = 0;
147 : 2500 : }
148 : :
149 : : static bool
150 : 625943 : pgaio_workerset_is_empty(PgAioWorkerSet *set)
151 : : {
152 : 625943 : return *set == 0;
153 : : }
154 : :
155 : : static PgAioWorkerSet
156 : 1534358 : pgaio_workerset_singleton(int worker)
157 : : {
158 : : Assert(worker >= 0 && worker < MAX_IO_WORKERS);
159 : 1534358 : return UINT64_C(1) << worker;
160 : : }
161 : :
162 : : static void
163 : 1325 : pgaio_workerset_all(PgAioWorkerSet *set)
164 : : {
165 : 1325 : *set = UINT64_MAX >> (PGAIO_WORKERSET_BITS - MAX_IO_WORKERS);
166 : 1325 : }
167 : :
168 : : static void
169 : 1325 : pgaio_workerset_subtract(PgAioWorkerSet *set1, const PgAioWorkerSet *set2)
170 : : {
171 : 1325 : *set1 &= ~*set2;
172 : 1325 : }
173 : :
174 : : static void
175 : 465170 : pgaio_workerset_insert(PgAioWorkerSet *set, int worker)
176 : : {
177 : : Assert(worker >= 0 && worker < MAX_IO_WORKERS);
178 : 465170 : *set |= pgaio_workerset_singleton(worker);
179 : 465170 : }
180 : :
181 : : static void
182 : 1069188 : pgaio_workerset_remove(PgAioWorkerSet *set, int worker)
183 : : {
184 : : Assert(worker >= 0 && worker < MAX_IO_WORKERS);
185 : 1069188 : *set &= ~pgaio_workerset_singleton(worker);
186 : 1069188 : }
187 : :
188 : : static void
189 : 11242 : pgaio_workerset_remove_lte(PgAioWorkerSet *set, int worker)
190 : : {
191 : : Assert(worker >= 0 && worker < MAX_IO_WORKERS);
192 : 11242 : *set &= (~(PgAioWorkerSet) 0) << (worker + 1);
193 : 11242 : }
194 : :
195 : : static int
196 : 15984 : pgaio_workerset_get_highest(PgAioWorkerSet *set)
197 : : {
198 : : Assert(!pgaio_workerset_is_empty(set));
199 : 15984 : return pg_leftmost_one_pos64(*set);
200 : : }
201 : :
202 : : static int
203 : 597719 : pgaio_workerset_get_lowest(PgAioWorkerSet *set)
204 : : {
205 : : Assert(!pgaio_workerset_is_empty(set));
206 : 597719 : return pg_rightmost_one_pos64(*set);
207 : : }
208 : :
209 : : static int
210 : 2388 : pgaio_workerset_pop_lowest(PgAioWorkerSet *set)
211 : : {
212 : 2388 : int worker = pgaio_workerset_get_lowest(set);
213 : :
214 : 2388 : pgaio_workerset_remove(set, worker);
215 : 2388 : return worker;
216 : : }
217 : :
218 : : #ifdef USE_ASSERT_CHECKING
219 : : static bool
220 : : pgaio_workerset_contains(PgAioWorkerSet *set, int worker)
221 : : {
222 : : Assert(worker >= 0 && worker < MAX_IO_WORKERS);
223 : : return (*set & pgaio_workerset_singleton(worker)) != 0;
224 : : }
225 : :
226 : : static int
227 : : pgaio_workerset_count(PgAioWorkerSet *set)
228 : : {
229 : : return pg_popcount64(*set);
230 : : }
231 : : #endif
232 : :
233 : : static void
234 : 1253 : pgaio_worker_shmem_request(void *arg)
235 : : {
236 : : size_t size;
237 : : int queue_size;
238 : :
239 : : /* Round size up to next power of two so we can make a mask. */
240 : 1253 : queue_size = pg_nextpower2_32(io_worker_queue_size);
241 : :
242 : 1253 : size = offsetof(PgAioWorkerSubmissionQueue, sqes) + sizeof(int) * queue_size;
243 : 1253 : ShmemRequestStruct(.name = "AioWorkerSubmissionQueue",
244 : : .size = size,
245 : : .ptr = (void **) &io_worker_submission_queue,
246 : : );
247 : :
248 : 1253 : size = offsetof(PgAioWorkerControl, workers) + sizeof(PgAioWorkerSlot) * MAX_IO_WORKERS;
249 : 1253 : ShmemRequestStruct(.name = "AioWorkerControl",
250 : : .size = size,
251 : : .ptr = (void **) &io_worker_control,
252 : : );
253 : 1253 : }
254 : :
255 : : static void
256 : 1250 : pgaio_worker_shmem_init(void *arg)
257 : : {
258 : : int queue_size;
259 : :
260 : : /* Round size up like in pgaio_worker_shmem_request() */
261 : 1250 : queue_size = pg_nextpower2_32(io_worker_queue_size);
262 : :
263 : 1250 : io_worker_submission_queue->size = queue_size;
264 : 1250 : io_worker_submission_queue->head = 0;
265 : 1250 : io_worker_submission_queue->tail = 0;
266 : 1250 : io_worker_control->grow = false;
267 : 1250 : pgaio_workerset_initialize(&io_worker_control->workerset);
268 : 1250 : pgaio_workerset_initialize(&io_worker_control->idle_workerset);
269 : :
270 [ + + ]: 41250 : for (int i = 0; i < MAX_IO_WORKERS; ++i)
271 : 40000 : io_worker_control->workers[i].proc_number = INVALID_PROC_NUMBER;
272 : 1250 : }
273 : :
274 : : /*
275 : : * Tell postmaster that we think a new worker is needed.
276 : : */
277 : : static void
278 : 46 : pgaio_worker_request_grow(void)
279 : : {
280 : : /*
281 : : * Suppress useless signaling if we already know that we're at the
282 : : * maximum. This uses an unlocked read of nworkers, but that's OK for
283 : : * this heuristic purpose.
284 : : */
285 [ - + ]: 46 : if (io_worker_control->nworkers >= io_max_workers)
286 : 0 : return;
287 : :
288 : : /* Already requested? */
289 [ + + ]: 46 : if (io_worker_control->grow)
290 : 13 : return;
291 : :
292 : 33 : io_worker_control->grow = true;
293 : 33 : pg_memory_barrier();
294 : :
295 : : /*
296 : : * If the postmaster has already been signaled, don't do it again until
297 : : * the postmaster clears this flag. There is no point in repeated signals
298 : : * if grow is being set and cleared repeatedly while the postmaster is
299 : : * waiting for io_worker_launch_interval, which it applies even to
300 : : * canceled requests.
301 : : */
302 [ + + ]: 33 : if (io_worker_control->grow_signal_sent)
303 : 3 : return;
304 : :
305 : 30 : io_worker_control->grow_signal_sent = true;
306 : 30 : pg_memory_barrier();
307 : 30 : SendPostmasterSignal(PMSIGNAL_IO_WORKER_GROW);
308 : : }
309 : :
310 : : /*
311 : : * Cancel any request for a new worker, after observing an empty queue.
312 : : */
313 : : static void
314 : 463845 : pgaio_worker_cancel_grow(void)
315 : : {
316 [ + + ]: 463845 : if (!io_worker_control->grow)
317 : 463812 : return;
318 : :
319 : 33 : io_worker_control->grow = false;
320 : 33 : pg_memory_barrier();
321 : : }
322 : :
323 : : /*
324 : : * Called by the postmaster to check if a new worker has been requested (but
325 : : * possibly canceled since).
326 : : */
327 : : bool
328 : 308754 : pgaio_worker_pm_test_grow_signal_sent(void)
329 : : {
330 : 308754 : pg_memory_barrier();
331 [ + - + + ]: 308754 : return io_worker_control && io_worker_control->grow_signal_sent;
332 : : }
333 : :
334 : : /*
335 : : * Called by the postmaster to check if a new worker has been requested and
336 : : * not canceled since.
337 : : */
338 : : bool
339 : 45 : pgaio_worker_pm_test_grow(void)
340 : : {
341 : 45 : pg_memory_barrier();
342 [ + - + + ]: 45 : return io_worker_control && io_worker_control->grow;
343 : : }
344 : :
345 : : /*
346 : : * Called by the postmaster to clear the request for a new worker.
347 : : */
348 : : void
349 : 35 : pgaio_worker_pm_clear_grow_signal_sent(void)
350 : : {
351 [ - + ]: 35 : if (!io_worker_control)
352 : 0 : return;
353 : :
354 : 35 : io_worker_control->grow = false;
355 : 35 : io_worker_control->grow_signal_sent = false;
356 : 35 : pg_memory_barrier();
357 : : }
358 : :
359 : : static int
360 : 619580 : pgaio_worker_choose_idle(int only_workers_above)
361 : : {
362 : : PgAioWorkerSet workerset;
363 : : int worker;
364 : :
365 : : Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));
366 : :
367 : 619580 : workerset = io_worker_control->idle_workerset;
368 [ + + ]: 619580 : if (only_workers_above >= 0)
369 : 11242 : pgaio_workerset_remove_lte(&workerset, only_workers_above);
370 [ + + ]: 619580 : if (pgaio_workerset_is_empty(&workerset))
371 : 25574 : return -1;
372 : :
373 : : /* Find the lowest numbered idle worker and mark it not idle. */
374 : 594006 : worker = pgaio_workerset_get_lowest(&workerset);
375 : 594006 : pgaio_workerset_remove(&io_worker_control->idle_workerset, worker);
376 : :
377 : 594006 : return worker;
378 : : }
379 : :
380 : : /*
381 : : * Try to wake a worker by setting its latch, to tell it there are IOs to
382 : : * process in the submission queue.
383 : : */
384 : : static void
385 : 596394 : pgaio_worker_wake(int worker)
386 : : {
387 : : ProcNumber proc_number;
388 : :
389 : : /*
390 : : * If the selected worker is concurrently exiting, then pgaio_worker_die()
391 : : * had not yet removed it as of when we saw it in idle_workerset. That's
392 : : * OK, because it will wake all remaining workers to close wakeup-vs-exit
393 : : * races: *someone* will see the queued IO. If there are no workers
394 : : * running, the postmaster will start a new one.
395 : : */
396 : 596394 : proc_number = io_worker_control->workers[worker].proc_number;
397 [ + + ]: 596394 : if (proc_number != INVALID_PROC_NUMBER)
398 : 596391 : SetLatch(&GetPGProcByNumber(proc_number)->procLatch);
399 : 596394 : }
400 : :
401 : : /*
402 : : * Try to wake a set of workers. Used on pool change, to close races
403 : : * described in the callers.
404 : : */
405 : : static void
406 : 2650 : pgaio_workerset_wake(PgAioWorkerSet workerset)
407 : : {
408 [ + + ]: 5038 : while (!pgaio_workerset_is_empty(&workerset))
409 : 2388 : pgaio_worker_wake(pgaio_workerset_pop_lowest(&workerset));
410 : 2650 : }
411 : :
412 : : static bool
413 : 608536 : pgaio_worker_submission_queue_insert(PgAioHandle *ioh)
414 : : {
415 : : PgAioWorkerSubmissionQueue *queue;
416 : : uint32 new_head;
417 : :
418 : : Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));
419 : :
420 : 608536 : queue = io_worker_submission_queue;
421 : 608536 : new_head = (queue->head + 1) & (queue->size - 1);
422 [ - + ]: 608536 : if (new_head == queue->tail)
423 : : {
424 [ # # ]: 0 : pgaio_debug(DEBUG3, "io queue is full, at %u elements",
425 : : io_worker_submission_queue->size);
426 : 0 : return false; /* full */
427 : : }
428 : :
429 : 608536 : queue->sqes[queue->head] = pgaio_io_get_id(ioh);
430 : 608536 : queue->head = new_head;
431 : :
432 : 608536 : return true;
433 : : }
434 : :
435 : : static int
436 : 933989 : pgaio_worker_submission_queue_consume(void)
437 : : {
438 : : PgAioWorkerSubmissionQueue *queue;
439 : : int result;
440 : :
441 : : Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));
442 : :
443 : 933989 : queue = io_worker_submission_queue;
444 [ + + ]: 933989 : if (queue->tail == queue->head)
445 : 463845 : return -1; /* empty */
446 : :
447 : 470144 : result = queue->sqes[queue->tail];
448 : 470144 : queue->tail = (queue->tail + 1) & (queue->size - 1);
449 : :
450 : 470144 : return result;
451 : : }
452 : :
453 : : static uint32
454 : 272237 : pgaio_worker_submission_queue_depth(void)
455 : : {
456 : : uint32 head;
457 : : uint32 tail;
458 : :
459 : : Assert(LWLockHeldByMeInMode(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE));
460 : :
461 : 272237 : head = io_worker_submission_queue->head;
462 : 272237 : tail = io_worker_submission_queue->tail;
463 : :
464 [ + + ]: 272237 : if (tail > head)
465 : 195 : head += io_worker_submission_queue->size;
466 : :
467 : : Assert(head >= tail);
468 : :
469 : 272237 : return head - tail;
470 : : }
471 : :
472 : : static bool
473 : 616249 : pgaio_worker_needs_synchronous_execution(PgAioHandle *ioh)
474 : : {
475 : : return
476 : 616249 : !IsUnderPostmaster
477 [ + + ]: 612613 : || ioh->flags & PGAIO_HF_REFERENCES_LOCAL
478 [ + + - + ]: 1228862 : || !pgaio_io_can_reopen(ioh);
479 : : }
480 : :
481 : : static int
482 : 610477 : pgaio_worker_submit(uint16 num_staged_ios, PgAioHandle **staged_ios)
483 : : {
484 : 610477 : PgAioHandle **synchronous_ios = NULL;
485 : 610477 : int nsync = 0;
486 : 610477 : int worker = -1;
487 : :
488 : : Assert(num_staged_ios <= PGAIO_SUBMIT_BATCH_SIZE);
489 : :
490 [ + + ]: 1221152 : for (int i = 0; i < num_staged_ios; i++)
491 : 610675 : pgaio_io_prepare_submit(staged_ios[i]);
492 : :
493 [ + + ]: 610477 : if (LWLockConditionalAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE))
494 : : {
495 [ + + ]: 1216874 : for (int i = 0; i < num_staged_ios; ++i)
496 : : {
497 : : Assert(!pgaio_worker_needs_synchronous_execution(staged_ios[i]));
498 [ - + ]: 608536 : if (!pgaio_worker_submission_queue_insert(staged_ios[i]))
499 : : {
500 : : /*
501 : : * Do the rest synchronously. If the queue is full, give up
502 : : * and do the rest synchronously. We're holding an exclusive
503 : : * lock on the queue so nothing can consume entries.
504 : : */
505 : 0 : synchronous_ios = &staged_ios[i];
506 : 0 : nsync = (num_staged_ios - i);
507 : :
508 : 0 : break;
509 : : }
510 : : }
511 : : /* Choose one worker to wake for this batch. */
512 : 608338 : worker = pgaio_worker_choose_idle(-1);
513 : 608338 : LWLockRelease(AioWorkerSubmissionQueueLock);
514 : :
515 : : /* Wake up chosen worker. It will wake peers if necessary. */
516 [ + + ]: 608338 : if (worker != -1)
517 : 591730 : pgaio_worker_wake(worker);
518 : : }
519 : : else
520 : : {
521 : : /* do everything synchronously, no wakeup needed */
522 : 2139 : synchronous_ios = staged_ios;
523 : 2139 : nsync = num_staged_ios;
524 : : }
525 : :
526 : : /* Run whatever is left synchronously. */
527 [ + + ]: 610477 : if (nsync > 0)
528 : : {
529 [ + + ]: 4278 : for (int i = 0; i < nsync; ++i)
530 : : {
531 : 2139 : pgaio_io_perform_synchronously(synchronous_ios[i]);
532 : : }
533 : : }
534 : :
535 : 610477 : return num_staged_ios;
536 : : }
537 : :
538 : : /*
539 : : * on_shmem_exit() callback that releases the worker's slot in
540 : : * io_worker_control.
541 : : */
542 : : static void
543 : 1325 : pgaio_worker_die(int code, Datum arg)
544 : : {
545 : : PgAioWorkerSet notify_set;
546 : :
547 : 1325 : LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
548 : 1325 : pgaio_workerset_remove(&io_worker_control->idle_workerset, MyIoWorkerId);
549 : 1325 : LWLockRelease(AioWorkerSubmissionQueueLock);
550 : :
551 : 1325 : LWLockAcquire(AioWorkerControlLock, LW_EXCLUSIVE);
552 : : Assert(io_worker_control->workers[MyIoWorkerId].proc_number == MyProcNumber);
553 : 1325 : io_worker_control->workers[MyIoWorkerId].proc_number = INVALID_PROC_NUMBER;
554 : : Assert(pgaio_workerset_contains(&io_worker_control->workerset, MyIoWorkerId));
555 : 1325 : pgaio_workerset_remove(&io_worker_control->workerset, MyIoWorkerId);
556 : 1325 : notify_set = io_worker_control->workerset;
557 : : Assert(io_worker_control->nworkers > 0);
558 : 1325 : io_worker_control->nworkers--;
559 : : Assert(pgaio_workerset_count(&io_worker_control->workerset) ==
560 : : io_worker_control->nworkers);
561 : 1325 : LWLockRelease(AioWorkerControlLock);
562 : :
563 : : /*
564 : : * Notify other workers on pool change. This allows the new highest
565 : : * worker to know that it is now the one that can time out, and closes a
566 : : * wakeup-loss race described in pgaio_worker_wake().
567 : : */
568 : 1325 : pgaio_workerset_wake(notify_set);
569 : 1325 : }
570 : :
571 : : /*
572 : : * Register the worker in shared memory, assign MyIoWorkerId and register a
573 : : * shutdown callback to release registration.
574 : : */
575 : : static void
576 : 1325 : pgaio_worker_register(void)
577 : : {
578 : : PgAioWorkerSet free_workerset;
579 : : PgAioWorkerSet old_workerset;
580 : :
581 : 1325 : MyIoWorkerId = -1;
582 : :
583 : 1325 : LWLockAcquire(AioWorkerControlLock, LW_EXCLUSIVE);
584 : : /* Find lowest unused worker ID. */
585 : 1325 : pgaio_workerset_all(&free_workerset);
586 : 1325 : pgaio_workerset_subtract(&free_workerset, &io_worker_control->workerset);
587 [ + - ]: 1325 : if (!pgaio_workerset_is_empty(&free_workerset))
588 : 1325 : MyIoWorkerId = pgaio_workerset_get_lowest(&free_workerset);
589 [ - + ]: 1325 : if (MyIoWorkerId == -1)
590 [ # # ]: 0 : elog(ERROR, "couldn't find a free worker ID");
591 : :
592 : : Assert(io_worker_control->workers[MyIoWorkerId].proc_number ==
593 : : INVALID_PROC_NUMBER);
594 : 1325 : io_worker_control->workers[MyIoWorkerId].proc_number = MyProcNumber;
595 : :
596 : 1325 : old_workerset = io_worker_control->workerset;
597 : : Assert(!pgaio_workerset_contains(&old_workerset, MyIoWorkerId));
598 : 1325 : pgaio_workerset_insert(&io_worker_control->workerset, MyIoWorkerId);
599 : 1325 : io_worker_control->nworkers++;
600 : : Assert(io_worker_control->nworkers <= MAX_IO_WORKERS);
601 : : Assert(pgaio_workerset_count(&io_worker_control->workerset) ==
602 : : io_worker_control->nworkers);
603 : 1325 : LWLockRelease(AioWorkerControlLock);
604 : :
605 : : /*
606 : : * Notify other workers on pool change. If we were the highest worker,
607 : : * this allows the new highest worker to know that it can time out.
608 : : */
609 : 1325 : pgaio_workerset_wake(old_workerset);
610 : :
611 : 1325 : on_shmem_exit(pgaio_worker_die, 0);
612 : 1325 : }
613 : :
614 : : static void
615 : 1640 : pgaio_worker_error_callback(void *arg)
616 : : {
617 : : ProcNumber owner;
618 : : PGPROC *owner_proc;
619 : : int32 owner_pid;
620 : 1640 : PgAioHandle *ioh = arg;
621 : :
622 [ - + ]: 1640 : if (!ioh)
623 : 0 : return;
624 : :
625 : : Assert(ioh->owner_procno != MyProcNumber);
626 : : Assert(MyBackendType == B_IO_WORKER);
627 : :
628 : 1640 : owner = ioh->owner_procno;
629 : 1640 : owner_proc = GetPGProcByNumber(owner);
630 : 1640 : owner_pid = owner_proc->pid;
631 : :
632 : 1640 : errcontext("I/O worker executing I/O on behalf of process %d", owner_pid);
633 : : }
634 : :
635 : : /*
636 : : * Check if this backend is allowed to time out, and thus should use a
637 : : * non-infinite sleep time. Only the highest-numbered worker is allowed to
638 : : * time out, and only if the pool is above io_min_workers. Serializing
639 : : * timeouts keeps IDs in a range 0..N without gaps, and avoids undershooting
640 : : * io_min_workers.
641 : : *
642 : : * The result is only instantaneously true and may be temporarily inconsistent
643 : : * in different workers around transitions, but all workers are woken up on
644 : : * pool size or GUC changes making the result eventually consistent.
645 : : */
646 : : static bool
647 : 463876 : pgaio_worker_can_timeout(void)
648 : : {
649 : : PgAioWorkerSet workerset;
650 : :
651 [ + + ]: 463876 : if (MyIoWorkerId < io_min_workers)
652 : 447892 : return false;
653 : :
654 : : /* Serialize against pool size changes. */
655 : 15984 : LWLockAcquire(AioWorkerControlLock, LW_SHARED);
656 : 15984 : workerset = io_worker_control->workerset;
657 : 15984 : LWLockRelease(AioWorkerControlLock);
658 : :
659 [ + + ]: 15984 : if (MyIoWorkerId != pgaio_workerset_get_highest(&workerset))
660 : 163 : return false;
661 : :
662 : 15821 : return true;
663 : : }
664 : :
665 : : /*
666 : : * Emit a WARNING if io_min_workers > io_max_workers, since the worker
667 : : * pool will never exceed io_max_workers regardless of the minimum setting.
668 : : */
669 : : static void
670 : 1368 : check_io_worker_gucs(void)
671 : : {
672 : : /* Only do the check in one worker, to limit noise */
673 [ + + ]: 1368 : if (MyIoWorkerId != 0)
674 : 721 : return;
675 : :
676 [ - + ]: 647 : if (io_min_workers > io_max_workers)
677 [ # # ]: 0 : ereport(WARNING,
678 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
679 : : errmsg("\"%s\" (%d) should be less than or equal to \"%s\" (%d)",
680 : : "io_min_workers", io_min_workers,
681 : : "io_max_workers", io_max_workers),
682 : : errdetail("The I/O worker pool will not exceed \"%s\" (%d) workers.",
683 : : "io_max_workers", io_max_workers)));
684 : : }
685 : :
686 : : void
687 : 1325 : IoWorkerMain(const void *startup_data, size_t startup_data_len)
688 : : {
689 : : sigjmp_buf local_sigjmp_buf;
690 : 1325 : TimestampTz idle_timeout_abs = 0;
691 : 1325 : int timeout_guc_used = 0;
692 : 1325 : PgAioHandle *volatile error_ioh = NULL;
693 : 1325 : ErrorContextCallback errcallback = {0};
694 : 1325 : volatile int error_errno = 0;
695 : : char cmd[128];
696 : 1325 : int hist_ios = 0;
697 : 1325 : int hist_wakeups = 0;
698 : :
699 : 1325 : AuxiliaryProcessMainCommon();
700 : :
701 : 1325 : pqsignal(SIGHUP, SignalHandlerForConfigReload);
702 : 1325 : pqsignal(SIGINT, die); /* to allow manually triggering worker restart */
703 : :
704 : : /*
705 : : * Ignore SIGTERM, will get explicit shutdown via SIGUSR2 later in the
706 : : * shutdown sequence, similar to checkpointer.
707 : : */
708 : 1325 : pqsignal(SIGTERM, PG_SIG_IGN);
709 : : /* SIGQUIT handler was already set up by InitPostmasterChild */
710 : 1325 : pqsignal(SIGALRM, PG_SIG_IGN);
711 : 1325 : pqsignal(SIGPIPE, PG_SIG_IGN);
712 : 1325 : pqsignal(SIGUSR1, procsignal_sigusr1_handler);
713 : 1325 : pqsignal(SIGUSR2, SignalHandlerForShutdownRequest);
714 : :
715 : : /* also registers a shutdown callback to unregister */
716 : 1325 : pgaio_worker_register();
717 : :
718 : 1325 : check_io_worker_gucs();
719 : :
720 : 1325 : sprintf(cmd, "%d", MyIoWorkerId);
721 : 1325 : set_ps_display(cmd);
722 : :
723 : 1325 : errcallback.callback = pgaio_worker_error_callback;
724 : 1325 : errcallback.previous = error_context_stack;
725 : 1325 : error_context_stack = &errcallback;
726 : :
727 : : /* see PostgresMain() */
728 [ + + ]: 1325 : if (sigsetjmp(local_sigjmp_buf, 1) != 0)
729 : : {
730 : 1 : error_context_stack = NULL;
731 : 1 : HOLD_INTERRUPTS();
732 : :
733 : 1 : EmitErrorReport();
734 : :
735 : : /*
736 : : * In the - very unlikely - case that the IO failed in a way that
737 : : * raises an error we need to mark the IO as failed.
738 : : *
739 : : * Need to do just enough error recovery so that we can mark the IO as
740 : : * failed and then exit (postmaster will start a new worker).
741 : : */
742 : 1 : LWLockReleaseAll();
743 : :
744 [ + - ]: 1 : if (error_ioh != NULL)
745 : : {
746 : : /* should never fail without setting error_errno */
747 : : Assert(error_errno != 0);
748 : :
749 : 1 : errno = error_errno;
750 : :
751 : 1 : START_CRIT_SECTION();
752 : 1 : pgaio_io_process_completion(error_ioh, -error_errno);
753 : 1 : END_CRIT_SECTION();
754 : : }
755 : :
756 : 1 : proc_exit(1);
757 : : }
758 : :
759 : : /* We can now handle ereport(ERROR) */
760 : 1325 : PG_exception_stack = &local_sigjmp_buf;
761 : :
762 : 1325 : sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
763 : :
764 [ + + ]: 935272 : while (!ShutdownRequestPending)
765 : : {
766 : : uint32 io_index;
767 : 933989 : int worker = -1;
768 : 933989 : int queue_depth = 0;
769 : 933989 : bool maybe_grow = false;
770 : :
771 : : /*
772 : : * Try to get a job to do.
773 : : *
774 : : * The lwlock acquisition also provides the necessary memory barrier
775 : : * to ensure that we don't see an outdated data in the handle.
776 : : */
777 : 933989 : LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
778 [ + + ]: 933989 : if ((io_index = pgaio_worker_submission_queue_consume()) == -1)
779 : : {
780 : : /* Nothing to do. Mark self idle. */
781 : 463845 : pgaio_workerset_insert(&io_worker_control->idle_workerset,
782 : : MyIoWorkerId);
783 : : }
784 : : else
785 : : {
786 : : /* Got one. Clear idle flag. */
787 : 470144 : pgaio_workerset_remove(&io_worker_control->idle_workerset,
788 : : MyIoWorkerId);
789 : :
790 : : /*
791 : : * See if we should wake up a higher numbered peer. Only do that
792 : : * if this worker is not receiving spurious wakeups itself. The
793 : : * intention is create a frontier beyond which idle workers stay
794 : : * asleep.
795 : : *
796 : : * This heuristic tries to discover the useful wakeup propagation
797 : : * chain length when IOs are very fast and workers wake up to find
798 : : * that all IOs have already been taken.
799 : : *
800 : : * If we chose not to wake a worker when we ideally should have,
801 : : * then the ratio will soon change to correct that.
802 : : */
803 [ + + ]: 470144 : if (hist_wakeups <= hist_ios)
804 : : {
805 : 272237 : queue_depth = pgaio_worker_submission_queue_depth();
806 [ + + ]: 272237 : if (queue_depth > 0)
807 : : {
808 : : /* Choose a worker higher than me to wake. */
809 : 11242 : worker = pgaio_worker_choose_idle(MyIoWorkerId);
810 [ + + ]: 11242 : if (worker == -1)
811 : 8966 : maybe_grow = true;
812 : : }
813 : : }
814 : : }
815 : 933989 : LWLockRelease(AioWorkerSubmissionQueueLock);
816 : :
817 : : /* Propagate wakeups. */
818 [ + + ]: 933989 : if (worker != -1)
819 : : {
820 : 2276 : pgaio_worker_wake(worker);
821 : : }
822 [ + + ]: 931713 : else if (maybe_grow)
823 : : {
824 : : /*
825 : : * We know there was at least one more item in the queue, and we
826 : : * failed to find a higher-numbered idle worker to wake. Now we
827 : : * decide if we should try to start one more worker.
828 : : *
829 : : * We do this with a simple heuristic: is the queue depth greater
830 : : * than the current number of workers?
831 : : *
832 : : * Consider the following situations:
833 : : *
834 : : * 1. The queue depth is constantly increasing, because IOs are
835 : : * arriving faster than they can possibly be serviced. It doesn't
836 : : * matter much which threshold we choose, as we will surely hit
837 : : * it. Crossing the current worker count is a useful signal
838 : : * because it's clearly too deep to avoid queuing latency already,
839 : : * but still leaves a small window of opportunity to improve the
840 : : * situation before the queue overflows.
841 : : *
842 : : * 2. The worker pool is keeping up, no latency is being
843 : : * introduced and an extra worker would be a waste of resources.
844 : : * Queue depth distributions tend to be heavily skewed, with long
845 : : * tails of low probability spikes (due to submission clustering,
846 : : * scheduling, jitter, stalls, noisy neighbors, etc). We want a
847 : : * number that is very unlikely to be triggered by an outlier, and
848 : : * we bet that an exponential or similar distribution whose
849 : : * outliers never reach this threshold must be almost entirely
850 : : * concentrated at the low end. If we do see a spike as big as
851 : : * the worker count, we take it as a signal that the distribution
852 : : * is surely too wide.
853 : : *
854 : : * On its own, this is an extremely crude signal. When combined
855 : : * with the wakeup propagation test that precedes it (but on its
856 : : * own tends to overshoot) and io_worker_launch_interval, the
857 : : * result is that we gradually test each pool size until we find
858 : : * one that doesn't trigger further expansion, and then hold it
859 : : * for at least io_worker_idle_timeout.
860 : : *
861 : : * XXX Perhaps ideas from queueing theory or control theory could
862 : : * do a better job of this.
863 : : */
864 : :
865 : : /* Read nworkers without lock for this heuristic purpose. */
866 [ + + ]: 8966 : if (queue_depth > io_worker_control->nworkers)
867 : 46 : pgaio_worker_request_grow();
868 : : }
869 : :
870 [ + + ]: 933989 : if (io_index != -1)
871 : : {
872 : 470144 : PgAioHandle *ioh = NULL;
873 : :
874 : : /* Cancel timeout and update wakeup:work ratio. */
875 : 470144 : idle_timeout_abs = 0;
876 [ + + ]: 470144 : if (++hist_ios == PGAIO_WORKER_WAKEUP_RATIO_SATURATE)
877 : : {
878 : 136158 : hist_wakeups /= 2;
879 : 136158 : hist_ios /= 2;
880 : : }
881 : :
882 : 470144 : ioh = &pgaio_ctl->io_handles[io_index];
883 : 470144 : error_ioh = ioh;
884 : 470144 : errcallback.arg = ioh;
885 : :
886 [ - + ]: 470144 : pgaio_debug_io(DEBUG4, ioh,
887 : : "worker %d processing IO",
888 : : MyIoWorkerId);
889 : :
890 : : /*
891 : : * Prevent interrupts between pgaio_io_reopen() and
892 : : * pgaio_io_perform_synchronously() that otherwise could lead to
893 : : * the FD getting closed in that window.
894 : : */
895 : 470144 : HOLD_INTERRUPTS();
896 : :
897 : : /*
898 : : * It's very unlikely, but possible, that reopen fails. E.g. due
899 : : * to memory allocations failing or file permissions changing or
900 : : * such. In that case we need to fail the IO.
901 : : *
902 : : * There's not really a good errno we can report here.
903 : : */
904 : 470144 : error_errno = ENOENT;
905 : 470144 : pgaio_io_reopen(ioh);
906 : :
907 : : /*
908 : : * To be able to exercise the reopen-fails path, allow injection
909 : : * points to trigger a failure at this point.
910 : : */
911 : 470144 : INJECTION_POINT("aio-worker-after-reopen", ioh);
912 : :
913 : 470143 : error_errno = 0;
914 : 470143 : error_ioh = NULL;
915 : :
916 : : /*
917 : : * As part of IO completion the buffer will be marked as NOACCESS,
918 : : * until the buffer is pinned again - which never happens in io
919 : : * workers. Therefore the next time there is IO for the same
920 : : * buffer, the memory will be considered inaccessible. To avoid
921 : : * that, explicitly allow access to the memory before reading data
922 : : * into it.
923 : : */
924 : : #ifdef USE_VALGRIND
925 : : {
926 : : struct iovec *iov;
927 : : uint16 iov_length = pgaio_io_get_iovec_length(ioh, &iov);
928 : :
929 : : for (int i = 0; i < iov_length; i++)
930 : : VALGRIND_MAKE_MEM_UNDEFINED(iov[i].iov_base, iov[i].iov_len);
931 : : }
932 : : #endif
933 : :
934 : : #ifdef PGAIO_WORKER_SHOW_PS_INFO
935 : : {
936 : : char *description = pgaio_io_get_target_description(ioh);
937 : :
938 : : sprintf(cmd, "%d: [%s] %s",
939 : : MyIoWorkerId,
940 : : pgaio_io_get_op_name(ioh),
941 : : description);
942 : : pfree(description);
943 : : set_ps_display(cmd);
944 : : }
945 : : #endif
946 : :
947 : : /*
948 : : * We don't expect this to ever fail with ERROR or FATAL, no need
949 : : * to keep error_ioh set to the IO.
950 : : * pgaio_io_perform_synchronously() contains a critical section to
951 : : * ensure we don't accidentally fail.
952 : : */
953 : 470143 : pgaio_io_perform_synchronously(ioh);
954 : :
955 : 470143 : RESUME_INTERRUPTS();
956 : 470143 : errcallback.arg = NULL;
957 : : }
958 : : else
959 : : {
960 : : int timeout_ms;
961 : :
962 : : /* Cancel new worker request if pending. */
963 : 463845 : pgaio_worker_cancel_grow();
964 : :
965 : : /* Compute the remaining allowed idle time. */
966 [ - + ]: 463845 : if (io_worker_idle_timeout == -1)
967 : : {
968 : : /* Never time out. */
969 : 0 : timeout_ms = -1;
970 : : }
971 : : else
972 : : {
973 : 463845 : TimestampTz now = GetCurrentTimestamp();
974 : :
975 : : /* If the GUC changes, reset timer. */
976 [ + + ]: 463845 : if (idle_timeout_abs != 0 &&
977 [ - + ]: 2939 : io_worker_idle_timeout != timeout_guc_used)
978 : 0 : idle_timeout_abs = 0;
979 : :
980 : : /* Only the highest-numbered worker can time out. */
981 [ + + ]: 463845 : if (pgaio_worker_can_timeout())
982 : : {
983 [ + + ]: 15790 : if (idle_timeout_abs == 0)
984 : : {
985 : : /*
986 : : * I have just been promoted to the timeout worker, or
987 : : * the GUC changed. Compute new absolute time from
988 : : * now.
989 : : */
990 : 12851 : idle_timeout_abs =
991 : 12851 : TimestampTzPlusMilliseconds(now,
992 : : io_worker_idle_timeout);
993 : 12851 : timeout_guc_used = io_worker_idle_timeout;
994 : : }
995 : 15790 : timeout_ms =
996 : 15790 : TimestampDifferenceMilliseconds(now, idle_timeout_abs);
997 : : }
998 : : else
999 : : {
1000 : : /* No timeout for me. */
1001 : 448055 : idle_timeout_abs = 0;
1002 : 448055 : timeout_ms = -1;
1003 : : }
1004 : : }
1005 : :
1006 : : #ifdef PGAIO_WORKER_SHOW_PS_INFO
1007 : : sprintf(cmd, "%d: idle, wakeups:ios = %d:%d",
1008 : : MyIoWorkerId, hist_wakeups, hist_ios);
1009 : : set_ps_display(cmd);
1010 : : #endif
1011 : :
1012 [ + + ]: 463845 : if (WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH | WL_TIMEOUT,
1013 : : timeout_ms,
1014 : : WAIT_EVENT_IO_WORKER_MAIN) == WL_TIMEOUT)
1015 : : {
1016 : : /* WL_TIMEOUT */
1017 [ + - ]: 31 : if (pgaio_worker_can_timeout())
1018 [ + - ]: 31 : if (GetCurrentTimestamp() >= idle_timeout_abs)
1019 : 31 : break;
1020 : : }
1021 : : else
1022 : : {
1023 : : /* WL_LATCH_SET */
1024 [ + + ]: 463808 : if (++hist_wakeups == PGAIO_WORKER_WAKEUP_RATIO_SATURATE)
1025 : : {
1026 : 103263 : hist_wakeups /= 2;
1027 : 103263 : hist_ios /= 2;
1028 : : }
1029 : : }
1030 : 463808 : ResetLatch(MyLatch);
1031 : : }
1032 : :
1033 [ + + ]: 933951 : CHECK_FOR_INTERRUPTS();
1034 : :
1035 [ + + ]: 933947 : if (ConfigReloadPending)
1036 : : {
1037 : 199 : int io_max_workers_prev = io_max_workers;
1038 : 199 : int io_min_workers_prev = io_min_workers;
1039 : :
1040 : 199 : ConfigReloadPending = false;
1041 : 199 : ProcessConfigFile(PGC_SIGHUP);
1042 : :
1043 : : /*
1044 : : * Emit a WARNING if io_min_workers > io_max_workers. If no bound
1045 : : * has changed, skip this to avoid too many log messages.
1046 : : */
1047 [ + + ]: 199 : if (io_min_workers_prev != io_min_workers ||
1048 [ - + ]: 156 : io_max_workers_prev != io_max_workers)
1049 : 43 : check_io_worker_gucs();
1050 : :
1051 : : /* If io_max_workers has been decreased, exit highest first. */
1052 [ - + ]: 199 : if (MyIoWorkerId >= io_max_workers)
1053 : 0 : break;
1054 : : }
1055 : : }
1056 : :
1057 : 1314 : error_context_stack = errcallback.previous;
1058 : 1314 : proc_exit(0);
1059 : : }
1060 : :
1061 : : bool
1062 : 318762 : pgaio_workers_enabled(void)
1063 : : {
1064 : 318762 : return io_method == IOMETHOD_WORKER;
1065 : : }
|
