PostgreSQL Source Code git master
Data Structures | Macros | Typedefs | Functions | Variables
bulk_write.c File Reference
#include "postgres.h"
#include "access/xloginsert.h"
#include "access/xlogrecord.h"
#include "storage/bufpage.h"
#include "storage/bulk_write.h"
#include "storage/proc.h"
#include "storage/smgr.h"
#include "utils/rel.h"
Include dependency graph for bulk_write.c:

Go to the source code of this file.

Data Structures

struct   PendingWrite
 
struct   BulkWriteState
 

Macros

#define  MAX_PENDING_WRITES   XLR_MAX_BLOCK_ID
 

Typedefs

typedef struct PendingWrite  PendingWrite
 

Functions

static void  smgr_bulk_flush (BulkWriteState *bulkstate)
 
BulkWriteState *  smgr_bulk_start_rel (Relation rel, ForkNumber forknum)
 
BulkWriteState *  smgr_bulk_start_smgr (SMgrRelation smgr, ForkNumber forknum, bool use_wal)
 
void  smgr_bulk_finish (BulkWriteState *bulkstate)
 
static int  buffer_cmp (const void *a, const void *b)
 
void  smgr_bulk_write (BulkWriteState *bulkstate, BlockNumber blocknum, BulkWriteBuffer buf, bool page_std)
 
BulkWriteBuffer  smgr_bulk_get_buf (BulkWriteState *bulkstate)
 

Variables

static const PGIOAlignedBlock  zero_buffer = {0}
 

Macro Definition Documentation

◆ MAX_PENDING_WRITES

#define MAX_PENDING_WRITES   XLR_MAX_BLOCK_ID

Definition at line 47 of file bulk_write.c.

Typedef Documentation

◆ PendingWrite

typedef struct PendingWrite PendingWrite

Function Documentation

◆ buffer_cmp()

static int buffer_cmp ( const void *  a,
const void *  b 
)
static

Definition at line 225 of file bulk_write.c.

226{
227 const PendingWrite *bufa = (const PendingWrite *) a;
228 const PendingWrite *bufb = (const PendingWrite *) b;
229
230 /* We should not see duplicated writes for the same block */
231 Assert(bufa->blkno != bufb->blkno);
232 if (bufa->blkno > bufb->blkno)
233 return 1;
234 else
235 return -1;
236}
Assert(PointerIsAligned(start, uint64))
b
int b
Definition: isn.c:74
a
int a
Definition: isn.c:73
BlockNumber blkno
Definition: bulk_write.c:54

References a, Assert(), b, and PendingWrite::blkno.

Referenced by smgr_bulk_flush().

◆ smgr_bulk_finish()

void smgr_bulk_finish ( BulkWriteState *  bulkstate )

Definition at line 130 of file bulk_write.c.

