PostgreSQL Source Code  git master
bulk_write.c File Reference
#include "postgres.h"
#include "access/xloginsert.h"
#include "access/xlogrecord.h"
#include "storage/bufmgr.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)
 
BulkWriteStatesmgr_bulk_start_rel (Relation rel, ForkNumber forknum)
 
BulkWriteStatesmgr_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 46 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 174 of file bulk_write.c.

175 {
176  const PendingWrite *bufa = (const PendingWrite *) a;
177  const PendingWrite *bufb = (const PendingWrite *) b;
178 
179  /* We should not see duplicated writes for the same block */
180  Assert(bufa->blkno != bufb->blkno);
181  if (bufa->blkno > bufb->blkno)
182  return 1;
183  else
184  return -1;
185 }
#define Assert(condition)
Definition: c.h:858
int b
Definition: isn.c:70
int a
Definition: isn.c:69
BlockNumber blkno
Definition: bulk_write.c:53

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

Referenced by smgr_bulk_flush().

◆ smgr_bulk_finish()

void smgr_bulk_finish ( BulkWriteState bulkstate)

Definition at line 129 of file bulk_write.c.

130 {
131  /* WAL-log and flush any remaining pages */
132  smgr_bulk_flush(bulkstate);
133 
134  /*
135  * When we wrote out the pages, we passed skipFsync=true to avoid the
136  * overhead of registering all the writes with the checkpointer. Register
137  * the whole relation now.
138  *
139  * There is one hole in that idea: If a checkpoint occurred while we were
140  * writing the pages, it already missed fsyncing the pages we had written
141  * before the checkpoint started. A crash later on would replay the WAL
142  * starting from the checkpoint, therefore it wouldn't replay our earlier
143  * WAL records. So if a checkpoint started after the bulk write, fsync
144  * the files now.
145  */
146  if (!SmgrIsTemp(bulkstate->smgr))
147  {
148  /*
149  * Prevent a checkpoint from starting between the GetRedoRecPtr() and
150  * smgrregistersync() calls.
151  */
154 
155  if (bulkstate->start_RedoRecPtr != GetRedoRecPtr())
156  {
157  /*
158  * A checkpoint occurred and it didn't know about our writes, so
159  * fsync() the relation ourselves.
160  */
162  smgrimmedsync(bulkstate->smgr, bulkstate->forknum);
163  elog(DEBUG1, "flushed relation because a checkpoint occurred concurrently");
164  }
165  else
166  {
167  smgrregistersync(bulkstate->smgr, bulkstate->forknum);
169  }
170  }
171 }
static void smgr_bulk_flush(BulkWriteState *bulkstate)
Definition: bulk_write.c:191
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:224
#define DELAY_CHKPT_START
Definition: proc.h:114
void smgrimmedsync(SMgrRelation reln, ForkNumber forknum)
Definition: smgr.c:789
void smgrregistersync(SMgrRelation reln, ForkNumber forknum)
Definition: smgr.c:757
#define SmgrIsTemp(smgr)
Definition: smgr.h:73
PGPROC * MyProc
Definition: proc.c:66
SMgrRelation smgr
Definition: bulk_write.c:63
XLogRecPtr start_RedoRecPtr
Definition: bulk_write.c:75
ForkNumber forknum
Definition: bulk_write.c:64
int delayChkptFlags
Definition: proc.h:235
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6393

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

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 191 of file bulk_write.c.

