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)
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 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.

{
    const PendingWrite *bufa = (const PendingWrite *) a;
    const PendingWrite *bufb = (const PendingWrite *) b;
 
    /* We should not see duplicated writes for the same block */
    Assert(bufa->blkno != bufb->blkno);
    if (bufa->blkno > bufb->blkno)
        return 1;
    else
        return -1;
}

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.

{
    /* WAL-log and flush any remaining pages */
    smgr_bulk_flush(bulkstate);
 
    /*
     * When we wrote out the pages, we passed skipFsync=true to avoid the
     * overhead of registering all the writes with the checkpointer.  Register
     * the whole relation now.
     *
     * There is one hole in that idea: If a checkpoint occurred while we were
     * writing the pages, it already missed fsyncing the pages we had written
     * before the checkpoint started.  A crash later on would replay the WAL
     * starting from the checkpoint, therefore it wouldn't replay our earlier
     * WAL records.  So if a checkpoint started after the bulk write, fsync
     * the files now.
     */
    if (!SmgrIsTemp(bulkstate->smgr))
    {
        /*
         * Prevent a checkpoint from starting between the GetRedoRecPtr() and
         * smgrregistersync() calls.
         */
        Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
        MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
        if (bulkstate->start_RedoRecPtr != GetRedoRecPtr())
        {
            /*
             * A checkpoint occurred and it didn't know about our writes, so
             * fsync() the relation ourselves.
             */
            MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
            smgrimmedsync(bulkstate->smgr, bulkstate->forknum);
            elog(DEBUG1, "flushed relation because a checkpoint occurred concurrently");
        }
        else
        {
            smgrregistersync(bulkstate->smgr, bulkstate->forknum);
            MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
        }
    }
}

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.

{
    int         npending = bulkstate->npending;
    PendingWrite *pending_writes = bulkstate->pending_writes;
 
    if (npending == 0)
        return;
 
    if (npending > 1)
        qsort(pending_writes, npending, sizeof(PendingWrite), buffer_cmp);
 
    if (bulkstate->use_wal)
    {
        BlockNumber blknos[MAX_PENDING_WRITES];
        Page        pages[MAX_PENDING_WRITES];
        bool        page_std = true;
 
        for (int i = 0; i < npending; i++)
        {
            blknos[i] = pending_writes[i].blkno;
            pages[i] = pending_writes[i].buf->data;
 
            /*
             * If any of the pages use !page_std, we log them all as such.
             * That's a bit wasteful, but in practice, a mix of standard and
             * non-standard page layout is rare.  None of the built-in AMs do
             * that.
             */
            if (!pending_writes[i].page_std)
                page_std = false;
        }
        log_newpages(&bulkstate->smgr->smgr_rlocator.locator, bulkstate->forknum,
                     npending, blknos, pages, page_std);
    }
 
    for (int i = 0; i < npending; i++)
    {
        BlockNumber blkno = pending_writes[i].blkno;
        Page        page = pending_writes[i].buf->data;
 
        PageSetChecksumInplace(page, blkno);
 
        if (blkno >= bulkstate->pages_written)
        {
            /*
             * If we have to write pages nonsequentially, fill in the space
             * with zeroes until we come back and overwrite.  This is not
             * logically necessary on standard Unix filesystems (unwritten
             * space will read as zeroes anyway), but it should help to avoid
             * fragmentation.  The dummy pages aren't WAL-logged though.
             */
            while (blkno > bulkstate->pages_written)
            {
                /* don't set checksum for all-zero page */
                smgrextend(bulkstate->smgr, bulkstate->forknum,
                           bulkstate->pages_written++,
                           &zero_buffer,
                           true);
            }
 
            smgrextend(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
            bulkstate->pages_written = pending_writes[i].blkno + 1;
        }
        else
            smgrwrite(bulkstate->smgr, bulkstate->forknum, blkno, page, true);
        pfree(page);
    }
 
    bulkstate->npending = 0;
}

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.

{
    return MemoryContextAllocAligned(bulkstate->memcxt, BLCKSZ, PG_IO_ALIGN_SIZE, 0);
}

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.

{
    return smgr_bulk_start_smgr(RelationGetSmgr(rel),
                                forknum,
                                RelationNeedsWAL(rel) || forknum == INIT_FORKNUM);
}

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.

{
    BulkWriteState *state;
 
    state = palloc(sizeof(BulkWriteState));
    state->smgr = smgr;
    state->forknum = forknum;
    state->use_wal = use_wal;
 
    state->npending = 0;
    state->pages_written = 0;
 
    state->start_RedoRecPtr = GetRedoRecPtr();
 
    /*
     * Remember the memory context.  We will use it to allocate all the
     * buffers later.
     */
    state->memcxt = CurrentMemoryContext;
 
    return state;
}

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.

{
    PendingWrite *w;
 
    w = &bulkstate->pending_writes[bulkstate->npending++];
    w->buf = buf;
    w->blkno = blocknum;
    w->page_std = page_std;
 
    if (bulkstate->npending == MAX_PENDING_WRITES)
        smgr_bulk_flush(bulkstate);
}

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().