131{
132 /* WAL-log and flush any remaining pages */
133 smgr_bulk_flush(bulkstate);
134
135 /*
136 * Fsync the relation, or register it for the next checkpoint, if
137 * necessary.
138 */
139 if (SmgrIsTemp(bulkstate->smgr))
140 {
141 /* Temporary relations don't need to be fsync'd, ever */
142 }
143 else if (!bulkstate->use_wal)
144 {
145 /*----------
146 * This is either an unlogged relation, or a permanent relation but we
147 * skipped WAL-logging because wal_level=minimal:
148 *
149 * A) Unlogged relation
150 *
151 * Unlogged relations will go away on crash, but they need to be
152 * fsync'd on a clean shutdown. It's sufficient to call
153 * smgrregistersync(), that ensures that the checkpointer will
154 * flush it at the shutdown checkpoint. (It will flush it on the
155 * next online checkpoint too, which is not strictly necessary.)
156 *
157 * Note that the init-fork of an unlogged relation is not
158 * considered unlogged for our purposes. It's treated like a
159 * regular permanent relation. The callers will pass use_wal=true
160 * for the init fork.
161 *
162 * B) Permanent relation, WAL-logging skipped because wal_level=minimal
163 *
164 * This is a new relation, and we didn't WAL-log the pages as we
165 * wrote, but they need to be fsync'd before commit.
166 *
167 * We don't need to do that here, however. The fsync() is done at
168 * commit, by smgrDoPendingSyncs() (*).
169 *
170 * (*) smgrDoPendingSyncs() might decide to WAL-log the whole
171 * relation at commit instead of fsyncing it, if the relation was
172 * very small, but it's smgrDoPendingSyncs() responsibility in any
173 * case.
174 *
175 * We cannot distinguish the two here, so conservatively assume it's
176 * an unlogged relation. A permanent relation with wal_level=minimal
177 * would require no actions, see above.
178 */
179 smgrregistersync(bulkstate->smgr, bulkstate->forknum);
180 }
181 else
182 {
183 /*
184 * Permanent relation, WAL-logged normally.
185 *
186 * We already WAL-logged all the pages, so they will be replayed from
187 * WAL on crash. However, when we wrote out the pages, we passed
188 * skipFsync=true to avoid the overhead of registering all the writes
189 * with the checkpointer. Register the whole relation now.
190 *
191 * There is one hole in that idea: If a checkpoint occurred while we
192 * were writing the pages, it already missed fsyncing the pages we had
193 * written before the checkpoint started. A crash later on would
194 * replay the WAL starting from the checkpoint, therefore it wouldn't
195 * replay our earlier WAL records. So if a checkpoint started after
196 * the bulk write, fsync the files now.
197 */
198
199 /*
200 * Prevent a checkpoint from starting between the GetRedoRecPtr() and
201 * smgrregistersync() calls.
202 */
205
206 if (bulkstate->start_RedoRecPtr != GetRedoRecPtr())
207 {
208 /*
209 * A checkpoint occurred and it didn't know about our writes, so
210 * fsync() the relation ourselves.
211 */
212 MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
213 smgrimmedsync(bulkstate->smgr, bulkstate->forknum);
214 elog(DEBUG1, "flushed relation because a checkpoint occurred concurrently");
215 }
216 else
217 {
218 smgrregistersync(bulkstate->smgr, bulkstate->forknum);
219 MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
220 }
221 }
222}
static void smgr_bulk_flush(BulkWriteState *bulkstate)
Definition: bulk_write.c:242
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:226
#define DELAY_CHKPT_START
Definition: proc.h:135
void smgrimmedsync(SMgrRelation reln, ForkNumber forknum)
Definition: smgr.c:965
void smgrregistersync(SMgrRelation reln, ForkNumber forknum)
Definition: smgr.c:931
#define SmgrIsTemp(smgr)
Definition: smgr.h:74
PGPROC * MyProc
Definition: proc.c:66
bool use_wal
Definition: bulk_write.c:66
SMgrRelation smgr
Definition: bulk_write.c:64
XLogRecPtr start_RedoRecPtr
Definition: bulk_write.c:76
ForkNumber forknum
Definition: bulk_write.c:65
int delayChkptFlags
Definition: proc.h:257
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6489

References Assert(), DEBUG1, DELAY_CHKPT_START, PGPROC::delayChkptFlags, elog, BulkWriteState::forknum, GetRedoRecPtr(), MyProc, BulkWriteState::smgr, smgr_bulk_flush(), smgrimmedsync(), SmgrIsTemp, smgrregistersync(), BulkWriteState::start_RedoRecPtr, and BulkWriteState::use_wal.

Referenced by _bt_load(), btbuildempty(), end_heap_rewrite(), gist_indexsortbuild(), RelationCopyStorage(), and spgbuildempty().

◆ smgr_bulk_flush()

static void smgr_bulk_flush ( BulkWriteState *  bulkstate )
static

Definition at line 242 of file bulk_write.c.

