Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * autoprewarm.c
4 : : * Periodically dump information about the blocks present in
5 : : * shared_buffers, and reload them on server restart.
6 : : *
7 : : * Due to locking considerations, we can't actually begin prewarming
8 : : * until the server reaches a consistent state. We need the catalogs
9 : : * to be consistent so that we can figure out which relation to lock,
10 : : * and we need to lock the relations so that we don't try to prewarm
11 : : * pages from a relation that is in the process of being dropped.
12 : : *
13 : : * While prewarming, autoprewarm will use two workers. There's a
14 : : * leader worker that reads and sorts the list of blocks to be
15 : : * prewarmed and then launches a per-database worker for each
16 : : * relevant database in turn. The former keeps running after the
17 : : * initial prewarm is complete to update the dump file periodically.
18 : : *
19 : : * Copyright (c) 2016-2026, PostgreSQL Global Development Group
20 : : *
21 : : * IDENTIFICATION
22 : : * contrib/pg_prewarm/autoprewarm.c
23 : : *
24 : : *-------------------------------------------------------------------------
25 : : */
26 : :
27 : : #include "postgres.h"
28 : :
29 : : #include <unistd.h>
30 : :
31 : : #include "access/relation.h"
32 : : #include "access/xact.h"
33 : : #include "pgstat.h"
34 : : #include "postmaster/bgworker.h"
35 : : #include "postmaster/interrupt.h"
36 : : #include "storage/buf_internals.h"
37 : : #include "storage/dsm.h"
38 : : #include "storage/dsm_registry.h"
39 : : #include "storage/fd.h"
40 : : #include "storage/ipc.h"
41 : : #include "storage/latch.h"
42 : : #include "storage/lwlock.h"
43 : : #include "storage/procsignal.h"
44 : : #include "storage/read_stream.h"
45 : : #include "storage/smgr.h"
46 : : #include "tcop/tcopprot.h"
47 : : #include "utils/guc.h"
48 : : #include "utils/rel.h"
49 : : #include "utils/relfilenumbermap.h"
50 : : #include "utils/timestamp.h"
51 : : #include "utils/wait_event.h"
52 : :
53 : : #define AUTOPREWARM_FILE "autoprewarm.blocks"
54 : :
55 : : /* Metadata for each block we dump. */
56 : : typedef struct BlockInfoRecord
57 : : {
58 : : Oid database;
59 : : Oid tablespace;
60 : : RelFileNumber filenumber;
61 : : ForkNumber forknum;
62 : : BlockNumber blocknum;
63 : : } BlockInfoRecord;
64 : :
65 : : /* Shared state information for autoprewarm bgworker. */
66 : : typedef struct AutoPrewarmSharedState
67 : : {
68 : : LWLock lock; /* mutual exclusion */
69 : : pid_t bgworker_pid; /* for main bgworker */
70 : : pid_t pid_using_dumpfile; /* for autoprewarm or block dump */
71 : :
72 : : /* Following items are for communication with per-database worker */
73 : : dsm_handle block_info_handle;
74 : : Oid database;
75 : : int prewarm_start_idx;
76 : : int prewarm_stop_idx;
77 : : int prewarmed_blocks;
78 : : } AutoPrewarmSharedState;
79 : :
80 : : /*
81 : : * Private data passed through the read stream API for our use in the
82 : : * callback.
83 : : */
84 : : typedef struct AutoPrewarmReadStreamData
85 : : {
86 : : /* The array of records containing the blocks we should prewarm. */
87 : : BlockInfoRecord *block_info;
88 : :
89 : : /*
90 : : * pos is the read stream callback's index into block_info. Because the
91 : : * read stream may read ahead, pos is likely to be ahead of the index in
92 : : * the main loop in autoprewarm_database_main().
93 : : */
94 : : int pos;
95 : : Oid tablespace;
96 : : RelFileNumber filenumber;
97 : : ForkNumber forknum;
98 : : BlockNumber nblocks;
99 : : } AutoPrewarmReadStreamData;
100 : :
101 : :
102 : : PGDLLEXPORT void autoprewarm_main(Datum main_arg);
103 : : PGDLLEXPORT void autoprewarm_database_main(Datum main_arg);
104 : :
105 : 3 : PG_FUNCTION_INFO_V1(autoprewarm_start_worker);
106 : 4 : PG_FUNCTION_INFO_V1(autoprewarm_dump_now);
107 : :
108 : : static void apw_load_buffers(void);
109 : : static int apw_dump_now(bool is_bgworker, bool dump_unlogged);
110 : : static void apw_start_leader_worker(void);
111 : : static void apw_start_database_worker(void);
112 : : static bool apw_init_shmem(void);
113 : : static void apw_detach_shmem(int code, Datum arg);
114 : : static int apw_compare_blockinfo(const void *p, const void *q);
115 : :
116 : : /* Pointer to shared-memory state. */
117 : : static AutoPrewarmSharedState *apw_state = NULL;
118 : :
119 : : /* GUC variables. */
120 : : static bool autoprewarm = true; /* start worker? */
121 : : static int autoprewarm_interval = 300; /* dump interval */
122 : :
123 : : /*
124 : : * Module load callback.
125 : : */
126 : : void
127 : 6 : _PG_init(void)
128 : : {
129 : 6 : DefineCustomIntVariable("pg_prewarm.autoprewarm_interval",
130 : : "Sets the interval between dumps of shared buffers",
131 : : "If set to zero, time-based dumping is disabled.",
132 : : &autoprewarm_interval,
133 : : 300,
134 : : 0, INT_MAX / 1000,
135 : : PGC_SIGHUP,
136 : : GUC_UNIT_S,
137 : : NULL,
138 : : NULL,
139 : : NULL);
140 : :
141 [ + + ]: 6 : if (!process_shared_preload_libraries_in_progress)
142 : 4 : return;
143 : :
144 : : /* can't define PGC_POSTMASTER variable after startup */
145 : 2 : DefineCustomBoolVariable("pg_prewarm.autoprewarm",
146 : : "Starts the autoprewarm worker.",
147 : : NULL,
148 : : &autoprewarm,
149 : : true,
150 : : PGC_POSTMASTER,
151 : : 0,
152 : : NULL,
153 : : NULL,
154 : : NULL);
155 : :
156 : 2 : MarkGUCPrefixReserved("pg_prewarm");
157 : :
158 : : /* Register autoprewarm worker, if enabled. */
159 [ + - ]: 2 : if (autoprewarm)
160 : 2 : apw_start_leader_worker();
161 : : }
162 : :
163 : : /*
164 : : * Main entry point for the leader autoprewarm process. Per-database workers
165 : : * have a separate entry point.
166 : : */
167 : : void
168 : 2 : autoprewarm_main(Datum main_arg)
169 : : {
170 : 2 : bool first_time = true;
171 : 2 : bool final_dump_allowed = true;
172 : 2 : TimestampTz last_dump_time = 0;
173 : :
174 : : /* Establish signal handlers; once that's done, unblock signals. */
175 : 2 : pqsignal(SIGTERM, SignalHandlerForShutdownRequest);
176 : 2 : pqsignal(SIGHUP, SignalHandlerForConfigReload);
177 : 2 : pqsignal(SIGUSR1, procsignal_sigusr1_handler);
178 : 2 : BackgroundWorkerUnblockSignals();
179 : :
180 : : /* Create (if necessary) and attach to our shared memory area. */
181 [ - + ]: 2 : if (apw_init_shmem())
182 : 0 : first_time = false;
183 : :
184 : : /*
185 : : * Set on-detach hook so that our PID will be cleared on exit.
186 : : *
187 : : * NB: Autoprewarm's state is stored in a DSM segment, and DSM segments
188 : : * are detached before calling the on_shmem_exit callbacks, so we must put
189 : : * apw_detach_shmem in the before_shmem_exit callback list.
190 : : */
191 : 2 : before_shmem_exit(apw_detach_shmem, 0);
192 : :
193 : : /*
194 : : * Store our PID in the shared memory area --- unless there's already
195 : : * another worker running, in which case just exit.
196 : : */
197 : 2 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
198 [ - + ]: 2 : if (apw_state->bgworker_pid != InvalidPid)
199 : : {
200 : 0 : LWLockRelease(&apw_state->lock);
201 [ # # ]: 0 : ereport(LOG,
202 : : (errmsg("autoprewarm worker is already running under PID %d",
203 : : (int) apw_state->bgworker_pid)));
204 : 0 : return;
205 : : }
206 : 2 : apw_state->bgworker_pid = MyProcPid;
207 : 2 : LWLockRelease(&apw_state->lock);
208 : :
209 : : /*
210 : : * Preload buffers from the dump file only if we just created the shared
211 : : * memory region. Otherwise, it's either already been done or shouldn't
212 : : * be done - e.g. because the old dump file has been overwritten since the
213 : : * server was started.
214 : : *
215 : : * There's not much point in performing a dump immediately after we finish
216 : : * preloading; so, if we do end up preloading, consider the last dump time
217 : : * to be equal to the current time.
218 : : *
219 : : * If apw_load_buffers() is terminated early by a shutdown request,
220 : : * prevent dumping out our state below the loop, because we'd effectively
221 : : * just truncate the saved state to however much we'd managed to preload.
222 : : */
223 [ + - ]: 2 : if (first_time)
224 : : {
225 : 2 : apw_load_buffers();
226 : 2 : final_dump_allowed = !ShutdownRequestPending;
227 : 2 : last_dump_time = GetCurrentTimestamp();
228 : : }
229 : :
230 : : /* Periodically dump buffers until terminated. */
231 [ + + ]: 5 : while (!ShutdownRequestPending)
232 : : {
233 : : /* In case of a SIGHUP, just reload the configuration. */
234 [ - + ]: 3 : if (ConfigReloadPending)
235 : : {
236 : 0 : ConfigReloadPending = false;
237 : 0 : ProcessConfigFile(PGC_SIGHUP);
238 : : }
239 : :
240 [ + - ]: 3 : if (autoprewarm_interval <= 0)
241 : : {
242 : : /* We're only dumping at shutdown, so just wait forever. */
243 : 3 : (void) WaitLatch(MyLatch,
244 : : WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
245 : : -1L,
246 : : PG_WAIT_EXTENSION);
247 : : }
248 : : else
249 : : {
250 : : TimestampTz next_dump_time;
251 : : long delay_in_ms;
252 : :
253 : : /* Compute the next dump time. */
254 : 0 : next_dump_time =
255 : 0 : TimestampTzPlusMilliseconds(last_dump_time,
256 : : autoprewarm_interval * 1000);
257 : : delay_in_ms =
258 : 0 : TimestampDifferenceMilliseconds(GetCurrentTimestamp(),
259 : : next_dump_time);
260 : :
261 : : /* Perform a dump if it's time. */
262 [ # # ]: 0 : if (delay_in_ms <= 0)
263 : : {
264 : 0 : last_dump_time = GetCurrentTimestamp();
265 : 0 : apw_dump_now(true, false);
266 : 0 : continue;
267 : : }
268 : :
269 : : /* Sleep until the next dump time. */
270 : 0 : (void) WaitLatch(MyLatch,
271 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
272 : : delay_in_ms,
273 : : PG_WAIT_EXTENSION);
274 : : }
275 : :
276 : : /* Reset the latch, loop. */
277 : 3 : ResetLatch(MyLatch);
278 : : }
279 : :
280 : : /*
281 : : * Dump one last time. We assume this is probably the result of a system
282 : : * shutdown, although it's possible that we've merely been terminated.
283 : : */
284 [ + - ]: 2 : if (final_dump_allowed)
285 : 2 : apw_dump_now(true, true);
286 : : }
287 : :
288 : : /*
289 : : * Read the dump file and launch per-database workers one at a time to
290 : : * prewarm the buffers found there.
291 : : */
292 : : static void
293 : 2 : apw_load_buffers(void)
294 : : {
295 : 2 : FILE *file = NULL;
296 : : int num_elements,
297 : : i;
298 : : BlockInfoRecord *blkinfo;
299 : : dsm_segment *seg;
300 : :
301 : : /*
302 : : * Skip the prewarm if the dump file is in use; otherwise, prevent any
303 : : * other process from writing it while we're using it.
304 : : */
305 : 2 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
306 [ + - ]: 2 : if (apw_state->pid_using_dumpfile == InvalidPid)
307 : 2 : apw_state->pid_using_dumpfile = MyProcPid;
308 : : else
309 : : {
310 : 0 : LWLockRelease(&apw_state->lock);
311 [ # # ]: 0 : ereport(LOG,
312 : : (errmsg("skipping prewarm because block dump file is being written by PID %d",
313 : : (int) apw_state->pid_using_dumpfile)));
314 : 1 : return;
315 : : }
316 : 2 : LWLockRelease(&apw_state->lock);
317 : :
318 : : /*
319 : : * Open the block dump file. Exit quietly if it doesn't exist, but report
320 : : * any other error.
321 : : */
322 : 2 : file = AllocateFile(AUTOPREWARM_FILE, "r");
323 [ + + ]: 2 : if (!file)
324 : : {
325 [ + - ]: 1 : if (errno == ENOENT)
326 : : {
327 : 1 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
328 : 1 : apw_state->pid_using_dumpfile = InvalidPid;
329 : 1 : LWLockRelease(&apw_state->lock);
330 : 1 : return; /* No file to load. */
331 : : }
332 [ # # ]: 0 : ereport(ERROR,
333 : : (errcode_for_file_access(),
334 : : errmsg("could not read file \"%s\": %m",
335 : : AUTOPREWARM_FILE)));
336 : : }
337 : :
338 : : /* First line of the file is a record count. */
339 [ - + ]: 1 : if (fscanf(file, "<<%d>>\n", &num_elements) != 1)
340 [ # # ]: 0 : ereport(ERROR,
341 : : (errcode_for_file_access(),
342 : : errmsg("could not read from file \"%s\": %m",
343 : : AUTOPREWARM_FILE)));
344 : :
345 : : /* Allocate a dynamic shared memory segment to store the record data. */
346 : 1 : seg = dsm_create(sizeof(BlockInfoRecord) * num_elements, 0);
347 : 1 : blkinfo = (BlockInfoRecord *) dsm_segment_address(seg);
348 : :
349 : : /* Read records, one per line. */
350 [ + + ]: 238 : for (i = 0; i < num_elements; i++)
351 : : {
352 : : unsigned forknum;
353 : :
354 [ - + ]: 237 : if (fscanf(file, "%u,%u,%u,%u,%u\n", &blkinfo[i].database,
355 : 237 : &blkinfo[i].tablespace, &blkinfo[i].filenumber,
356 : 237 : &forknum, &blkinfo[i].blocknum) != 5)
357 [ # # ]: 0 : ereport(ERROR,
358 : : (errmsg("autoprewarm block dump file is corrupted at line %d",
359 : : i + 1)));
360 : 237 : blkinfo[i].forknum = forknum;
361 : : }
362 : :
363 : 1 : FreeFile(file);
364 : :
365 : : /* Sort the blocks to be loaded. */
366 : 1 : qsort(blkinfo, num_elements, sizeof(BlockInfoRecord),
367 : : apw_compare_blockinfo);
368 : :
369 : : /* Populate shared memory state. */
370 : 1 : apw_state->block_info_handle = dsm_segment_handle(seg);
371 : 1 : apw_state->prewarm_start_idx = apw_state->prewarm_stop_idx = 0;
372 : 1 : apw_state->prewarmed_blocks = 0;
373 : :
374 : : /* Don't prewarm more than we can fit. */
375 [ - + ]: 1 : if (num_elements > NBuffers)
376 : : {
377 : 0 : num_elements = NBuffers;
378 [ # # ]: 0 : ereport(LOG,
379 : : (errmsg("autoprewarm capping prewarmed blocks to %d (shared_buffers size)",
380 : : NBuffers)));
381 : : }
382 : :
383 : : /* Get the info position of the first block of the next database. */
384 [ + + ]: 2 : while (apw_state->prewarm_start_idx < num_elements)
385 : : {
386 : 1 : int j = apw_state->prewarm_start_idx;
387 : 1 : Oid current_db = blkinfo[j].database;
388 : :
389 : : /*
390 : : * Advance the prewarm_stop_idx to the first BlockInfoRecord that does
391 : : * not belong to this database.
392 : : */
393 : 1 : j++;
394 [ + + ]: 237 : while (j < num_elements)
395 : : {
396 [ + + ]: 236 : if (current_db != blkinfo[j].database)
397 : : {
398 : : /*
399 : : * Combine BlockInfoRecords for global objects with those of
400 : : * the database.
401 : : */
402 [ - + ]: 1 : if (current_db != InvalidOid)
403 : 0 : break;
404 : 1 : current_db = blkinfo[j].database;
405 : : }
406 : :
407 : 236 : j++;
408 : : }
409 : :
410 : : /*
411 : : * If we reach this point with current_db == InvalidOid, then only
412 : : * BlockInfoRecords belonging to global objects exist. We can't
413 : : * prewarm without a database connection, so just bail out.
414 : : */
415 [ - + ]: 1 : if (current_db == InvalidOid)
416 : 0 : break;
417 : :
418 : : /* Configure stop point and database for next per-database worker. */
419 : 1 : apw_state->prewarm_stop_idx = j;
420 : 1 : apw_state->database = current_db;
421 : : Assert(apw_state->prewarm_start_idx < apw_state->prewarm_stop_idx);
422 : :
423 : : /*
424 : : * Likewise, don't launch if we've already been told to shut down.
425 : : * (The launch would fail anyway, but we might as well skip it.)
426 : : */
427 [ - + ]: 1 : if (ShutdownRequestPending)
428 : 0 : break;
429 : :
430 : : /*
431 : : * Start a per-database worker to load blocks for this database; this
432 : : * function will return once the per-database worker exits.
433 : : */
434 : 1 : apw_start_database_worker();
435 : :
436 : : /* Prepare for next database. */
437 : 1 : apw_state->prewarm_start_idx = apw_state->prewarm_stop_idx;
438 : : }
439 : :
440 : : /* Clean up. */
441 : 1 : dsm_detach(seg);
442 : 1 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
443 : 1 : apw_state->block_info_handle = DSM_HANDLE_INVALID;
444 : 1 : apw_state->pid_using_dumpfile = InvalidPid;
445 : 1 : LWLockRelease(&apw_state->lock);
446 : :
447 : : /* Report our success, if we were able to finish. */
448 [ + - ]: 1 : if (!ShutdownRequestPending)
449 [ + - ]: 1 : ereport(LOG,
450 : : (errmsg("autoprewarm successfully prewarmed %d of %d previously-loaded blocks",
451 : : apw_state->prewarmed_blocks, num_elements)));
452 : : }
453 : :
454 : : /*
455 : : * Return the next block number of a specific relation and fork to read
456 : : * according to the array of BlockInfoRecord.
457 : : */
458 : : static BlockNumber
459 : 312 : apw_read_stream_next_block(ReadStream *stream,
460 : : void *callback_private_data,
461 : : void *per_buffer_data)
462 : : {
463 : 312 : AutoPrewarmReadStreamData *p = callback_private_data;
464 : :
465 [ - + ]: 312 : CHECK_FOR_INTERRUPTS();
466 : :
467 [ + + ]: 312 : while (p->pos < apw_state->prewarm_stop_idx)
468 : : {
469 : 311 : BlockInfoRecord blk = p->block_info[p->pos];
470 : :
471 [ + + ]: 311 : if (blk.tablespace != p->tablespace)
472 : 311 : return InvalidBlockNumber;
473 : :
474 [ + + ]: 310 : if (blk.filenumber != p->filenumber)
475 : 55 : return InvalidBlockNumber;
476 : :
477 [ + + ]: 255 : if (blk.forknum != p->forknum)
478 : 18 : return InvalidBlockNumber;
479 : :
480 : 237 : p->pos++;
481 : :
482 : : /*
483 : : * Check whether blocknum is valid and within fork file size.
484 : : * Fast-forward through any invalid blocks. We want p->pos to reflect
485 : : * the location of the next relation or fork before ending the stream.
486 : : */
487 [ - + ]: 237 : if (blk.blocknum >= p->nblocks)
488 : 0 : continue;
489 : :
490 : 237 : return blk.blocknum;
491 : : }
492 : :
493 : 1 : return InvalidBlockNumber;
494 : : }
495 : :
496 : : /*
497 : : * Prewarm all blocks for one database (and possibly also global objects, if
498 : : * those got grouped with this database).
499 : : */
500 : : void
501 : 1 : autoprewarm_database_main(Datum main_arg)
502 : : {
503 : : BlockInfoRecord *block_info;
504 : : int i;
505 : : BlockInfoRecord blk;
506 : : dsm_segment *seg;
507 : :
508 : : /* Establish signal handlers; once that's done, unblock signals. */
509 : 1 : pqsignal(SIGTERM, die);
510 : 1 : BackgroundWorkerUnblockSignals();
511 : :
512 : : /* Connect to correct database and get block information. */
513 : 1 : apw_init_shmem();
514 : 1 : seg = dsm_attach(apw_state->block_info_handle);
515 [ - + ]: 1 : if (seg == NULL)
516 [ # # ]: 0 : ereport(ERROR,
517 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
518 : : errmsg("could not map dynamic shared memory segment")));
519 : 1 : BackgroundWorkerInitializeConnectionByOid(apw_state->database, InvalidOid, 0);
520 : 1 : block_info = (BlockInfoRecord *) dsm_segment_address(seg);
521 : :
522 : 1 : i = apw_state->prewarm_start_idx;
523 : 1 : blk = block_info[i];
524 : :
525 : : /*
526 : : * Loop until we run out of blocks to prewarm or until we run out of
527 : : * buffers.
528 : : */
529 [ + + ]: 58 : while (i < apw_state->prewarm_stop_idx)
530 : : {
531 : 57 : Oid tablespace = blk.tablespace;
532 : 57 : RelFileNumber filenumber = blk.filenumber;
533 : : Oid reloid;
534 : : Relation rel;
535 : :
536 : : /*
537 : : * All blocks between prewarm_start_idx and prewarm_stop_idx should
538 : : * belong either to global objects or the same database.
539 : : */
540 : : Assert(blk.database == apw_state->database || blk.database == 0);
541 : :
542 : 57 : StartTransactionCommand();
543 : :
544 : 57 : reloid = RelidByRelfilenumber(blk.tablespace, blk.filenumber);
545 [ + - - + ]: 114 : if (!OidIsValid(reloid) ||
546 : 57 : (rel = try_relation_open(reloid, AccessShareLock)) == NULL)
547 : : {
548 : : /* We failed to open the relation, so there is nothing to close. */
549 : 0 : CommitTransactionCommand();
550 : :
551 : : /*
552 : : * Fast-forward to the next relation. We want to skip all of the
553 : : * other records referencing this relation since we know we can't
554 : : * open it. That way, we avoid repeatedly trying and failing to
555 : : * open the same relation.
556 : : */
557 [ # # ]: 0 : for (; i < apw_state->prewarm_stop_idx; i++)
558 : : {
559 : 0 : blk = block_info[i];
560 [ # # ]: 0 : if (blk.tablespace != tablespace ||
561 [ # # ]: 0 : blk.filenumber != filenumber)
562 : : break;
563 : : }
564 : :
565 : : /* Time to try and open our newfound relation */
566 : 0 : continue;
567 : : }
568 : :
569 : : /*
570 : : * We have a relation; now let's loop until we find a valid fork of
571 : : * the relation or we run out of buffers. Once we've read from all
572 : : * valid forks or run out of options, we'll close the relation and
573 : : * move on.
574 : : */
575 [ + + ]: 132 : while (i < apw_state->prewarm_stop_idx)
576 : : {
577 : : ForkNumber forknum;
578 : : BlockNumber nblocks;
579 : : struct AutoPrewarmReadStreamData p;
580 : : ReadStream *stream;
581 : : Buffer buf;
582 : :
583 : 131 : blk = block_info[i];
584 : :
585 : : /* Stop when we reach a different relation. */
586 [ + + ]: 131 : if (blk.tablespace != tablespace ||
587 [ + + ]: 130 : blk.filenumber != filenumber)
588 : : break;
589 : :
590 : 75 : forknum = blk.forknum;
591 : :
592 : : /*
593 : : * smgrexists is not safe for illegal forknum, hence check whether
594 : : * the passed forknum is valid before using it in smgrexists.
595 : : */
596 [ + - ]: 75 : if (blk.forknum <= InvalidForkNumber ||
597 [ + - ]: 75 : blk.forknum > MAX_FORKNUM ||
598 [ - + ]: 75 : !smgrexists(RelationGetSmgr(rel), blk.forknum))
599 : : {
600 : : /*
601 : : * Fast-forward to the next fork. We want to skip all of the
602 : : * other records referencing this fork since we already know
603 : : * it's not valid.
604 : : */
605 [ # # ]: 0 : for (; i < apw_state->prewarm_stop_idx; i++)
606 : : {
607 : 0 : blk = block_info[i];
608 [ # # ]: 0 : if (blk.tablespace != tablespace ||
609 [ # # ]: 0 : blk.filenumber != filenumber ||
610 [ # # ]: 0 : blk.forknum != forknum)
611 : : break;
612 : : }
613 : :
614 : : /* Time to check if this newfound fork is valid */
615 : 0 : continue;
616 : : }
617 : :
618 : 75 : nblocks = RelationGetNumberOfBlocksInFork(rel, blk.forknum);
619 : :
620 : 75 : p = (struct AutoPrewarmReadStreamData)
621 : : {
622 : : .block_info = block_info,
623 : : .pos = i,
624 : : .tablespace = tablespace,
625 : : .filenumber = filenumber,
626 : : .forknum = forknum,
627 : : .nblocks = nblocks,
628 : : };
629 : :
630 : 75 : stream = read_stream_begin_relation(READ_STREAM_MAINTENANCE |
631 : : READ_STREAM_DEFAULT |
632 : : READ_STREAM_USE_BATCHING,
633 : : NULL,
634 : : rel,
635 : : p.forknum,
636 : : apw_read_stream_next_block,
637 : : &p,
638 : : 0);
639 : :
640 : : /*
641 : : * Loop until we've prewarmed all the blocks from this fork. The
642 : : * read stream callback will check that we still have free buffers
643 : : * before requesting each block from the read stream API.
644 : : */
645 [ + + ]: 312 : while ((buf = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
646 : : {
647 : 237 : apw_state->prewarmed_blocks++;
648 : 237 : ReleaseBuffer(buf);
649 : : }
650 : :
651 : 75 : read_stream_end(stream);
652 : :
653 : : /*
654 : : * Advance i past all the blocks just prewarmed. Note that the
655 : : * callback might have advanced the index beyond the last valid
656 : : * block, so don't access block_info[i] yet.
657 : : */
658 : 75 : i = p.pos;
659 : : }
660 : :
661 : 57 : relation_close(rel, AccessShareLock);
662 : 57 : CommitTransactionCommand();
663 : : }
664 : :
665 : 1 : dsm_detach(seg);
666 : 1 : }
667 : :
668 : : /*
669 : : * Dump information on blocks in shared buffers. We use a text format here
670 : : * so that it's easy to understand and even change the file contents if
671 : : * necessary.
672 : : * Returns the number of blocks dumped.
673 : : */
674 : : static int
675 : 3 : apw_dump_now(bool is_bgworker, bool dump_unlogged)
676 : : {
677 : : int num_blocks;
678 : : int i;
679 : : int ret;
680 : : BlockInfoRecord *block_info_array;
681 : : BufferDesc *bufHdr;
682 : : FILE *file;
683 : : char transient_dump_file_path[MAXPGPATH];
684 : : pid_t pid;
685 : :
686 : 3 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
687 : 3 : pid = apw_state->pid_using_dumpfile;
688 [ + - ]: 3 : if (apw_state->pid_using_dumpfile == InvalidPid)
689 : 3 : apw_state->pid_using_dumpfile = MyProcPid;
690 : 3 : LWLockRelease(&apw_state->lock);
691 : :
692 [ - + ]: 3 : if (pid != InvalidPid)
693 : : {
694 [ # # ]: 0 : if (!is_bgworker)
695 [ # # ]: 0 : ereport(ERROR,
696 : : (errmsg("could not perform block dump because dump file is being used by PID %d",
697 : : (int) apw_state->pid_using_dumpfile)));
698 : :
699 [ # # ]: 0 : ereport(LOG,
700 : : (errmsg("skipping block dump because it is already being performed by PID %d",
701 : : (int) apw_state->pid_using_dumpfile)));
702 : 0 : return 0;
703 : : }
704 : :
705 : : /*
706 : : * With sufficiently large shared_buffers, allocation will exceed 1GB, so
707 : : * allow for a huge allocation to prevent outright failure.
708 : : *
709 : : * (In the future, it might be a good idea to redesign this to use a more
710 : : * memory-efficient data structure.)
711 : : */
712 : : block_info_array = (BlockInfoRecord *)
713 : 3 : palloc_extended((sizeof(BlockInfoRecord) * NBuffers), MCXT_ALLOC_HUGE);
714 : :
715 [ + + ]: 49155 : for (num_blocks = 0, i = 0; i < NBuffers; i++)
716 : : {
717 : : uint64 buf_state;
718 : :
719 [ - + ]: 49152 : CHECK_FOR_INTERRUPTS();
720 : :
721 : 49152 : bufHdr = GetBufferDescriptor(i);
722 : :
723 : : /* Lock each buffer header before inspecting. */
724 : 49152 : buf_state = LockBufHdr(bufHdr);
725 : :
726 : : /*
727 : : * Unlogged tables will be automatically truncated after a crash or
728 : : * unclean shutdown. In such cases we need not prewarm them. Dump them
729 : : * only if requested by caller.
730 : : */
731 [ + + ]: 49152 : if (buf_state & BM_TAG_VALID &&
732 [ - + - - ]: 713 : ((buf_state & BM_PERMANENT) || dump_unlogged))
733 : : {
734 : 713 : block_info_array[num_blocks].database = bufHdr->tag.dbOid;
735 : 713 : block_info_array[num_blocks].tablespace = bufHdr->tag.spcOid;
736 : 1426 : block_info_array[num_blocks].filenumber =
737 : 713 : BufTagGetRelNumber(&bufHdr->tag);
738 : 1426 : block_info_array[num_blocks].forknum =
739 : 713 : BufTagGetForkNum(&bufHdr->tag);
740 : 713 : block_info_array[num_blocks].blocknum = bufHdr->tag.blockNum;
741 : 713 : ++num_blocks;
742 : : }
743 : :
744 : 49152 : UnlockBufHdr(bufHdr);
745 : : }
746 : :
747 : 3 : snprintf(transient_dump_file_path, MAXPGPATH, "%s.tmp", AUTOPREWARM_FILE);
748 : 3 : file = AllocateFile(transient_dump_file_path, "w");
749 [ - + ]: 3 : if (!file)
750 [ # # ]: 0 : ereport(ERROR,
751 : : (errcode_for_file_access(),
752 : : errmsg("could not open file \"%s\": %m",
753 : : transient_dump_file_path)));
754 : :
755 : 3 : ret = fprintf(file, "<<%d>>\n", num_blocks);
756 [ - + ]: 3 : if (ret < 0)
757 : : {
758 : 0 : int save_errno = errno;
759 : :
760 : 0 : FreeFile(file);
761 : 0 : unlink(transient_dump_file_path);
762 : 0 : errno = save_errno;
763 [ # # ]: 0 : ereport(ERROR,
764 : : (errcode_for_file_access(),
765 : : errmsg("could not write to file \"%s\": %m",
766 : : transient_dump_file_path)));
767 : : }
768 : :
769 [ + + ]: 716 : for (i = 0; i < num_blocks; i++)
770 : : {
771 [ - + ]: 713 : CHECK_FOR_INTERRUPTS();
772 : :
773 : 713 : ret = fprintf(file, "%u,%u,%u,%u,%u\n",
774 : 713 : block_info_array[i].database,
775 : 713 : block_info_array[i].tablespace,
776 : 713 : block_info_array[i].filenumber,
777 : 713 : (uint32) block_info_array[i].forknum,
778 : 713 : block_info_array[i].blocknum);
779 [ - + ]: 713 : if (ret < 0)
780 : : {
781 : 0 : int save_errno = errno;
782 : :
783 : 0 : FreeFile(file);
784 : 0 : unlink(transient_dump_file_path);
785 : 0 : errno = save_errno;
786 [ # # ]: 0 : ereport(ERROR,
787 : : (errcode_for_file_access(),
788 : : errmsg("could not write to file \"%s\": %m",
789 : : transient_dump_file_path)));
790 : : }
791 : : }
792 : :
793 : 3 : pfree(block_info_array);
794 : :
795 : : /*
796 : : * Rename transient_dump_file_path to AUTOPREWARM_FILE to make things
797 : : * permanent.
798 : : */
799 : 3 : ret = FreeFile(file);
800 [ - + ]: 3 : if (ret != 0)
801 : : {
802 : 0 : int save_errno = errno;
803 : :
804 : 0 : unlink(transient_dump_file_path);
805 : 0 : errno = save_errno;
806 [ # # ]: 0 : ereport(ERROR,
807 : : (errcode_for_file_access(),
808 : : errmsg("could not close file \"%s\": %m",
809 : : transient_dump_file_path)));
810 : : }
811 : :
812 : 3 : (void) durable_rename(transient_dump_file_path, AUTOPREWARM_FILE, ERROR);
813 : 3 : apw_state->pid_using_dumpfile = InvalidPid;
814 : :
815 [ - + ]: 3 : ereport(DEBUG1,
816 : : (errmsg_internal("wrote block details for %d blocks", num_blocks)));
817 : 3 : return num_blocks;
818 : : }
819 : :
820 : : /*
821 : : * SQL-callable function to launch autoprewarm.
822 : : */
823 : : Datum
824 : 0 : autoprewarm_start_worker(PG_FUNCTION_ARGS)
825 : : {
826 : : pid_t pid;
827 : :
828 [ # # ]: 0 : if (!autoprewarm)
829 [ # # ]: 0 : ereport(ERROR,
830 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
831 : : errmsg("autoprewarm is disabled")));
832 : :
833 : 0 : apw_init_shmem();
834 : 0 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
835 : 0 : pid = apw_state->bgworker_pid;
836 : 0 : LWLockRelease(&apw_state->lock);
837 : :
838 [ # # ]: 0 : if (pid != InvalidPid)
839 [ # # ]: 0 : ereport(ERROR,
840 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
841 : : errmsg("autoprewarm worker is already running under PID %d",
842 : : (int) pid)));
843 : :
844 : 0 : apw_start_leader_worker();
845 : :
846 : 0 : PG_RETURN_VOID();
847 : : }
848 : :
849 : : /*
850 : : * SQL-callable function to perform an immediate block dump.
851 : : *
852 : : * Note: this is declared to return int8, as insurance against some
853 : : * very distant day when we might make NBuffers wider than int.
854 : : */
855 : : Datum
856 : 1 : autoprewarm_dump_now(PG_FUNCTION_ARGS)
857 : : {
858 : : int num_blocks;
859 : :
860 : 1 : apw_init_shmem();
861 : :
862 [ + - ]: 1 : PG_ENSURE_ERROR_CLEANUP(apw_detach_shmem, 0);
863 : : {
864 : 1 : num_blocks = apw_dump_now(false, true);
865 : : }
866 [ - + ]: 1 : PG_END_ENSURE_ERROR_CLEANUP(apw_detach_shmem, 0);
867 : :
868 : 1 : PG_RETURN_INT64((int64) num_blocks);
869 : : }
870 : :
871 : : static void
872 : 2 : apw_init_state(void *ptr, void *arg)
873 : : {
874 : 2 : AutoPrewarmSharedState *state = (AutoPrewarmSharedState *) ptr;
875 : :
876 : 2 : LWLockInitialize(&state->lock, LWLockNewTrancheId("autoprewarm"));
877 : 2 : state->bgworker_pid = InvalidPid;
878 : 2 : state->pid_using_dumpfile = InvalidPid;
879 : 2 : }
880 : :
881 : : /*
882 : : * Allocate and initialize autoprewarm related shared memory, if not already
883 : : * done, and set up backend-local pointer to that state. Returns true if an
884 : : * existing shared memory segment was found.
885 : : */
886 : : static bool
887 : 4 : apw_init_shmem(void)
888 : : {
889 : : bool found;
890 : :
891 : 4 : apw_state = GetNamedDSMSegment("autoprewarm",
892 : : sizeof(AutoPrewarmSharedState),
893 : : apw_init_state,
894 : : &found, NULL);
895 : :
896 : 4 : return found;
897 : : }
898 : :
899 : : /*
900 : : * Clear our PID from autoprewarm shared state.
901 : : */
902 : : static void
903 : 2 : apw_detach_shmem(int code, Datum arg)
904 : : {
905 : 2 : LWLockAcquire(&apw_state->lock, LW_EXCLUSIVE);
906 [ - + ]: 2 : if (apw_state->pid_using_dumpfile == MyProcPid)
907 : 0 : apw_state->pid_using_dumpfile = InvalidPid;
908 [ + - ]: 2 : if (apw_state->bgworker_pid == MyProcPid)
909 : 2 : apw_state->bgworker_pid = InvalidPid;
910 : 2 : LWLockRelease(&apw_state->lock);
911 : 2 : }
912 : :
913 : : /*
914 : : * Start autoprewarm leader worker process.
915 : : */
916 : : static void
917 : 2 : apw_start_leader_worker(void)
918 : : {
919 : 2 : BackgroundWorker worker = {0};
920 : : BackgroundWorkerHandle *handle;
921 : : BgwHandleStatus status;
922 : : pid_t pid;
923 : :
924 : 2 : worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
925 : 2 : worker.bgw_start_time = BgWorkerStart_ConsistentState;
926 : 2 : strcpy(worker.bgw_library_name, "pg_prewarm");
927 : 2 : strcpy(worker.bgw_function_name, "autoprewarm_main");
928 : 2 : strcpy(worker.bgw_name, "autoprewarm leader");
929 : 2 : strcpy(worker.bgw_type, "autoprewarm leader");
930 : :
931 [ + - ]: 2 : if (process_shared_preload_libraries_in_progress)
932 : : {
933 : 2 : RegisterBackgroundWorker(&worker);
934 : 2 : return;
935 : : }
936 : :
937 : : /* must set notify PID to wait for startup */
938 : 0 : worker.bgw_notify_pid = MyProcPid;
939 : :
940 [ # # ]: 0 : if (!RegisterDynamicBackgroundWorker(&worker, &handle))
941 [ # # ]: 0 : ereport(ERROR,
942 : : (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
943 : : errmsg("could not register background process"),
944 : : errhint("You may need to increase \"max_worker_processes\".")));
945 : :
946 : 0 : status = WaitForBackgroundWorkerStartup(handle, &pid);
947 [ # # ]: 0 : if (status != BGWH_STARTED)
948 [ # # ]: 0 : ereport(ERROR,
949 : : (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
950 : : errmsg("could not start background process"),
951 : : errhint("More details may be available in the server log.")));
952 : : }
953 : :
954 : : /*
955 : : * Start autoprewarm per-database worker process.
956 : : */
957 : : static void
958 : 1 : apw_start_database_worker(void)
959 : : {
960 : 1 : BackgroundWorker worker = {0};
961 : : BackgroundWorkerHandle *handle;
962 : :
963 : 1 : worker.bgw_flags =
964 : : BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
965 : 1 : worker.bgw_start_time = BgWorkerStart_ConsistentState;
966 : 1 : worker.bgw_restart_time = BGW_NEVER_RESTART;
967 : 1 : strcpy(worker.bgw_library_name, "pg_prewarm");
968 : 1 : strcpy(worker.bgw_function_name, "autoprewarm_database_main");
969 : 1 : strcpy(worker.bgw_name, "autoprewarm worker");
970 : 1 : strcpy(worker.bgw_type, "autoprewarm worker");
971 : :
972 : : /* must set notify PID to wait for shutdown */
973 : 1 : worker.bgw_notify_pid = MyProcPid;
974 : :
975 [ - + ]: 1 : if (!RegisterDynamicBackgroundWorker(&worker, &handle))
976 [ # # ]: 0 : ereport(ERROR,
977 : : (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
978 : : errmsg("registering dynamic bgworker autoprewarm failed"),
979 : : errhint("Consider increasing the configuration parameter \"%s\".", "max_worker_processes")));
980 : :
981 : : /*
982 : : * Ignore return value; if it fails, postmaster has died, but we have
983 : : * checks for that elsewhere.
984 : : */
985 : 1 : WaitForBackgroundWorkerShutdown(handle);
986 : 1 : }
987 : :
988 : : /* Compare member elements to check whether they are not equal. */
989 : : #define cmp_member_elem(fld) \
990 : : do { \
991 : : if (a->fld < b->fld) \
992 : : return -1; \
993 : : else if (a->fld > b->fld) \
994 : : return 1; \
995 : : } while(0)
996 : :
997 : : /*
998 : : * apw_compare_blockinfo
999 : : *
1000 : : * We depend on all records for a particular database being consecutive
1001 : : * in the dump file; each per-database worker will preload blocks until
1002 : : * it sees a block for some other database. Sorting by tablespace,
1003 : : * filenumber, forknum, and blocknum isn't critical for correctness, but
1004 : : * helps us get a sequential I/O pattern.
1005 : : */
1006 : : static int
1007 : 2029 : apw_compare_blockinfo(const void *p, const void *q)
1008 : : {
1009 : 2029 : const BlockInfoRecord *a = (const BlockInfoRecord *) p;
1010 : 2029 : const BlockInfoRecord *b = (const BlockInfoRecord *) q;
1011 : :
1012 [ + + + + ]: 2029 : cmp_member_elem(database);
1013 [ - + - + ]: 1964 : cmp_member_elem(tablespace);
1014 [ + + + + ]: 1964 : cmp_member_elem(filenumber);
1015 [ + + + + ]: 719 : cmp_member_elem(forknum);
1016 [ + + + - ]: 609 : cmp_member_elem(blocknum);
1017 : :
1018 : 0 : return 0;
1019 : : }
|