192 {
193  int npending = bulkstate->npending;
194  PendingWrite *pending_writes = bulkstate->pending_writes;
195 
196  if (npending == 0)
197  return;
198 
199  if (npending > 1)
200  qsort(pending_writes, npending, sizeof(PendingWrite), buffer_cmp);
201 
202  if (bulkstate->use_wal)
203  {
205  Page pages[MAX_PENDING_WRITES];
206  bool page_std = true;
207 
208  for (int i = 0; i < npending; i++)
209  {
210  blknos[i] = pending_writes[i].blkno;
211  pages[i] = pending_writes[i].buf->data;
212 
213  /*
214  * If any of the pages use !page_std, we log them all as such.
215  * That's a bit wasteful, but in practice, a mix of standard and
216  * non-standard page layout is rare. None of the built-in AMs do
217  * that.
218  */
219  if (!pending_writes[i].page_std)
220  page_std = false;
221  }
222  log_newpages(&bulkstate->smgr->smgr_rlocator.locator, bulkstate->forknum,
223  npending, blknos, pages, page_std);
224  }
225 
226  for (int i = 0; i < npending; i++)
227  {
228  BlockNumber blkno = pending_writes[i].blkno;
229  Page page = pending_writes[i].buf->data;
230 
231  PageSetChecksumInplace(page, blkno);
232 
233  if (blkno >= bulkstate->pages_written)
234  {
235  /*
236  * If we have to write pages nonsequentially, fill in the space
237  * with zeroes until we come back and overwrite. This is not
238  * logically necessary on standard Unix filesystems (unwritten
239  * space will read as zeroes anyway), but it should help to avoid
240  * fragmentation. The dummy pages aren't WAL-logged though.
241  */
242  while (blkno > bulkstate->pages_written)
243  {
244  /* don't set checksum for all-zero page */
245  smgrextend(bulkstate->smgr, bulkstate->forknum,
246  bulkstate->pages_written++,
247  &zero_buffer,
248  true);
249  }
250 
251  smgrextend(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
252  bulkstate->pages_written = pending_writes[i].blkno + 1;
253  }
254  else
255  smgrwrite(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
256  pfree(page);
257  }
258 
259  bulkstate->npending = 0;
260 }
uint32 BlockNumber
Definition: block.h:31
void PageSetChecksumInplace(Page page, BlockNumber blkno)
Definition: bufpage.c:1542
Pointer Page
Definition: bufpage.h:78
static const PGIOAlignedBlock zero_buffer
Definition: bulk_write.c:48
#define MAX_PENDING_WRITES
Definition: bulk_write.c:46
static int buffer_cmp(const void *a, const void *b)
Definition: bulk_write.c:174
int i
Definition: isn.c:73
void pfree(void *pointer)
Definition: mcxt.c:1521
#define qsort(a, b, c, d)
Definition: port.h:453
void smgrextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const void *buffer, bool skipFsync)
Definition: smgr.c:535
static void smgrwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const void *buffer, bool skipFsync)
Definition: smgr.h:121
BlockNumber pages_written
Definition: bulk_write.c:72
PendingWrite pending_writes[MAX_PENDING_WRITES]
Definition: bulk_write.c:69
BulkWriteBuffer buf
Definition: bulk_write.c:52
RelFileLocator locator
RelFileLocatorBackend smgr_rlocator
Definition: smgr.h:37
char data[BLCKSZ]
Definition: c.h:1137
void log_newpages(RelFileLocator *rlocator, ForkNumber forknum, int num_pages, BlockNumber *blknos, Page *pages, bool page_std)
Definition: xloginsert.c:1175

References PendingWrite::blkno, PendingWrite::buf, buffer_cmp(), PGIOAlignedBlock::data, BulkWriteState::forknum, i, RelFileLocatorBackend::locator, log_newpages(), MAX_PENDING_WRITES, BulkWriteState::npending, BulkWriteState::pages_written, PageSetChecksumInplace(), BulkWriteState::pending_writes, pfree(), qsort, 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 295 of file bulk_write.c.

296 {
297  return MemoryContextAllocAligned(bulkstate->memcxt, BLCKSZ, PG_IO_ALIGN_SIZE, 0);
298 }
void * MemoryContextAllocAligned(MemoryContext context, Size size, Size alignto, int flags)
Definition: mcxt.c:1409
#define PG_IO_ALIGN_SIZE
MemoryContext memcxt
Definition: bulk_write.c:77

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 86 of file bulk_write.c.

87 {
89  forknum,
90  RelationNeedsWAL(rel) || forknum == INIT_FORKNUM);
91 }
BulkWriteState * smgr_bulk_start_smgr(SMgrRelation smgr, ForkNumber forknum, bool use_wal)
Definition: bulk_write.c:99
static SMgrRelation RelationGetSmgr(Relation rel)
Definition: rel.h:567
#define RelationNeedsWAL(relation)
Definition: rel.h:628
@ INIT_FORKNUM
Definition: relpath.h:53

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 99 of file bulk_write.c.

100 {
102 
103  state = palloc(sizeof(BulkWriteState));
104  state->smgr = smgr;
105  state->forknum = forknum;
106  state->use_wal = use_wal;
107 
108  state->npending = 0;
109  state->pages_written = 0;
110 
111  state->start_RedoRecPtr = GetRedoRecPtr();
112 
113  /*
114  * Remember the memory context. We will use it to allocate all the
115  * buffers later.
116  */
117  state->memcxt = CurrentMemoryContext;
118 
119  return state;
120 }
MemoryContext CurrentMemoryContext
Definition: mcxt.c:143
void * palloc(Size size)
Definition: mcxt.c:1317
Definition: regguts.h:323

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

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 271 of file bulk_write.c.

272 {
273  PendingWrite *w;
274 
275  w = &bulkstate->pending_writes[bulkstate->npending++];
276  w->buf = buf;
277  w->blkno = blocknum;
278  w->page_std = page_std;
279 
280  if (bulkstate->npending == MAX_PENDING_WRITES)
281  smgr_bulk_flush(bulkstate);
282 }
static char * buf
Definition: pg_test_fsync.c:73
bool page_std
Definition: bulk_write.c:54

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 48 of file bulk_write.c.

Referenced by smgr_bulk_flush().