243{
244 int npending = bulkstate->npending;
245 PendingWrite *pending_writes = bulkstate->pending_writes;
246
247 if (npending == 0)
248 return;
249
250 if (npending > 1)
251 qsort(pending_writes, npending, sizeof(PendingWrite), buffer_cmp);
252
253 if (bulkstate->use_wal)
254 {
257 bool page_std = true;
258
259 for (int i = 0; i < npending; i++)
260 {
261 blknos[i] = pending_writes[i].blkno;
262 pages[i] = pending_writes[i].buf->data;
263
264 /*
265 * If any of the pages use !page_std, we log them all as such.
266 * That's a bit wasteful, but in practice, a mix of standard and
267 * non-standard page layout is rare. None of the built-in AMs do
268 * that.
269 */
270 if (!pending_writes[i].page_std)
271 page_std = false;
272 }
273 log_newpages(&bulkstate->smgr->smgr_rlocator.locator, bulkstate->forknum,
274 npending, blknos, pages, page_std);
275 }
276
277 for (int i = 0; i < npending; i++)
278 {
279 BlockNumber blkno = pending_writes[i].blkno;
280 Page page = pending_writes[i].buf->data;
281
282 PageSetChecksumInplace(page, blkno);
283
284 if (blkno >= bulkstate->relsize)
285 {
286 /*
287 * If we have to write pages nonsequentially, fill in the space
288 * with zeroes until we come back and overwrite. This is not
289 * logically necessary on standard Unix filesystems (unwritten
290 * space will read as zeroes anyway), but it should help to avoid
291 * fragmentation. The dummy pages aren't WAL-logged though.
292 */
293 while (blkno > bulkstate->relsize)
294 {
295 /* don't set checksum for all-zero page */
296 smgrextend(bulkstate->smgr, bulkstate->forknum,
297 bulkstate->relsize,
299 true);
300 bulkstate->relsize++;
301 }
302
303 smgrextend(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
304 bulkstate->relsize++;
305 }
306 else
307 smgrwrite(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
308 pfree(page);
309 }
310
311 bulkstate->npending = 0;
312}
uint32 BlockNumber
Definition: block.h:31
void PageSetChecksumInplace(Page page, BlockNumber blkno)
Definition: bufpage.c:1541
PageData * Page
Definition: bufpage.h:82
static const PGIOAlignedBlock zero_buffer
Definition: bulk_write.c:49
#define MAX_PENDING_WRITES
Definition: bulk_write.c:47
static int buffer_cmp(const void *a, const void *b)
Definition: bulk_write.c:225
i
int i
Definition: isn.c:77
void pfree(void *pointer)
Definition: mcxt.c:1594
#define qsort(a, b, c, d)
Definition: port.h:479
void smgrextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const void *buffer, bool skipFsync)
Definition: smgr.c:620
static void smgrwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const void *buffer, bool skipFsync)
Definition: smgr.h:131
BlockNumber relsize
Definition: bulk_write.c:73
int npending
Definition: bulk_write.c:69
PendingWrite pending_writes[MAX_PENDING_WRITES]
Definition: bulk_write.c:70
BulkWriteBuffer buf
Definition: bulk_write.c:53
RelFileLocator locator
Definition: relfilelocator.h:75
RelFileLocatorBackend smgr_rlocator
Definition: smgr.h:38
char data[BLCKSZ]
Definition: c.h:1136
void log_newpages(RelFileLocator *rlocator, ForkNumber forknum, int num_pages, BlockNumber *blknos, Page *pages, bool page_std)
Definition: xloginsert.c:1187

References PendingWrite::blkno, PendingWrite::buf, buffer_cmp(), PGIOAlignedBlock::data, BulkWriteState::forknum, i, RelFileLocatorBackend::locator, log_newpages(), MAX_PENDING_WRITES, BulkWriteState::npending, PageSetChecksumInplace(), BulkWriteState::pending_writes, pfree(), qsort, BulkWriteState::relsize, BulkWriteState::smgr, SMgrRelationData::smgr_rlocator, smgrextend(), smgrwrite(), BulkWriteState::use_wal, and zero_buffer.

Referenced by smgr_bulk_finish(), and smgr_bulk_write().

◆ smgr_bulk_get_buf()

BulkWriteBuffer smgr_bulk_get_buf ( BulkWriteState *  bulkstate )

Definition at line 347 of file bulk_write.c.

348{
349 return MemoryContextAllocAligned(bulkstate->memcxt, BLCKSZ, PG_IO_ALIGN_SIZE, 0);
350}
void * MemoryContextAllocAligned(MemoryContext context, Size size, Size alignto, int flags)
Definition: mcxt.c:1460
#define PG_IO_ALIGN_SIZE
MemoryContext memcxt
Definition: bulk_write.c:78

