Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * xloginsert.c
4 : * Functions for constructing WAL records
5 : *
6 : * Constructing a WAL record begins with a call to XLogBeginInsert,
7 : * followed by a number of XLogRegister* calls. The registered data is
8 : * collected in private working memory, and finally assembled into a chain
9 : * of XLogRecData structs by a call to XLogRecordAssemble(). See
10 : * access/transam/README for details.
11 : *
12 : * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
13 : * Portions Copyright (c) 1994, Regents of the University of California
14 : *
15 : * src/backend/access/transam/xloginsert.c
16 : *
17 : *-------------------------------------------------------------------------
18 : */
19 :
20 : #include "postgres.h"
21 :
22 : #include "access/xact.h"
23 : #include "access/xlog.h"
24 : #include "access/xlog_internal.h"
25 : #include "access/xloginsert.h"
26 : #include "catalog/pg_control.h"
27 : #include "common/pg_lzcompress.h"
28 : #include "executor/instrument.h"
29 : #include "miscadmin.h"
30 : #include "pg_trace.h"
31 : #include "replication/origin.h"
32 : #include "storage/bufmgr.h"
33 : #include "storage/proc.h"
34 : #include "utils/memutils.h"
35 :
36 : /* Buffer size required to store a compressed version of backup block image */
37 : #define PGLZ_MAX_BLCKSZ PGLZ_MAX_OUTPUT(BLCKSZ)
38 :
39 : /*
40 : * For each block reference registered with XLogRegisterBuffer, we fill in
41 : * a registered_buffer struct.
42 : */
43 : typedef struct
44 : {
45 : bool in_use; /* is this slot in use? */
46 : uint8 flags; /* REGBUF_* flags */
47 : RelFileNode rnode; /* identifies the relation and block */
48 : ForkNumber forkno;
49 : BlockNumber block;
50 : Page page; /* page content */
51 : uint32 rdata_len; /* total length of data in rdata chain */
52 : XLogRecData *rdata_head; /* head of the chain of data registered with
53 : * this block */
54 : XLogRecData *rdata_tail; /* last entry in the chain, or &rdata_head if
55 : * empty */
56 :
57 : XLogRecData bkp_rdatas[2]; /* temporary rdatas used to hold references to
58 : * backup block data in XLogRecordAssemble() */
59 :
60 : /* buffer to store a compressed version of backup block image */
61 : char compressed_page[PGLZ_MAX_BLCKSZ];
62 : } registered_buffer;
63 :
64 : static registered_buffer *registered_buffers;
65 : static int max_registered_buffers; /* allocated size */
66 : static int max_registered_block_id = 0; /* highest block_id + 1 currently
67 : * registered */
68 :
69 : /*
70 : * A chain of XLogRecDatas to hold the "main data" of a WAL record, registered
71 : * with XLogRegisterData(...).
72 : */
73 : static XLogRecData *mainrdata_head;
74 : static XLogRecData *mainrdata_last = (XLogRecData *) &mainrdata_head;
75 : static uint32 mainrdata_len; /* total # of bytes in chain */
76 :
77 : /* flags for the in-progress insertion */
78 : static uint8 curinsert_flags = 0;
79 :
80 : /*
81 : * These are used to hold the record header while constructing a record.
82 : * 'hdr_scratch' is not a plain variable, but is palloc'd at initialization,
83 : * because we want it to be MAXALIGNed and padding bytes zeroed.
84 : *
85 : * For simplicity, it's allocated large enough to hold the headers for any
86 : * WAL record.
87 : */
88 : static XLogRecData hdr_rdt;
89 : static char *hdr_scratch = NULL;
90 :
91 : #define SizeOfXlogOrigin (sizeof(RepOriginId) + sizeof(char))
92 :
93 : #define HEADER_SCRATCH_SIZE \
94 : (SizeOfXLogRecord + \
95 : MaxSizeOfXLogRecordBlockHeader * (XLR_MAX_BLOCK_ID + 1) + \
96 : SizeOfXLogRecordDataHeaderLong + SizeOfXlogOrigin)
97 :
98 : /*
99 : * An array of XLogRecData structs, to hold registered data.
100 : */
101 : static XLogRecData *rdatas;
102 : static int num_rdatas; /* entries currently used */
103 : static int max_rdatas; /* allocated size */
104 :
105 : static bool begininsert_called = false;
106 :
107 : /* Memory context to hold the registered buffer and data references. */
108 : static MemoryContext xloginsert_cxt;
109 :
110 : static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
111 : XLogRecPtr RedoRecPtr, bool doPageWrites,
112 : XLogRecPtr *fpw_lsn, int *num_fpi);
113 : static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
114 : uint16 hole_length, char *dest, uint16 *dlen);
115 :
116 : /*
117 : * Begin constructing a WAL record. This must be called before the
118 : * XLogRegister* functions and XLogInsert().
119 : */
120 : void
121 33665182 : XLogBeginInsert(void)
122 : {
123 : Assert(max_registered_block_id == 0);
124 : Assert(mainrdata_last == (XLogRecData *) &mainrdata_head);
125 : Assert(mainrdata_len == 0);
126 :
127 : /* cross-check on whether we should be here or not */
128 33665182 : if (!XLogInsertAllowed())
129 0 : elog(ERROR, "cannot make new WAL entries during recovery");
130 :
131 33665182 : if (begininsert_called)
132 0 : elog(ERROR, "XLogBeginInsert was already called");
133 :
134 33665182 : begininsert_called = true;
135 33665182 : }
136 :
137 : /*
138 : * Ensure that there are enough buffer and data slots in the working area,
139 : * for subsequent XLogRegisterBuffer, XLogRegisterData and XLogRegisterBufData
140 : * calls.
141 : *
142 : * There is always space for a small number of buffers and data chunks, enough
143 : * for most record types. This function is for the exceptional cases that need
144 : * more.
145 : */
146 : void
147 8776 : XLogEnsureRecordSpace(int max_block_id, int ndatas)
148 : {
149 : int nbuffers;
150 :
151 : /*
152 : * This must be called before entering a critical section, because
153 : * allocating memory inside a critical section can fail. repalloc() will
154 : * check the same, but better to check it here too so that we fail
155 : * consistently even if the arrays happen to be large enough already.
156 : */
157 : Assert(CritSectionCount == 0);
158 :
159 : /* the minimum values can't be decreased */
160 8776 : if (max_block_id < XLR_NORMAL_MAX_BLOCK_ID)
161 2628 : max_block_id = XLR_NORMAL_MAX_BLOCK_ID;
162 8776 : if (ndatas < XLR_NORMAL_RDATAS)
163 8760 : ndatas = XLR_NORMAL_RDATAS;
164 :
165 8776 : if (max_block_id > XLR_MAX_BLOCK_ID)
166 0 : elog(ERROR, "maximum number of WAL record block references exceeded");
167 8776 : nbuffers = max_block_id + 1;
168 :
169 8776 : if (nbuffers > max_registered_buffers)
170 : {
171 340 : registered_buffers = (registered_buffer *)
172 340 : repalloc(registered_buffers, sizeof(registered_buffer) * nbuffers);
173 :
174 : /*
175 : * At least the padding bytes in the structs must be zeroed, because
176 : * they are included in WAL data, but initialize it all for tidiness.
177 : */
178 340 : MemSet(®istered_buffers[max_registered_buffers], 0,
179 : (nbuffers - max_registered_buffers) * sizeof(registered_buffer));
180 340 : max_registered_buffers = nbuffers;
181 : }
182 :
183 8776 : if (ndatas > max_rdatas)
184 : {
185 16 : rdatas = (XLogRecData *) repalloc(rdatas, sizeof(XLogRecData) * ndatas);
186 16 : max_rdatas = ndatas;
187 : }
188 8776 : }
189 :
190 : /*
191 : * Reset WAL record construction buffers.
192 : */
193 : void
194 33688470 : XLogResetInsertion(void)
195 : {
196 : int i;
197 :
198 67705802 : for (i = 0; i < max_registered_block_id; i++)
199 34017332 : registered_buffers[i].in_use = false;
200 :
201 33688470 : num_rdatas = 0;
202 33688470 : max_registered_block_id = 0;
203 33688470 : mainrdata_len = 0;
204 33688470 : mainrdata_last = (XLogRecData *) &mainrdata_head;
205 33688470 : curinsert_flags = 0;
206 33688470 : begininsert_called = false;
207 33688470 : }
208 :
209 : /*
210 : * Register a reference to a buffer with the WAL record being constructed.
211 : * This must be called for every page that the WAL-logged operation modifies.
212 : */
213 : void
214 33849848 : XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags)
215 : {
216 : registered_buffer *regbuf;
217 :
218 : /* NO_IMAGE doesn't make sense with FORCE_IMAGE */
219 : Assert(!((flags & REGBUF_FORCE_IMAGE) && (flags & (REGBUF_NO_IMAGE))));
220 : Assert(begininsert_called);
221 :
222 33849848 : if (block_id >= max_registered_block_id)
223 : {
224 33248342 : if (block_id >= max_registered_buffers)
225 0 : elog(ERROR, "too many registered buffers");
226 33248342 : max_registered_block_id = block_id + 1;
227 : }
228 :
229 33849848 : regbuf = ®istered_buffers[block_id];
230 :
231 33849848 : BufferGetTag(buffer, ®buf->rnode, ®buf->forkno, ®buf->block);
232 33849848 : regbuf->page = BufferGetPage(buffer);
233 33849848 : regbuf->flags = flags;
234 33849848 : regbuf->rdata_tail = (XLogRecData *) ®buf->rdata_head;
235 33849848 : regbuf->rdata_len = 0;
236 :
237 : /*
238 : * Check that this page hasn't already been registered with some other
239 : * block_id.
240 : */
241 : #ifdef USE_ASSERT_CHECKING
242 : {
243 : int i;
244 :
245 : for (i = 0; i < max_registered_block_id; i++)
246 : {
247 : registered_buffer *regbuf_old = ®istered_buffers[i];
248 :
249 : if (i == block_id || !regbuf_old->in_use)
250 : continue;
251 :
252 : Assert(!RelFileNodeEquals(regbuf_old->rnode, regbuf->rnode) ||
253 : regbuf_old->forkno != regbuf->forkno ||
254 : regbuf_old->block != regbuf->block);
255 : }
256 : }
257 : #endif
258 :
259 33849848 : regbuf->in_use = true;
260 33849848 : }
261 :
262 : /*
263 : * Like XLogRegisterBuffer, but for registering a block that's not in the
264 : * shared buffer pool (i.e. when you don't have a Buffer for it).
265 : */
266 : void
267 147974 : XLogRegisterBlock(uint8 block_id, RelFileNode *rnode, ForkNumber forknum,
268 : BlockNumber blknum, Page page, uint8 flags)
269 : {
270 : registered_buffer *regbuf;
271 :
272 : /* This is currently only used to WAL-log a full-page image of a page */
273 : Assert(flags & REGBUF_FORCE_IMAGE);
274 : Assert(begininsert_called);
275 :
276 147974 : if (block_id >= max_registered_block_id)
277 147974 : max_registered_block_id = block_id + 1;
278 :
279 147974 : if (block_id >= max_registered_buffers)
280 0 : elog(ERROR, "too many registered buffers");
281 :
282 147974 : regbuf = ®istered_buffers[block_id];
283 :
284 147974 : regbuf->rnode = *rnode;
285 147974 : regbuf->forkno = forknum;
286 147974 : regbuf->block = blknum;
287 147974 : regbuf->page = page;
288 147974 : regbuf->flags = flags;
289 147974 : regbuf->rdata_tail = (XLogRecData *) ®buf->rdata_head;
290 147974 : regbuf->rdata_len = 0;
291 :
292 : /*
293 : * Check that this page hasn't already been registered with some other
294 : * block_id.
295 : */
296 : #ifdef USE_ASSERT_CHECKING
297 : {
298 : int i;
299 :
300 : for (i = 0; i < max_registered_block_id; i++)
301 : {
302 : registered_buffer *regbuf_old = ®istered_buffers[i];
303 :
304 : if (i == block_id || !regbuf_old->in_use)
305 : continue;
306 :
307 : Assert(!RelFileNodeEquals(regbuf_old->rnode, regbuf->rnode) ||
308 : regbuf_old->forkno != regbuf->forkno ||
309 : regbuf_old->block != regbuf->block);
310 : }
311 : }
312 : #endif
313 :
314 147974 : regbuf->in_use = true;
315 147974 : }
316 :
317 : /*
318 : * Add data to the WAL record that's being constructed.
319 : *
320 : * The data is appended to the "main chunk", available at replay with
321 : * XLogRecGetData().
322 : */
323 : void
324 34840262 : XLogRegisterData(char *data, int len)
325 : {
326 : XLogRecData *rdata;
327 :
328 : Assert(begininsert_called);
329 :
330 34840262 : if (num_rdatas >= max_rdatas)
331 0 : elog(ERROR, "too much WAL data");
332 34840262 : rdata = &rdatas[num_rdatas++];
333 :
334 34840262 : rdata->data = data;
335 34840262 : rdata->len = len;
336 :
337 : /*
338 : * we use the mainrdata_last pointer to track the end of the chain, so no
339 : * need to clear 'next' here.
340 : */
341 :
342 34840262 : mainrdata_last->next = rdata;
343 34840262 : mainrdata_last = rdata;
344 :
345 34840262 : mainrdata_len += len;
346 34840262 : }
347 :
348 : /*
349 : * Add buffer-specific data to the WAL record that's being constructed.
350 : *
351 : * Block_id must reference a block previously registered with
352 : * XLogRegisterBuffer(). If this is called more than once for the same
353 : * block_id, the data is appended.
354 : *
355 : * The maximum amount of data that can be registered per block is 65535
356 : * bytes. That should be plenty; if you need more than BLCKSZ bytes to
357 : * reconstruct the changes to the page, you might as well just log a full
358 : * copy of it. (the "main data" that's not associated with a block is not
359 : * limited)
360 : */
361 : void
362 50023876 : XLogRegisterBufData(uint8 block_id, char *data, int len)
363 : {
364 : registered_buffer *regbuf;
365 : XLogRecData *rdata;
366 :
367 : Assert(begininsert_called);
368 :
369 : /* find the registered buffer struct */
370 50023876 : regbuf = ®istered_buffers[block_id];
371 50023876 : if (!regbuf->in_use)
372 0 : elog(ERROR, "no block with id %d registered with WAL insertion",
373 : block_id);
374 :
375 50023876 : if (num_rdatas >= max_rdatas)
376 0 : elog(ERROR, "too much WAL data");
377 50023876 : rdata = &rdatas[num_rdatas++];
378 :
379 50023876 : rdata->data = data;
380 50023876 : rdata->len = len;
381 :
382 50023876 : regbuf->rdata_tail->next = rdata;
383 50023876 : regbuf->rdata_tail = rdata;
384 50023876 : regbuf->rdata_len += len;
385 50023876 : }
386 :
387 : /*
388 : * Set insert status flags for the upcoming WAL record.
389 : *
390 : * The flags that can be used here are:
391 : * - XLOG_INCLUDE_ORIGIN, to determine if the replication origin should be
392 : * included in the record.
393 : * - XLOG_MARK_UNIMPORTANT, to signal that the record is not important for
394 : * durability, which allows to avoid triggering WAL archiving and other
395 : * background activity.
396 : */
397 : void
398 20471202 : XLogSetRecordFlags(uint8 flags)
399 : {
400 : Assert(begininsert_called);
401 20471202 : curinsert_flags = flags;
402 20471202 : }
403 :
404 : /*
405 : * Insert an XLOG record having the specified RMID and info bytes, with the
406 : * body of the record being the data and buffer references registered earlier
407 : * with XLogRegister* calls.
408 : *
409 : * Returns XLOG pointer to end of record (beginning of next record).
410 : * This can be used as LSN for data pages affected by the logged action.
411 : * (LSN is the XLOG point up to which the XLOG must be flushed to disk
412 : * before the data page can be written out. This implements the basic
413 : * WAL rule "write the log before the data".)
414 : */
415 : XLogRecPtr
416 33665182 : XLogInsert(RmgrId rmid, uint8 info)
417 : {
418 : XLogRecPtr EndPos;
419 :
420 : /* XLogBeginInsert() must have been called. */
421 33665182 : if (!begininsert_called)
422 0 : elog(ERROR, "XLogBeginInsert was not called");
423 :
424 : /*
425 : * The caller can set rmgr bits, XLR_SPECIAL_REL_UPDATE and
426 : * XLR_CHECK_CONSISTENCY; the rest are reserved for use by me.
427 : */
428 33665182 : if ((info & ~(XLR_RMGR_INFO_MASK |
429 : XLR_SPECIAL_REL_UPDATE |
430 : XLR_CHECK_CONSISTENCY)) != 0)
431 0 : elog(PANIC, "invalid xlog info mask %02X", info);
432 :
433 : TRACE_POSTGRESQL_WAL_INSERT(rmid, info);
434 :
435 : /*
436 : * In bootstrap mode, we don't actually log anything but XLOG resources;
437 : * return a phony record pointer.
438 : */
439 33665182 : if (IsBootstrapProcessingMode() && rmid != RM_XLOG_ID)
440 : {
441 4283470 : XLogResetInsertion();
442 4283470 : EndPos = SizeOfXLogLongPHD; /* start of 1st chkpt record */
443 4283470 : return EndPos;
444 : }
445 :
446 : do
447 : {
448 : XLogRecPtr RedoRecPtr;
449 : bool doPageWrites;
450 : XLogRecPtr fpw_lsn;
451 : XLogRecData *rdt;
452 29381736 : int num_fpi = 0;
453 :
454 : /*
455 : * Get values needed to decide whether to do full-page writes. Since
456 : * we don't yet have an insertion lock, these could change under us,
457 : * but XLogInsertRecord will recheck them once it has a lock.
458 : */
459 29381736 : GetFullPageWriteInfo(&RedoRecPtr, &doPageWrites);
460 :
461 29381736 : rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites,
462 : &fpw_lsn, &num_fpi);
463 :
464 29381736 : EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi);
465 29381736 : } while (EndPos == InvalidXLogRecPtr);
466 :
467 29381712 : XLogResetInsertion();
468 :
469 29381712 : return EndPos;
470 : }
471 :
472 : /*
473 : * Assemble a WAL record from the registered data and buffers into an
474 : * XLogRecData chain, ready for insertion with XLogInsertRecord().
475 : *
476 : * The record header fields are filled in, except for the xl_prev field. The
477 : * calculated CRC does not include the record header yet.
478 : *
479 : * If there are any registered buffers, and a full-page image was not taken
480 : * of all of them, *fpw_lsn is set to the lowest LSN among such pages. This
481 : * signals that the assembled record is only good for insertion on the
482 : * assumption that the RedoRecPtr and doPageWrites values were up-to-date.
483 : */
484 : static XLogRecData *
485 29381736 : XLogRecordAssemble(RmgrId rmid, uint8 info,
486 : XLogRecPtr RedoRecPtr, bool doPageWrites,
487 : XLogRecPtr *fpw_lsn, int *num_fpi)
488 : {
489 : XLogRecData *rdt;
490 29381736 : uint32 total_len = 0;
491 : int block_id;
492 : pg_crc32c rdata_crc;
493 29381736 : registered_buffer *prev_regbuf = NULL;
494 : XLogRecData *rdt_datas_last;
495 : XLogRecord *rechdr;
496 29381736 : char *scratch = hdr_scratch;
497 :
498 : /*
499 : * Note: this function can be called multiple times for the same record.
500 : * All the modifications we do to the rdata chains below must handle that.
501 : */
502 :
503 : /* The record begins with the fixed-size header */
504 29381736 : rechdr = (XLogRecord *) scratch;
505 29381736 : scratch += SizeOfXLogRecord;
506 :
507 29381736 : hdr_rdt.next = NULL;
508 29381736 : rdt_datas_last = &hdr_rdt;
509 29381736 : hdr_rdt.data = hdr_scratch;
510 :
511 : /*
512 : * Enforce consistency checks for this record if user is looking for it.
513 : * Do this before at the beginning of this routine to give the possibility
514 : * for callers of XLogInsert() to pass XLR_CHECK_CONSISTENCY directly for
515 : * a record.
516 : */
517 29381736 : if (wal_consistency_checking[rmid])
518 0 : info |= XLR_CHECK_CONSISTENCY;
519 :
520 : /*
521 : * Make an rdata chain containing all the data portions of all block
522 : * references. This includes the data for full-page images. Also append
523 : * the headers for the block references in the scratch buffer.
524 : */
525 29381736 : *fpw_lsn = InvalidXLogRecPtr;
526 59269920 : for (block_id = 0; block_id < max_registered_block_id; block_id++)
527 : {
528 29888184 : registered_buffer *regbuf = ®istered_buffers[block_id];
529 : bool needs_backup;
530 : bool needs_data;
531 : XLogRecordBlockHeader bkpb;
532 : XLogRecordBlockImageHeader bimg;
533 29888184 : XLogRecordBlockCompressHeader cbimg = {0};
534 : bool samerel;
535 29888184 : bool is_compressed = false;
536 : bool include_image;
537 :
538 29888184 : if (!regbuf->in_use)
539 19510 : continue;
540 :
541 : /* Determine if this block needs to be backed up */
542 29868674 : if (regbuf->flags & REGBUF_FORCE_IMAGE)
543 187320 : needs_backup = true;
544 29681354 : else if (regbuf->flags & REGBUF_NO_IMAGE)
545 603906 : needs_backup = false;
546 29077448 : else if (!doPageWrites)
547 146250 : needs_backup = false;
548 : else
549 : {
550 : /*
551 : * We assume page LSN is first data on *every* page that can be
552 : * passed to XLogInsert, whether it has the standard page layout
553 : * or not.
554 : */
555 28931198 : XLogRecPtr page_lsn = PageGetLSN(regbuf->page);
556 :
557 28931198 : needs_backup = (page_lsn <= RedoRecPtr);
558 28931198 : if (!needs_backup)
559 : {
560 28849998 : if (*fpw_lsn == InvalidXLogRecPtr || page_lsn < *fpw_lsn)
561 28226810 : *fpw_lsn = page_lsn;
562 : }
563 : }
564 :
565 : /* Determine if the buffer data needs to included */
566 29868674 : if (regbuf->rdata_len == 0)
567 3255472 : needs_data = false;
568 26613202 : else if ((regbuf->flags & REGBUF_KEEP_DATA) != 0)
569 285350 : needs_data = true;
570 : else
571 26327852 : needs_data = !needs_backup;
572 :
573 29868674 : bkpb.id = block_id;
574 29868674 : bkpb.fork_flags = regbuf->forkno;
575 29868674 : bkpb.data_length = 0;
576 :
577 29868674 : if ((regbuf->flags & REGBUF_WILL_INIT) == REGBUF_WILL_INIT)
578 428454 : bkpb.fork_flags |= BKPBLOCK_WILL_INIT;
579 :
580 : /*
581 : * If needs_backup is true or WAL checking is enabled for current
582 : * resource manager, log a full-page write for the current block.
583 : */
584 29868674 : include_image = needs_backup || (info & XLR_CHECK_CONSISTENCY) != 0;
585 :
586 29868674 : if (include_image)
587 : {
588 268520 : Page page = regbuf->page;
589 268520 : uint16 compressed_len = 0;
590 :
591 : /*
592 : * The page needs to be backed up, so calculate its hole length
593 : * and offset.
594 : */
595 268520 : if (regbuf->flags & REGBUF_STANDARD)
596 : {
597 : /* Assume we can omit data between pd_lower and pd_upper */
598 237626 : uint16 lower = ((PageHeader) page)->pd_lower;
599 237626 : uint16 upper = ((PageHeader) page)->pd_upper;
600 :
601 237626 : if (lower >= SizeOfPageHeaderData &&
602 237530 : upper > lower &&
603 : upper <= BLCKSZ)
604 : {
605 237530 : bimg.hole_offset = lower;
606 237530 : cbimg.hole_length = upper - lower;
607 : }
608 : else
609 : {
610 : /* No "hole" to remove */
611 96 : bimg.hole_offset = 0;
612 96 : cbimg.hole_length = 0;
613 : }
614 : }
615 : else
616 : {
617 : /* Not a standard page header, don't try to eliminate "hole" */
618 30894 : bimg.hole_offset = 0;
619 30894 : cbimg.hole_length = 0;
620 : }
621 :
622 : /*
623 : * Try to compress a block image if wal_compression is enabled
624 : */
625 268520 : if (wal_compression)
626 : {
627 : is_compressed =
628 0 : XLogCompressBackupBlock(page, bimg.hole_offset,
629 0 : cbimg.hole_length,
630 0 : regbuf->compressed_page,
631 : &compressed_len);
632 : }
633 :
634 : /*
635 : * Fill in the remaining fields in the XLogRecordBlockHeader
636 : * struct
637 : */
638 268520 : bkpb.fork_flags |= BKPBLOCK_HAS_IMAGE;
639 :
640 : /* Report a full page image constructed for the WAL record */
641 268520 : *num_fpi += 1;
642 :
643 : /*
644 : * Construct XLogRecData entries for the page content.
645 : */
646 268520 : rdt_datas_last->next = ®buf->bkp_rdatas[0];
647 268520 : rdt_datas_last = rdt_datas_last->next;
648 :
649 268520 : bimg.bimg_info = (cbimg.hole_length == 0) ? 0 : BKPIMAGE_HAS_HOLE;
650 :
651 : /*
652 : * If WAL consistency checking is enabled for the resource manager
653 : * of this WAL record, a full-page image is included in the record
654 : * for the block modified. During redo, the full-page is replayed
655 : * only if BKPIMAGE_APPLY is set.
656 : */
657 268520 : if (needs_backup)
658 268520 : bimg.bimg_info |= BKPIMAGE_APPLY;
659 :
660 268520 : if (is_compressed)
661 : {
662 0 : bimg.length = compressed_len;
663 0 : bimg.bimg_info |= BKPIMAGE_IS_COMPRESSED;
664 :
665 0 : rdt_datas_last->data = regbuf->compressed_page;
666 0 : rdt_datas_last->len = compressed_len;
667 : }
668 : else
669 : {
670 268520 : bimg.length = BLCKSZ - cbimg.hole_length;
671 :
672 268520 : if (cbimg.hole_length == 0)
673 : {
674 30990 : rdt_datas_last->data = page;
675 30990 : rdt_datas_last->len = BLCKSZ;
676 : }
677 : else
678 : {
679 : /* must skip the hole */
680 237530 : rdt_datas_last->data = page;
681 237530 : rdt_datas_last->len = bimg.hole_offset;
682 :
683 237530 : rdt_datas_last->next = ®buf->bkp_rdatas[1];
684 237530 : rdt_datas_last = rdt_datas_last->next;
685 :
686 237530 : rdt_datas_last->data =
687 237530 : page + (bimg.hole_offset + cbimg.hole_length);
688 237530 : rdt_datas_last->len =
689 237530 : BLCKSZ - (bimg.hole_offset + cbimg.hole_length);
690 : }
691 : }
692 :
693 268520 : total_len += bimg.length;
694 : }
695 :
696 29868674 : if (needs_data)
697 : {
698 : /*
699 : * Link the caller-supplied rdata chain for this buffer to the
700 : * overall list.
701 : */
702 26559638 : bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
703 26559638 : bkpb.data_length = regbuf->rdata_len;
704 26559638 : total_len += regbuf->rdata_len;
705 :
706 26559638 : rdt_datas_last->next = regbuf->rdata_head;
707 26559638 : rdt_datas_last = regbuf->rdata_tail;
708 : }
709 :
710 29868674 : if (prev_regbuf && RelFileNodeEquals(regbuf->rnode, prev_regbuf->rnode))
711 : {
712 1039908 : samerel = true;
713 1039908 : bkpb.fork_flags |= BKPBLOCK_SAME_REL;
714 : }
715 : else
716 28828766 : samerel = false;
717 29868674 : prev_regbuf = regbuf;
718 :
719 : /* Ok, copy the header to the scratch buffer */
720 29868674 : memcpy(scratch, &bkpb, SizeOfXLogRecordBlockHeader);
721 29868674 : scratch += SizeOfXLogRecordBlockHeader;
722 29868674 : if (include_image)
723 : {
724 268520 : memcpy(scratch, &bimg, SizeOfXLogRecordBlockImageHeader);
725 268520 : scratch += SizeOfXLogRecordBlockImageHeader;
726 268520 : if (cbimg.hole_length != 0 && is_compressed)
727 : {
728 0 : memcpy(scratch, &cbimg,
729 : SizeOfXLogRecordBlockCompressHeader);
730 0 : scratch += SizeOfXLogRecordBlockCompressHeader;
731 : }
732 : }
733 29868674 : if (!samerel)
734 : {
735 28828766 : memcpy(scratch, ®buf->rnode, sizeof(RelFileNode));
736 28828766 : scratch += sizeof(RelFileNode);
737 : }
738 29868674 : memcpy(scratch, ®buf->block, sizeof(BlockNumber));
739 29868674 : scratch += sizeof(BlockNumber);
740 : }
741 :
742 : /* followed by the record's origin, if any */
743 29381736 : if ((curinsert_flags & XLOG_INCLUDE_ORIGIN) &&
744 16227600 : replorigin_session_origin != InvalidRepOriginId)
745 : {
746 2286 : *(scratch++) = (char) XLR_BLOCK_ID_ORIGIN;
747 2286 : memcpy(scratch, &replorigin_session_origin, sizeof(replorigin_session_origin));
748 2286 : scratch += sizeof(replorigin_session_origin);
749 : }
750 :
751 : /* followed by main data, if any */
752 29381736 : if (mainrdata_len > 0)
753 : {
754 29158412 : if (mainrdata_len > 255)
755 : {
756 85260 : *(scratch++) = (char) XLR_BLOCK_ID_DATA_LONG;
757 85260 : memcpy(scratch, &mainrdata_len, sizeof(uint32));
758 85260 : scratch += sizeof(uint32);
759 : }
760 : else
761 : {
762 29073152 : *(scratch++) = (char) XLR_BLOCK_ID_DATA_SHORT;
763 29073152 : *(scratch++) = (uint8) mainrdata_len;
764 : }
765 29158412 : rdt_datas_last->next = mainrdata_head;
766 29158412 : rdt_datas_last = mainrdata_last;
767 29158412 : total_len += mainrdata_len;
768 : }
769 29381736 : rdt_datas_last->next = NULL;
770 :
771 29381736 : hdr_rdt.len = (scratch - hdr_scratch);
772 29381736 : total_len += hdr_rdt.len;
773 :
774 : /*
775 : * Calculate CRC of the data
776 : *
777 : * Note that the record header isn't added into the CRC initially since we
778 : * don't know the prev-link yet. Thus, the CRC will represent the CRC of
779 : * the whole record in the order: rdata, then backup blocks, then record
780 : * header.
781 : */
782 29381736 : INIT_CRC32C(rdata_crc);
783 29381736 : COMP_CRC32C(rdata_crc, hdr_scratch + SizeOfXLogRecord, hdr_rdt.len - SizeOfXLogRecord);
784 102176476 : for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
785 72794740 : COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
786 :
787 : /*
788 : * Fill in the fields in the record header. Prev-link is filled in later,
789 : * once we know where in the WAL the record will be inserted. The CRC does
790 : * not include the record header yet.
791 : */
792 29381736 : rechdr->xl_xid = GetCurrentTransactionIdIfAny();
793 29381736 : rechdr->xl_tot_len = total_len;
794 29381736 : rechdr->xl_info = info;
795 29381736 : rechdr->xl_rmid = rmid;
796 29381736 : rechdr->xl_prev = InvalidXLogRecPtr;
797 29381736 : rechdr->xl_crc = rdata_crc;
798 :
799 29381736 : return &hdr_rdt;
800 : }
801 :
802 : /*
803 : * Create a compressed version of a backup block image.
804 : *
805 : * Returns false if compression fails (i.e., compressed result is actually
806 : * bigger than original). Otherwise, returns true and sets 'dlen' to
807 : * the length of compressed block image.
808 : */
809 : static bool
810 0 : XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
811 : char *dest, uint16 *dlen)
812 : {
813 0 : int32 orig_len = BLCKSZ - hole_length;
814 : int32 len;
815 0 : int32 extra_bytes = 0;
816 : char *source;
817 : PGAlignedBlock tmp;
818 :
819 0 : if (hole_length != 0)
820 : {
821 : /* must skip the hole */
822 0 : source = tmp.data;
823 0 : memcpy(source, page, hole_offset);
824 0 : memcpy(source + hole_offset,
825 0 : page + (hole_offset + hole_length),
826 0 : BLCKSZ - (hole_length + hole_offset));
827 :
828 : /*
829 : * Extra data needs to be stored in WAL record for the compressed
830 : * version of block image if the hole exists.
831 : */
832 0 : extra_bytes = SizeOfXLogRecordBlockCompressHeader;
833 : }
834 : else
835 0 : source = page;
836 :
837 : /*
838 : * We recheck the actual size even if pglz_compress() reports success and
839 : * see if the number of bytes saved by compression is larger than the
840 : * length of extra data needed for the compressed version of block image.
841 : */
842 0 : len = pglz_compress(source, orig_len, dest, PGLZ_strategy_default);
843 0 : if (len >= 0 &&
844 0 : len + extra_bytes < orig_len)
845 : {
846 0 : *dlen = (uint16) len; /* successful compression */
847 0 : return true;
848 : }
849 0 : return false;
850 : }
851 :
852 : /*
853 : * Determine whether the buffer referenced has to be backed up.
854 : *
855 : * Since we don't yet have the insert lock, fullPageWrites and forcePageWrites
856 : * could change later, so the result should be used for optimization purposes
857 : * only.
858 : */
859 : bool
860 218256 : XLogCheckBufferNeedsBackup(Buffer buffer)
861 : {
862 : XLogRecPtr RedoRecPtr;
863 : bool doPageWrites;
864 : Page page;
865 :
866 218256 : GetFullPageWriteInfo(&RedoRecPtr, &doPageWrites);
867 :
868 218256 : page = BufferGetPage(buffer);
869 :
870 218256 : if (doPageWrites && PageGetLSN(page) <= RedoRecPtr)
871 1134 : return true; /* buffer requires backup */
872 :
873 217122 : return false; /* buffer does not need to be backed up */
874 : }
875 :
876 : /*
877 : * Write a backup block if needed when we are setting a hint. Note that
878 : * this may be called for a variety of page types, not just heaps.
879 : *
880 : * Callable while holding just share lock on the buffer content.
881 : *
882 : * We can't use the plain backup block mechanism since that relies on the
883 : * Buffer being exclusively locked. Since some modifications (setting LSN, hint
884 : * bits) are allowed in a sharelocked buffer that can lead to wal checksum
885 : * failures. So instead we copy the page and insert the copied data as normal
886 : * record data.
887 : *
888 : * We only need to do something if page has not yet been full page written in
889 : * this checkpoint round. The LSN of the inserted wal record is returned if we
890 : * had to write, InvalidXLogRecPtr otherwise.
891 : *
892 : * It is possible that multiple concurrent backends could attempt to write WAL
893 : * records. In that case, multiple copies of the same block would be recorded
894 : * in separate WAL records by different backends, though that is still OK from
895 : * a correctness perspective.
896 : */
897 : XLogRecPtr
898 5252 : XLogSaveBufferForHint(Buffer buffer, bool buffer_std)
899 : {
900 5252 : XLogRecPtr recptr = InvalidXLogRecPtr;
901 : XLogRecPtr lsn;
902 : XLogRecPtr RedoRecPtr;
903 :
904 : /*
905 : * Ensure no checkpoint can change our view of RedoRecPtr.
906 : */
907 : Assert(MyProc->delayChkpt);
908 :
909 : /*
910 : * Update RedoRecPtr so that we can make the right decision
911 : */
912 5252 : RedoRecPtr = GetRedoRecPtr();
913 :
914 : /*
915 : * We assume page LSN is first data on *every* page that can be passed to
916 : * XLogInsert, whether it has the standard page layout or not. Since we're
917 : * only holding a share-lock on the page, we must take the buffer header
918 : * lock when we look at the LSN.
919 : */
920 5252 : lsn = BufferGetLSNAtomic(buffer);
921 :
922 5252 : if (lsn <= RedoRecPtr)
923 : {
924 : int flags;
925 : PGAlignedBlock copied_buffer;
926 3484 : char *origdata = (char *) BufferGetBlock(buffer);
927 : RelFileNode rnode;
928 : ForkNumber forkno;
929 : BlockNumber blkno;
930 :
931 : /*
932 : * Copy buffer so we don't have to worry about concurrent hint bit or
933 : * lsn updates. We assume pd_lower/upper cannot be changed without an
934 : * exclusive lock, so the contents bkp are not racy.
935 : */
936 3484 : if (buffer_std)
937 : {
938 : /* Assume we can omit data between pd_lower and pd_upper */
939 2796 : Page page = BufferGetPage(buffer);
940 2796 : uint16 lower = ((PageHeader) page)->pd_lower;
941 2796 : uint16 upper = ((PageHeader) page)->pd_upper;
942 :
943 2796 : memcpy(copied_buffer.data, origdata, lower);
944 2796 : memcpy(copied_buffer.data + upper, origdata + upper, BLCKSZ - upper);
945 : }
946 : else
947 688 : memcpy(copied_buffer.data, origdata, BLCKSZ);
948 :
949 3484 : XLogBeginInsert();
950 :
951 3484 : flags = REGBUF_FORCE_IMAGE;
952 3484 : if (buffer_std)
953 2796 : flags |= REGBUF_STANDARD;
954 :
955 3484 : BufferGetTag(buffer, &rnode, &forkno, &blkno);
956 3484 : XLogRegisterBlock(0, &rnode, forkno, blkno, copied_buffer.data, flags);
957 :
958 3484 : recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI_FOR_HINT);
959 : }
960 :
961 5252 : return recptr;
962 : }
963 :
964 : /*
965 : * Write a WAL record containing a full image of a page. Caller is responsible
966 : * for writing the page to disk after calling this routine.
967 : *
968 : * Note: If you're using this function, you should be building pages in private
969 : * memory and writing them directly to smgr. If you're using buffers, call
970 : * log_newpage_buffer instead.
971 : *
972 : * If the page follows the standard page layout, with a PageHeader and unused
973 : * space between pd_lower and pd_upper, set 'page_std' to true. That allows
974 : * the unused space to be left out from the WAL record, making it smaller.
975 : */
976 : XLogRecPtr
977 144490 : log_newpage(RelFileNode *rnode, ForkNumber forkNum, BlockNumber blkno,
978 : Page page, bool page_std)
979 : {
980 : int flags;
981 : XLogRecPtr recptr;
982 :
983 144490 : flags = REGBUF_FORCE_IMAGE;
984 144490 : if (page_std)
985 144418 : flags |= REGBUF_STANDARD;
986 :
987 144490 : XLogBeginInsert();
988 144490 : XLogRegisterBlock(0, rnode, forkNum, blkno, page, flags);
989 144490 : recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI);
990 :
991 : /*
992 : * The page may be uninitialized. If so, we can't set the LSN because that
993 : * would corrupt the page.
994 : */
995 144490 : if (!PageIsNew(page))
996 : {
997 144490 : PageSetLSN(page, recptr);
998 : }
999 :
1000 144490 : return recptr;
1001 : }
1002 :
1003 : /*
1004 : * Write a WAL record containing a full image of a page.
1005 : *
1006 : * Caller should initialize the buffer and mark it dirty before calling this
1007 : * function. This function will set the page LSN.
1008 : *
1009 : * If the page follows the standard page layout, with a PageHeader and unused
1010 : * space between pd_lower and pd_upper, set 'page_std' to true. That allows
1011 : * the unused space to be left out from the WAL record, making it smaller.
1012 : */
1013 : XLogRecPtr
1014 16 : log_newpage_buffer(Buffer buffer, bool page_std)
1015 : {
1016 16 : Page page = BufferGetPage(buffer);
1017 : RelFileNode rnode;
1018 : ForkNumber forkNum;
1019 : BlockNumber blkno;
1020 :
1021 : /* Shared buffers should be modified in a critical section. */
1022 : Assert(CritSectionCount > 0);
1023 :
1024 16 : BufferGetTag(buffer, &rnode, &forkNum, &blkno);
1025 :
1026 16 : return log_newpage(&rnode, forkNum, blkno, page, page_std);
1027 : }
1028 :
1029 : /*
1030 : * WAL-log a range of blocks in a relation.
1031 : *
1032 : * An image of all pages with block numbers 'startblk' <= X < 'endblk' is
1033 : * written to the WAL. If the range is large, this is done in multiple WAL
1034 : * records.
1035 : *
1036 : * If all page follows the standard page layout, with a PageHeader and unused
1037 : * space between pd_lower and pd_upper, set 'page_std' to true. That allows
1038 : * the unused space to be left out from the WAL records, making them smaller.
1039 : *
1040 : * NOTE: This function acquires exclusive-locks on the pages. Typically, this
1041 : * is used on a newly-built relation, and the caller is holding a
1042 : * AccessExclusiveLock on it, so no other backend can be accessing it at the
1043 : * same time. If that's not the case, you must ensure that this does not
1044 : * cause a deadlock through some other means.
1045 : */
1046 : void
1047 6012 : log_newpage_range(Relation rel, ForkNumber forkNum,
1048 : BlockNumber startblk, BlockNumber endblk,
1049 : bool page_std)
1050 : {
1051 : int flags;
1052 : BlockNumber blkno;
1053 :
1054 6012 : flags = REGBUF_FORCE_IMAGE;
1055 6012 : if (page_std)
1056 612 : flags |= REGBUF_STANDARD;
1057 :
1058 : /*
1059 : * Iterate over all the pages in the range. They are collected into
1060 : * batches of XLR_MAX_BLOCK_ID pages, and a single WAL-record is written
1061 : * for each batch.
1062 : */
1063 6012 : XLogEnsureRecordSpace(XLR_MAX_BLOCK_ID - 1, 0);
1064 :
1065 6012 : blkno = startblk;
1066 12372 : while (blkno < endblk)
1067 : {
1068 : Buffer bufpack[XLR_MAX_BLOCK_ID];
1069 : XLogRecPtr recptr;
1070 : int nbufs;
1071 : int i;
1072 :
1073 6360 : CHECK_FOR_INTERRUPTS();
1074 :
1075 : /* Collect a batch of blocks. */
1076 6360 : nbufs = 0;
1077 43558 : while (nbufs < XLR_MAX_BLOCK_ID && blkno < endblk)
1078 : {
1079 37198 : Buffer buf = ReadBufferExtended(rel, forkNum, blkno,
1080 : RBM_NORMAL, NULL);
1081 :
1082 37198 : LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
1083 :
1084 : /*
1085 : * Completely empty pages are not WAL-logged. Writing a WAL record
1086 : * would change the LSN, and we don't want that. We want the page
1087 : * to stay empty.
1088 : */
1089 37198 : if (!PageIsNew(BufferGetPage(buf)))
1090 37198 : bufpack[nbufs++] = buf;
1091 : else
1092 0 : UnlockReleaseBuffer(buf);
1093 37198 : blkno++;
1094 : }
1095 :
1096 : /* Write WAL record for this batch. */
1097 6360 : XLogBeginInsert();
1098 :
1099 6360 : START_CRIT_SECTION();
1100 43558 : for (i = 0; i < nbufs; i++)
1101 : {
1102 37198 : XLogRegisterBuffer(i, bufpack[i], flags);
1103 37198 : MarkBufferDirty(bufpack[i]);
1104 : }
1105 :
1106 6360 : recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI);
1107 :
1108 43558 : for (i = 0; i < nbufs; i++)
1109 : {
1110 37198 : PageSetLSN(BufferGetPage(bufpack[i]), recptr);
1111 37198 : UnlockReleaseBuffer(bufpack[i]);
1112 : }
1113 6360 : END_CRIT_SECTION();
1114 : }
1115 6012 : }
1116 :
1117 : /*
1118 : * Allocate working buffers needed for WAL record construction.
1119 : */
1120 : void
1121 18180 : InitXLogInsert(void)
1122 : {
1123 : /* Initialize the working areas */
1124 18180 : if (xloginsert_cxt == NULL)
1125 : {
1126 12420 : xloginsert_cxt = AllocSetContextCreate(TopMemoryContext,
1127 : "WAL record construction",
1128 : ALLOCSET_DEFAULT_SIZES);
1129 : }
1130 :
1131 18180 : if (registered_buffers == NULL)
1132 : {
1133 12420 : registered_buffers = (registered_buffer *)
1134 12420 : MemoryContextAllocZero(xloginsert_cxt,
1135 : sizeof(registered_buffer) * (XLR_NORMAL_MAX_BLOCK_ID + 1));
1136 12420 : max_registered_buffers = XLR_NORMAL_MAX_BLOCK_ID + 1;
1137 : }
1138 18180 : if (rdatas == NULL)
1139 : {
1140 12420 : rdatas = MemoryContextAlloc(xloginsert_cxt,
1141 : sizeof(XLogRecData) * XLR_NORMAL_RDATAS);
1142 12420 : max_rdatas = XLR_NORMAL_RDATAS;
1143 : }
1144 :
1145 : /*
1146 : * Allocate a buffer to hold the header information for a WAL record.
1147 : */
1148 18180 : if (hdr_scratch == NULL)
1149 12420 : hdr_scratch = MemoryContextAllocZero(xloginsert_cxt,
1150 : HEADER_SCRATCH_SIZE);
1151 18180 : }
|