References BulkWriteState::memcxt, MemoryContextAllocAligned(), and PG_IO_ALIGN_SIZE.

Referenced by _bt_blnewpage(), _bt_uppershutdown(), btbuildempty(), gist_indexsortbuild(), gist_indexsortbuild_levelstate_flush(), raw_heap_insert(), RelationCopyStorage(), and spgbuildempty().

◆ smgr_bulk_start_rel()

BulkWriteState * smgr_bulk_start_rel ( Relation  rel,
ForkNumber  forknum 
)

Definition at line 87 of file bulk_write.c.

88{
90 forknum,
91 RelationNeedsWAL(rel) || forknum == INIT_FORKNUM);
92}
BulkWriteState * smgr_bulk_start_smgr(SMgrRelation smgr, ForkNumber forknum, bool use_wal)
Definition: bulk_write.c:100
static SMgrRelation RelationGetSmgr(Relation rel)
Definition: rel.h:576
#define RelationNeedsWAL(relation)
Definition: rel.h:637
@ INIT_FORKNUM
Definition: relpath.h:61

References INIT_FORKNUM, RelationGetSmgr(), RelationNeedsWAL, and smgr_bulk_start_smgr().

Referenced by _bt_load(), begin_heap_rewrite(), btbuildempty(), gist_indexsortbuild(), and spgbuildempty().

◆ smgr_bulk_start_smgr()

BulkWriteState * smgr_bulk_start_smgr ( SMgrRelation  smgr,
ForkNumber  forknum,
bool  use_wal 
)

Definition at line 100 of file bulk_write.c.

101{
103
104 state = palloc(sizeof(BulkWriteState));
105 state->smgr = smgr;
106 state->forknum = forknum;
107 state->use_wal = use_wal;
108
109 state->npending = 0;
110 state->relsize = smgrnblocks(smgr, forknum);
111
112 state->start_RedoRecPtr = GetRedoRecPtr();
113
114 /*
115 * Remember the memory context. We will use it to allocate all the
116 * buffers later.
117 */
118 state->memcxt = CurrentMemoryContext;
119
120 return state;
121}
void * palloc(Size size)
Definition: mcxt.c:1365
MemoryContext CurrentMemoryContext
Definition: mcxt.c:160
BlockNumber smgrnblocks(SMgrRelation reln, ForkNumber forknum)
Definition: smgr.c:819
Definition: regguts.h:323

References CurrentMemoryContext, GetRedoRecPtr(), palloc(), and smgrnblocks().

Referenced by RelationCopyStorage(), and smgr_bulk_start_rel().

◆ smgr_bulk_write()

void smgr_bulk_write ( BulkWriteState *  bulkstate,
BlockNumber  blocknum,
BulkWriteBuffer  buf,
bool  page_std 
)

Definition at line 323 of file bulk_write.c.

324{
325 PendingWrite *w;
326
327 w = &bulkstate->pending_writes[bulkstate->npending++];
328 w->buf = buf;
329 w->blkno = blocknum;
330 w->page_std = page_std;
331
332 if (bulkstate->npending == MAX_PENDING_WRITES)
333 smgr_bulk_flush(bulkstate);
334}
static char * buf
Definition: pg_test_fsync.c:72
bool page_std
Definition: bulk_write.c:55

References PendingWrite::blkno, PendingWrite::buf, buf, MAX_PENDING_WRITES, BulkWriteState::npending, PendingWrite::page_std, BulkWriteState::pending_writes, and smgr_bulk_flush().

Referenced by _bt_blwritepage(), btbuildempty(), end_heap_rewrite(), gist_indexsortbuild(), gist_indexsortbuild_levelstate_flush(), raw_heap_insert(), RelationCopyStorage(), and spgbuildempty().

Variable Documentation

◆ zero_buffer

const PGIOAlignedBlock zero_buffer = {0}
static

Definition at line 49 of file bulk_write.c.

Referenced by smgr_bulk_flush().

AltStyle によって変換されたページ (->オリジナル) /