localbuf.c File Reference

#include "postgres.h"
#include "access/parallel.h"
#include "executor/instrument.h"
#include "pgstat.h"
#include "storage/aio.h"
#include "storage/buf_internals.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "utils/guc_hooks.h"
#include "utils/memdebug.h"
#include "utils/memutils.h"
#include "utils/resowner.h"

Include dependency graph for localbuf.c:

Go to the source code of this file.

Data Structures
struct	LocalBufferLookupEnt

Macros
#define	LocalBufHdrGetBlock(bufHdr)    LocalBufferBlockPointers[-((bufHdr)->buf_id + 2)]

Functions
static void	InitLocalBuffers (void)
static Block	GetLocalBufferStorage (void)
static Buffer	GetLocalVictimBuffer (void)
PrefetchBufferResult	PrefetchLocalBuffer (SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum)
BufferDesc *	LocalBufferAlloc (SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum, bool *foundPtr)
void	FlushLocalBuffer (BufferDesc *bufHdr, SMgrRelation reln)
uint32	GetLocalPinLimit (void)
uint32	GetAdditionalLocalPinLimit (void)
void	LimitAdditionalLocalPins (uint32 *additional_pins)
BlockNumber	ExtendBufferedRelLocal (BufferManagerRelation bmr, ForkNumber fork, uint32 flags, uint32 extend_by, BlockNumber extend_upto, Buffer *buffers, uint32 *extended_by)
void	MarkLocalBufferDirty (Buffer buffer)
bool	StartLocalBufferIO (BufferDesc *bufHdr, bool forInput, bool nowait)
void	TerminateLocalBufferIO (BufferDesc *bufHdr, bool clear_dirty, uint32 set_flag_bits, bool release_aio)
void	InvalidateLocalBuffer (BufferDesc *bufHdr, bool check_unreferenced)
void	DropRelationLocalBuffers (RelFileLocator rlocator, ForkNumber forkNum, BlockNumber firstDelBlock)
void	DropRelationAllLocalBuffers (RelFileLocator rlocator)
bool	PinLocalBuffer (BufferDesc *buf_hdr, bool adjust_usagecount)
void	UnpinLocalBuffer (Buffer buffer)
void	UnpinLocalBufferNoOwner (Buffer buffer)
bool	check_temp_buffers (int *newval, void **extra, GucSource source)
static void	CheckForLocalBufferLeaks (void)
void	AtEOXact_LocalBuffers (bool isCommit)
void	AtProcExit_LocalBuffers (void)

Variables
int	NLocBuffer = 0
BufferDesc *	LocalBufferDescriptors = NULL
Block *	LocalBufferBlockPointers = NULL
int32 *	LocalRefCount = NULL
static int	nextFreeLocalBufId = 0
static HTAB *	LocalBufHash = NULL
static int	NLocalPinnedBuffers = 0

Macro Definition Documentation

LocalBufHdrGetBlock

#define LocalBufHdrGetBlock

(

bufHdr

)

LocalBufferBlockPointers[-((bufHdr)->buf_id + 2)]

Definition at line 41 of file localbuf.c.

Function Documentation

AtEOXact_LocalBuffers()

void AtEOXact_LocalBuffers

(

bool

isCommit

)

Definition at line 993 of file localbuf.c.

{
    CheckForLocalBufferLeaks();
}

References CheckForLocalBufferLeaks().

Referenced by AtEOXact_Buffers().

AtProcExit_LocalBuffers()

void AtProcExit_LocalBuffers

(

void

)

Definition at line 1004 of file localbuf.c.

{
    /*
     * We shouldn't be holding any remaining pins; if we are, and assertions
     * aren't enabled, we'll fail later in DropRelationBuffers while trying to
     * drop the temp rels.
     */
    CheckForLocalBufferLeaks();
}

References CheckForLocalBufferLeaks().

Referenced by AtProcExit_Buffers().

check_temp_buffers()

bool check_temp_buffers	(	int *	newval,
		void **	extra,
		GucSource	source
	)

Definition at line 868 of file localbuf.c.

{
    /*
     * Once local buffers have been initialized, it's too late to change this.
     * However, if this is only a test call, allow it.
     */
    if (source != PGC_S_TEST && NLocBuffer && NLocBuffer != *newval)
    {
        GUC_check_errdetail("\"temp_buffers\" cannot be changed after any temporary tables have been accessed in the session.");
        return false;
    }
    return true;
}

References GUC_check_errdetail, newval, NLocBuffer, PGC_S_TEST, and source.

CheckForLocalBufferLeaks()

static void CheckForLocalBufferLeaks ( void  )

static

Definition at line 960 of file localbuf.c.

{
#ifdef USE_ASSERT_CHECKING
    if (LocalRefCount)
    {
        int         RefCountErrors = 0;
        int         i;
 
        for (i = 0; i < NLocBuffer; i++)
        {
            if (LocalRefCount[i] != 0)
            {
                Buffer      b = -i - 1;
                char       *s;
 
                s = DebugPrintBufferRefcount(b);
                elog(WARNING, "local buffer refcount leak: %s", s);
                pfree(s);
 
                RefCountErrors++;
            }
        }
        Assert(RefCountErrors == 0);
    }
#endif
}

References Assert(), b, DebugPrintBufferRefcount(), elog, i, LocalRefCount, NLocBuffer, pfree(), and WARNING.

Referenced by AtEOXact_LocalBuffers(), and AtProcExit_LocalBuffers().

DropRelationAllLocalBuffers()

void DropRelationAllLocalBuffers

(

RelFileLocator

rlocator

)

Definition at line 693 of file localbuf.c.

{
    int         i;
 
    for (i = 0; i < NLocBuffer; i++)
    {
        BufferDesc *bufHdr = GetLocalBufferDescriptor(i);
        uint32      buf_state;
 
        buf_state = pg_atomic_read_u32(&bufHdr->state);
 
        if ((buf_state & BM_TAG_VALID) &&
            BufTagMatchesRelFileLocator(&bufHdr->tag, &rlocator))
        {
            InvalidateLocalBuffer(bufHdr, true);
        }
    }
}

References BM_TAG_VALID, BufTagMatchesRelFileLocator(), GetLocalBufferDescriptor(), i, InvalidateLocalBuffer(), NLocBuffer, pg_atomic_read_u32(), BufferDesc::state, and BufferDesc::tag.

Referenced by DropRelationsAllBuffers().

DropRelationLocalBuffers()

void DropRelationLocalBuffers	(	RelFileLocator	rlocator,
		ForkNumber	forkNum,
		BlockNumber	firstDelBlock
	)

Definition at line 663 of file localbuf.c.

{
    int         i;
 
    for (i = 0; i < NLocBuffer; i++)
    {
        BufferDesc *bufHdr = GetLocalBufferDescriptor(i);
        uint32      buf_state;
 
        buf_state = pg_atomic_read_u32(&bufHdr->state);
 
        if ((buf_state & BM_TAG_VALID) &&
            BufTagMatchesRelFileLocator(&bufHdr->tag, &rlocator) &&
            BufTagGetForkNum(&bufHdr->tag) == forkNum &&
            bufHdr->tag.blockNum >= firstDelBlock)
        {
            InvalidateLocalBuffer(bufHdr, true);
        }
    }
}

References buftag::blockNum, BM_TAG_VALID, BufTagGetForkNum(), BufTagMatchesRelFileLocator(), GetLocalBufferDescriptor(), i, InvalidateLocalBuffer(), NLocBuffer, pg_atomic_read_u32(), BufferDesc::state, and BufferDesc::tag.

Referenced by DropRelationBuffers().

ExtendBufferedRelLocal()

BlockNumber ExtendBufferedRelLocal	(	BufferManagerRelation	bmr,
		ForkNumber	fork,
		uint32	flags,
		uint32	extend_by,
		BlockNumber	extend_upto,
		Buffer *	buffers,
		uint32 *	extended_by
	)

Definition at line 345 of file localbuf.c.

{
    BlockNumber first_block;
    instr_time  io_start;
 
    /* Initialize local buffers if first request in this session */
    if (LocalBufHash == NULL)
        InitLocalBuffers();
 
    LimitAdditionalLocalPins(&extend_by);
 
    for (uint32 i = 0; i < extend_by; i++)
    {
        BufferDesc *buf_hdr;
        Block       buf_block;
 
        buffers[i] = GetLocalVictimBuffer();
        buf_hdr = GetLocalBufferDescriptor(-buffers[i] - 1);
        buf_block = LocalBufHdrGetBlock(buf_hdr);
 
        /* new buffers are zero-filled */
        MemSet(buf_block, 0, BLCKSZ);
    }
 
    first_block = smgrnblocks(bmr.smgr, fork);
 
    if (extend_upto != InvalidBlockNumber)
    {
        /*
         * In contrast to shared relations, nothing could change the relation
         * size concurrently. Thus we shouldn't end up finding that we don't
         * need to do anything.
         */
        Assert(first_block <= extend_upto);
 
        Assert((uint64) first_block + extend_by <= extend_upto);
    }
 
    /* Fail if relation is already at maximum possible length */
    if ((uint64) first_block + extend_by >= MaxBlockNumber)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("cannot extend relation %s beyond %u blocks",
                        relpath(bmr.smgr->smgr_rlocator, fork).str,
                        MaxBlockNumber)));
 
    for (uint32 i = 0; i < extend_by; i++)
    {
        int         victim_buf_id;
        BufferDesc *victim_buf_hdr;
        BufferTag   tag;
        LocalBufferLookupEnt *hresult;
        bool        found;
 
        victim_buf_id = -buffers[i] - 1;
        victim_buf_hdr = GetLocalBufferDescriptor(victim_buf_id);
 
        /* in case we need to pin an existing buffer below */
        ResourceOwnerEnlarge(CurrentResourceOwner);
 
        InitBufferTag(&tag, &bmr.smgr->smgr_rlocator.locator, fork, first_block + i);
 
        hresult = (LocalBufferLookupEnt *)
            hash_search(LocalBufHash, &tag, HASH_ENTER, &found);
        if (found)
        {
            BufferDesc *existing_hdr;
            uint32      buf_state;
 
            UnpinLocalBuffer(BufferDescriptorGetBuffer(victim_buf_hdr));
 
            existing_hdr = GetLocalBufferDescriptor(hresult->id);
            PinLocalBuffer(existing_hdr, false);
            buffers[i] = BufferDescriptorGetBuffer(existing_hdr);
 
            /*
             * Clear the BM_VALID bit, do StartLocalBufferIO() and proceed.
             */
            buf_state = pg_atomic_read_u32(&existing_hdr->state);
            Assert(buf_state & BM_TAG_VALID);
            Assert(!(buf_state & BM_DIRTY));
            buf_state &= ~BM_VALID;
            pg_atomic_unlocked_write_u32(&existing_hdr->state, buf_state);
 
            /* no need to loop for local buffers */
            StartLocalBufferIO(existing_hdr, true, false);
        }
        else
        {
            uint32      buf_state = pg_atomic_read_u32(&victim_buf_hdr->state);
 
            Assert(!(buf_state & (BM_VALID | BM_TAG_VALID | BM_DIRTY | BM_JUST_DIRTIED)));
 
            victim_buf_hdr->tag = tag;
 
            buf_state |= BM_TAG_VALID | BUF_USAGECOUNT_ONE;
 
            pg_atomic_unlocked_write_u32(&victim_buf_hdr->state, buf_state);
 
            hresult->id = victim_buf_id;
 
            StartLocalBufferIO(victim_buf_hdr, true, false);
        }
    }
 
    io_start = pgstat_prepare_io_time(track_io_timing);
 
    /* actually extend relation */
    smgrzeroextend(bmr.smgr, fork, first_block, extend_by, false);
 
    pgstat_count_io_op_time(IOOBJECT_TEMP_RELATION, IOCONTEXT_NORMAL, IOOP_EXTEND,
                            io_start, 1, extend_by * BLCKSZ);
 
    for (uint32 i = 0; i < extend_by; i++)
    {
        Buffer      buf = buffers[i];
        BufferDesc *buf_hdr;
        uint32      buf_state;
 
        buf_hdr = GetLocalBufferDescriptor(-buf - 1);
 
        buf_state = pg_atomic_read_u32(&buf_hdr->state);
        buf_state |= BM_VALID;
        pg_atomic_unlocked_write_u32(&buf_hdr->state, buf_state);
    }
 
    *extended_by = extend_by;
 
    pgBufferUsage.local_blks_written += extend_by;
 
    return first_block;
}

References Assert(), BM_DIRTY, BM_JUST_DIRTIED, BM_TAG_VALID, BM_VALID, buf, BUF_USAGECOUNT_ONE, BufferDescriptorGetBuffer(), CurrentResourceOwner, ereport, errcode(), errmsg(), ERROR, GetLocalBufferDescriptor(), GetLocalVictimBuffer(), HASH_ENTER, hash_search(), i, LocalBufferLookupEnt::id, InitBufferTag(), InitLocalBuffers(), InvalidBlockNumber, IOCONTEXT_NORMAL, IOOBJECT_TEMP_RELATION, IOOP_EXTEND, LimitAdditionalLocalPins(), BufferUsage::local_blks_written, LocalBufHash, LocalBufHdrGetBlock, RelFileLocatorBackend::locator, MaxBlockNumber, MemSet, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), pgBufferUsage, pgstat_count_io_op_time(), pgstat_prepare_io_time(), PinLocalBuffer(), relpath, ResourceOwnerEnlarge(), BufferManagerRelation::smgr, SMgrRelationData::smgr_rlocator, smgrnblocks(), smgrzeroextend(), StartLocalBufferIO(), BufferDesc::state, BufferDesc::tag, track_io_timing, and UnpinLocalBuffer().

Referenced by ExtendBufferedRelCommon().

FlushLocalBuffer()

void FlushLocalBuffer	(	BufferDesc *	bufHdr,
		SMgrRelation	reln
	)

Definition at line 182 of file localbuf.c.

{
    instr_time  io_start;
    Page        localpage = (char *) LocalBufHdrGetBlock(bufHdr);
 
    Assert(LocalRefCount[-BufferDescriptorGetBuffer(bufHdr) - 1] > 0);
 
    /*
     * Try to start an I/O operation.  There currently are no reasons for
     * StartLocalBufferIO to return false, so we raise an error in that case.
     */
    if (!StartLocalBufferIO(bufHdr, false, false))
        elog(ERROR, "failed to start write IO on local buffer");
 
    /* Find smgr relation for buffer */
    if (reln == NULL)
        reln = smgropen(BufTagGetRelFileLocator(&bufHdr->tag),
                        MyProcNumber);
 
    PageSetChecksumInplace(localpage, bufHdr->tag.blockNum);
 
    io_start = pgstat_prepare_io_time(track_io_timing);
 
    /* And write... */
    smgrwrite(reln,
              BufTagGetForkNum(&bufHdr->tag),
              bufHdr->tag.blockNum,
              localpage,
              false);
 
    /* Temporary table I/O does not use Buffer Access Strategies */
    pgstat_count_io_op_time(IOOBJECT_TEMP_RELATION, IOCONTEXT_NORMAL,
                            IOOP_WRITE, io_start, 1, BLCKSZ);
 
    /* Mark not-dirty */
    TerminateLocalBufferIO(bufHdr, true, 0, false);
 
    pgBufferUsage.local_blks_written++;
}

References Assert(), buftag::blockNum, BufferDescriptorGetBuffer(), BufTagGetForkNum(), BufTagGetRelFileLocator(), elog, ERROR, IOCONTEXT_NORMAL, IOOBJECT_TEMP_RELATION, IOOP_WRITE, BufferUsage::local_blks_written, LocalBufHdrGetBlock, LocalRefCount, MyProcNumber, PageSetChecksumInplace(), pgBufferUsage, pgstat_count_io_op_time(), pgstat_prepare_io_time(), smgropen(), smgrwrite(), StartLocalBufferIO(), BufferDesc::tag, TerminateLocalBufferIO(), and track_io_timing.

Referenced by FlushRelationBuffers(), GetLocalVictimBuffer(), and invalidate_rel_block().

GetAdditionalLocalPinLimit()

uint32 GetAdditionalLocalPinLimit

(

void

)

Definition at line 314 of file localbuf.c.

{
    Assert(NLocalPinnedBuffers <= num_temp_buffers);
    return num_temp_buffers - NLocalPinnedBuffers;
}

References Assert(), NLocalPinnedBuffers, and num_temp_buffers.

Referenced by read_stream_start_pending_read().

GetLocalBufferStorage()

static Block GetLocalBufferStorage ( void  )

static

Definition at line 892 of file localbuf.c.

{
    static char *cur_block = NULL;
    static int  next_buf_in_block = 0;
    static int  num_bufs_in_block = 0;
    static int  total_bufs_allocated = 0;
    static MemoryContext LocalBufferContext = NULL;
 
    char       *this_buf;
 
    Assert(total_bufs_allocated < NLocBuffer);
 
    if (next_buf_in_block >= num_bufs_in_block)
    {
        /* Need to make a new request to memmgr */
        int         num_bufs;
 
        /*
         * We allocate local buffers in a context of their own, so that the
         * space eaten for them is easily recognizable in MemoryContextStats
         * output.  Create the context on first use.
         */
        if (LocalBufferContext == NULL)
            LocalBufferContext =
                AllocSetContextCreate(TopMemoryContext,
                                      "LocalBufferContext",
                                      ALLOCSET_DEFAULT_SIZES);
 
        /* Start with a 16-buffer request; subsequent ones double each time */
        num_bufs = Max(num_bufs_in_block * 2, 16);
        /* But not more than what we need for all remaining local bufs */
        num_bufs = Min(num_bufs, NLocBuffer - total_bufs_allocated);
        /* And don't overflow MaxAllocSize, either */
        num_bufs = Min(num_bufs, MaxAllocSize / BLCKSZ);
 
        /* Buffers should be I/O aligned. */
        cur_block = (char *)
            TYPEALIGN(PG_IO_ALIGN_SIZE,
                      MemoryContextAlloc(LocalBufferContext,
                                         num_bufs * BLCKSZ + PG_IO_ALIGN_SIZE));
        next_buf_in_block = 0;
        num_bufs_in_block = num_bufs;
    }
 
    /* Allocate next buffer in current memory block */
    this_buf = cur_block + next_buf_in_block * BLCKSZ;
    next_buf_in_block++;
    total_bufs_allocated++;
 
    /*
     * Caller's PinLocalBuffer() was too early for Valgrind updates, so do it
     * here.  The block is actually undefined, but we want consistency with
     * the regular case of not needing to allocate memory.  This is
     * specifically needed when method_io_uring.c fills the block, because
     * Valgrind doesn't recognize io_uring reads causing undefined memory to
     * become defined.
     */
    VALGRIND_MAKE_MEM_DEFINED(this_buf, BLCKSZ);
 
    return (Block) this_buf;
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert(), Max, MaxAllocSize, MemoryContextAlloc(), Min, NLocBuffer, PG_IO_ALIGN_SIZE, TopMemoryContext, TYPEALIGN, and VALGRIND_MAKE_MEM_DEFINED.

Referenced by GetLocalVictimBuffer().

GetLocalPinLimit()

uint32 GetLocalPinLimit

(

void

)

Definition at line 306 of file localbuf.c.

{
    /* Every backend has its own temporary buffers, and can pin them all. */
    return num_temp_buffers;
}

References num_temp_buffers.

Referenced by read_stream_begin_impl().

GetLocalVictimBuffer()

static Buffer GetLocalVictimBuffer ( void  )

static

Definition at line 223 of file localbuf.c.

{
    int         victim_bufid;
    int         trycounter;
    BufferDesc *bufHdr;
 
    ResourceOwnerEnlarge(CurrentResourceOwner);
 
    /*
     * Need to get a new buffer.  We use a clock sweep algorithm (essentially
     * the same as what freelist.c does now...)
     */
    trycounter = NLocBuffer;
    for (;;)
    {
        victim_bufid = nextFreeLocalBufId;
 
        if (++nextFreeLocalBufId >= NLocBuffer)
            nextFreeLocalBufId = 0;
 
        bufHdr = GetLocalBufferDescriptor(victim_bufid);
 
        if (LocalRefCount[victim_bufid] == 0)
        {
            uint32      buf_state = pg_atomic_read_u32(&bufHdr->state);
 
            if (BUF_STATE_GET_USAGECOUNT(buf_state) > 0)
            {
                buf_state -= BUF_USAGECOUNT_ONE;
                pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
                trycounter = NLocBuffer;
            }
            else if (BUF_STATE_GET_REFCOUNT(buf_state) > 0)
            {
                /*
                 * This can be reached if the backend initiated AIO for this
                 * buffer and then errored out.
                 */
            }
            else
            {
                /* Found a usable buffer */
                PinLocalBuffer(bufHdr, false);
                break;
            }
        }
        else if (--trycounter == 0)
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                     errmsg("no empty local buffer available")));
    }
 
    /*
     * lazy memory allocation: allocate space on first use of a buffer.
     */
    if (LocalBufHdrGetBlock(bufHdr) == NULL)
    {
        /* Set pointer for use by BufferGetBlock() macro */
        LocalBufHdrGetBlock(bufHdr) = GetLocalBufferStorage();
    }
 
    /*
     * this buffer is not referenced but it might still be dirty. if that's
     * the case, write it out before reusing it!
     */
    if (pg_atomic_read_u32(&bufHdr->state) & BM_DIRTY)
        FlushLocalBuffer(bufHdr, NULL);
 
    /*
     * Remove the victim buffer from the hashtable and mark as invalid.
     */
    if (pg_atomic_read_u32(&bufHdr->state) & BM_TAG_VALID)
    {
        InvalidateLocalBuffer(bufHdr, false);
 
        pgstat_count_io_op(IOOBJECT_TEMP_RELATION, IOCONTEXT_NORMAL, IOOP_EVICT, 1, 0);
    }
 
    return BufferDescriptorGetBuffer(bufHdr);
}

References BM_DIRTY, BM_TAG_VALID, BUF_STATE_GET_REFCOUNT, BUF_STATE_GET_USAGECOUNT, BUF_USAGECOUNT_ONE, BufferDescriptorGetBuffer(), CurrentResourceOwner, ereport, errcode(), errmsg(), ERROR, FlushLocalBuffer(), GetLocalBufferDescriptor(), GetLocalBufferStorage(), InvalidateLocalBuffer(), IOCONTEXT_NORMAL, IOOBJECT_TEMP_RELATION, IOOP_EVICT, LocalBufHdrGetBlock, LocalRefCount, nextFreeLocalBufId, NLocBuffer, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), pgstat_count_io_op(), PinLocalBuffer(), ResourceOwnerEnlarge(), and BufferDesc::state.

Referenced by ExtendBufferedRelLocal(), and LocalBufferAlloc().

InitLocalBuffers()

static void InitLocalBuffers ( void  )

static

Definition at line 719 of file localbuf.c.

{
    int         nbufs = num_temp_buffers;
    HASHCTL     info;
    int         i;
 
    /*
     * Parallel workers can't access data in temporary tables, because they
     * have no visibility into the local buffers of their leader.  This is a
     * convenient, low-cost place to provide a backstop check for that.  Note
     * that we don't wish to prevent a parallel worker from accessing catalog
     * metadata about a temp table, so checks at higher levels would be
     * inappropriate.
     */
    if (IsParallelWorker())
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_TRANSACTION_STATE),
                 errmsg("cannot access temporary tables during a parallel operation")));
 
    /* Allocate and zero buffer headers and auxiliary arrays */
    LocalBufferDescriptors = (BufferDesc *) calloc(nbufs, sizeof(BufferDesc));
    LocalBufferBlockPointers = (Block *) calloc(nbufs, sizeof(Block));
    LocalRefCount = (int32 *) calloc(nbufs, sizeof(int32));
    if (!LocalBufferDescriptors || !LocalBufferBlockPointers || !LocalRefCount)
        ereport(FATAL,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("out of memory")));
 
    nextFreeLocalBufId = 0;
 
    /* initialize fields that need to start off nonzero */
    for (i = 0; i < nbufs; i++)
    {
        BufferDesc *buf = GetLocalBufferDescriptor(i);
 
        /*
         * negative to indicate local buffer. This is tricky: shared buffers
         * start with 0. We have to start with -2. (Note that the routine
         * BufferDescriptorGetBuffer adds 1 to buf_id so our first buffer id
         * is -1.)
         */
        buf->buf_id = -i - 2;
 
        pgaio_wref_clear(&buf->io_wref);
 
        /*
         * Intentionally do not initialize the buffer's atomic variable
         * (besides zeroing the underlying memory above). That way we get
         * errors on platforms without atomics, if somebody (re-)introduces
         * atomic operations for local buffers.
         */
    }
 
    /* Create the lookup hash table */
    info.keysize = sizeof(BufferTag);
    info.entrysize = sizeof(LocalBufferLookupEnt);
 
    LocalBufHash = hash_create("Local Buffer Lookup Table",
                               nbufs,
                               &info,
                               HASH_ELEM | HASH_BLOBS);
 
    if (!LocalBufHash)
        elog(ERROR, "could not initialize local buffer hash table");
 
    /* Initialization done, mark buffers allocated */
    NLocBuffer = nbufs;
}

References buf, calloc, elog, HASHCTL::entrysize, ereport, errcode(), errmsg(), ERROR, FATAL, GetLocalBufferDescriptor(), HASH_BLOBS, hash_create(), HASH_ELEM, i, IsParallelWorker, HASHCTL::keysize, LocalBufferBlockPointers, LocalBufferDescriptors, LocalBufHash, LocalRefCount, nextFreeLocalBufId, NLocBuffer, num_temp_buffers, and pgaio_wref_clear().

Referenced by ExtendBufferedRelLocal(), LocalBufferAlloc(), and PrefetchLocalBuffer().

InvalidateLocalBuffer()

void InvalidateLocalBuffer	(	BufferDesc *	bufHdr,
		bool	check_unreferenced
	)

Definition at line 603 of file localbuf.c.

{
    Buffer      buffer = BufferDescriptorGetBuffer(bufHdr);
    int         bufid = -buffer - 1;
    uint32      buf_state;
    LocalBufferLookupEnt *hresult;
 
    /*
     * It's possible that we started IO on this buffer before e.g. aborting
     * the transaction that created a table. We need to wait for that IO to
     * complete before removing / reusing the buffer.
     */
    if (pgaio_wref_valid(&bufHdr->io_wref))
    {
        PgAioWaitRef iow = bufHdr->io_wref;
 
        pgaio_wref_wait(&iow);
        Assert(!pgaio_wref_valid(&bufHdr->io_wref));
    }
 
    buf_state = pg_atomic_read_u32(&bufHdr->state);
 
    /*
     * We need to test not just LocalRefCount[bufid] but also the BufferDesc
     * itself, as the latter is used to represent a pin by the AIO subsystem.
     * This can happen if AIO is initiated and then the query errors out.
     */
    if (check_unreferenced &&
        (LocalRefCount[bufid] != 0 || BUF_STATE_GET_REFCOUNT(buf_state) != 0))
        elog(ERROR, "block %u of %s is still referenced (local %u)",
             bufHdr->tag.blockNum,
             relpathbackend(BufTagGetRelFileLocator(&bufHdr->tag),
                            MyProcNumber,
                            BufTagGetForkNum(&bufHdr->tag)).str,
             LocalRefCount[bufid]);
 
    /* Remove entry from hashtable */
    hresult = (LocalBufferLookupEnt *)
        hash_search(LocalBufHash, &bufHdr->tag, HASH_REMOVE, NULL);
    if (!hresult)               /* shouldn't happen */
        elog(ERROR, "local buffer hash table corrupted");
    /* Mark buffer invalid */
    ClearBufferTag(&bufHdr->tag);
    buf_state &= ~BUF_FLAG_MASK;
    buf_state &= ~BUF_USAGECOUNT_MASK;
    pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
}

References Assert(), buftag::blockNum, BUF_STATE_GET_REFCOUNT, BufferDescriptorGetBuffer(), BufTagGetForkNum(), BufTagGetRelFileLocator(), ClearBufferTag(), elog, ERROR, HASH_REMOVE, hash_search(), BufferDesc::io_wref, LocalBufHash, LocalRefCount, MyProcNumber, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), pgaio_wref_valid(), pgaio_wref_wait(), relpathbackend, BufferDesc::state, and BufferDesc::tag.

Referenced by DropRelationAllLocalBuffers(), DropRelationLocalBuffers(), GetLocalVictimBuffer(), invalidate_rel_block(), and modify_rel_block().

LimitAdditionalLocalPins()

void LimitAdditionalLocalPins

(

uint32 *

additional_pins

)

Definition at line 322 of file localbuf.c.

{
    uint32      max_pins;
 
    if (*additional_pins <= 1)
        return;
 
    /*
     * In contrast to LimitAdditionalPins() other backends don't play a role
     * here. We can allow up to NLocBuffer pins in total, but it might not be
     * initialized yet so read num_temp_buffers.
     */
    max_pins = (num_temp_buffers - NLocalPinnedBuffers);
 
    if (*additional_pins >= max_pins)
        *additional_pins = max_pins;
}

References NLocalPinnedBuffers, and num_temp_buffers.

Referenced by ExtendBufferedRelLocal().

LocalBufferAlloc()

BufferDesc * LocalBufferAlloc	(	SMgrRelation	smgr,
		ForkNumber	forkNum,
		BlockNumber	blockNum,
		bool *	foundPtr
	)

Definition at line 118 of file localbuf.c.

{
    BufferTag   newTag;         /* identity of requested block */
    LocalBufferLookupEnt *hresult;
    BufferDesc *bufHdr;
    Buffer      victim_buffer;
    int         bufid;
    bool        found;
 
    InitBufferTag(&newTag, &smgr->smgr_rlocator.locator, forkNum, blockNum);
 
    /* Initialize local buffers if first request in this session */
    if (LocalBufHash == NULL)
        InitLocalBuffers();
 
    ResourceOwnerEnlarge(CurrentResourceOwner);
 
    /* See if the desired buffer already exists */
    hresult = (LocalBufferLookupEnt *)
        hash_search(LocalBufHash, &newTag, HASH_FIND, NULL);
 
    if (hresult)
    {
        bufid = hresult->id;
        bufHdr = GetLocalBufferDescriptor(bufid);
        Assert(BufferTagsEqual(&bufHdr->tag, &newTag));
 
        *foundPtr = PinLocalBuffer(bufHdr, true);
    }
    else
    {
        uint32      buf_state;
 
        victim_buffer = GetLocalVictimBuffer();
        bufid = -victim_buffer - 1;
        bufHdr = GetLocalBufferDescriptor(bufid);
 
        hresult = (LocalBufferLookupEnt *)
            hash_search(LocalBufHash, &newTag, HASH_ENTER, &found);
        if (found)              /* shouldn't happen */
            elog(ERROR, "local buffer hash table corrupted");
        hresult->id = bufid;
 
        /*
         * it's all ours now.
         */
        bufHdr->tag = newTag;
 
        buf_state = pg_atomic_read_u32(&bufHdr->state);
        buf_state &= ~(BUF_FLAG_MASK | BUF_USAGECOUNT_MASK);
        buf_state |= BM_TAG_VALID | BUF_USAGECOUNT_ONE;
        pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
 
        *foundPtr = false;
    }
 
    return bufHdr;
}

References Assert(), BM_TAG_VALID, BUF_FLAG_MASK, BUF_USAGECOUNT_MASK, BUF_USAGECOUNT_ONE, BufferTagsEqual(), CurrentResourceOwner, elog, ERROR, GetLocalBufferDescriptor(), GetLocalVictimBuffer(), HASH_ENTER, HASH_FIND, hash_search(), LocalBufferLookupEnt::id, InitBufferTag(), InitLocalBuffers(), LocalBufHash, RelFileLocatorBackend::locator, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), PinLocalBuffer(), ResourceOwnerEnlarge(), SMgrRelationData::smgr_rlocator, BufferDesc::state, and BufferDesc::tag.

Referenced by PinBufferForBlock().

MarkLocalBufferDirty()

void MarkLocalBufferDirty

(

Buffer

buffer

)

Definition at line 489 of file localbuf.c.

{
    int         bufid;
    BufferDesc *bufHdr;
    uint32      buf_state;
 
    Assert(BufferIsLocal(buffer));
 
#ifdef LBDEBUG
    fprintf(stderr, "LB DIRTY %d\n", buffer);
#endif
 
    bufid = -buffer - 1;
 
    Assert(LocalRefCount[bufid] > 0);
 
    bufHdr = GetLocalBufferDescriptor(bufid);
 
    buf_state = pg_atomic_read_u32(&bufHdr->state);
 
    if (!(buf_state & BM_DIRTY))
        pgBufferUsage.local_blks_dirtied++;
 
    buf_state |= BM_DIRTY;
 
    pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
}

References Assert(), BM_DIRTY, BufferIsLocal, fprintf, GetLocalBufferDescriptor(), BufferUsage::local_blks_dirtied, LocalRefCount, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), pgBufferUsage, and BufferDesc::state.

Referenced by MarkBufferDirty(), and MarkBufferDirtyHint().

PinLocalBuffer()

bool PinLocalBuffer	(	BufferDesc *	buf_hdr,
		bool	adjust_usagecount
	)

Definition at line 796 of file localbuf.c.

{
    uint32      buf_state;
    Buffer      buffer = BufferDescriptorGetBuffer(buf_hdr);
    int         bufid = -buffer - 1;
 
    buf_state = pg_atomic_read_u32(&buf_hdr->state);
 
    if (LocalRefCount[bufid] == 0)
    {
        NLocalPinnedBuffers++;
        buf_state += BUF_REFCOUNT_ONE;
        if (adjust_usagecount &&
            BUF_STATE_GET_USAGECOUNT(buf_state) < BM_MAX_USAGE_COUNT)
        {
            buf_state += BUF_USAGECOUNT_ONE;
        }
        pg_atomic_unlocked_write_u32(&buf_hdr->state, buf_state);
 
        /*
         * See comment in PinBuffer().
         *
         * If the buffer isn't allocated yet, it'll be marked as defined in
         * GetLocalBufferStorage().
         */
        if (LocalBufHdrGetBlock(buf_hdr) != NULL)
            VALGRIND_MAKE_MEM_DEFINED(LocalBufHdrGetBlock(buf_hdr), BLCKSZ);
    }
    LocalRefCount[bufid]++;
    ResourceOwnerRememberBuffer(CurrentResourceOwner,
                                BufferDescriptorGetBuffer(buf_hdr));
 
    return buf_state & BM_VALID;
}

References BM_MAX_USAGE_COUNT, BM_VALID, BUF_REFCOUNT_ONE, BUF_STATE_GET_USAGECOUNT, BUF_USAGECOUNT_ONE, BufferDescriptorGetBuffer(), CurrentResourceOwner, LocalBufHdrGetBlock, LocalRefCount, NLocalPinnedBuffers, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), ResourceOwnerRememberBuffer(), BufferDesc::state, and VALGRIND_MAKE_MEM_DEFINED.

Referenced by ExtendBufferedRelLocal(), FlushRelationBuffers(), GetLocalVictimBuffer(), LocalBufferAlloc(), and ReadRecentBuffer().

PrefetchLocalBuffer()

PrefetchBufferResult PrefetchLocalBuffer	(	SMgrRelation	smgr,
		ForkNumber	forkNum,
		BlockNumber	blockNum
	)

Definition at line 71 of file localbuf.c.

{
    PrefetchBufferResult result = {InvalidBuffer, false};
    BufferTag   newTag;         /* identity of requested block */
    LocalBufferLookupEnt *hresult;
 
    InitBufferTag(&newTag, &smgr->smgr_rlocator.locator, forkNum, blockNum);
 
    /* Initialize local buffers if first request in this session */
    if (LocalBufHash == NULL)
        InitLocalBuffers();
 
    /* See if the desired buffer already exists */
    hresult = (LocalBufferLookupEnt *)
        hash_search(LocalBufHash, &newTag, HASH_FIND, NULL);
 
    if (hresult)
    {
        /* Yes, so nothing to do */
        result.recent_buffer = -hresult->id - 1;
    }
    else
    {
#ifdef USE_PREFETCH
        /* Not in buffers, so initiate prefetch */
        if ((io_direct_flags & IO_DIRECT_DATA) == 0 &&
            smgrprefetch(smgr, forkNum, blockNum, 1))
        {
            result.initiated_io = true;
        }
#endif                          /* USE_PREFETCH */
    }
 
    return result;
}

References HASH_FIND, hash_search(), LocalBufferLookupEnt::id, InitBufferTag(), PrefetchBufferResult::initiated_io, InitLocalBuffers(), InvalidBuffer, IO_DIRECT_DATA, io_direct_flags, LocalBufHash, RelFileLocatorBackend::locator, PrefetchBufferResult::recent_buffer, SMgrRelationData::smgr_rlocator, and smgrprefetch().

Referenced by PrefetchBuffer().

StartLocalBufferIO()

bool StartLocalBufferIO	(	BufferDesc *	bufHdr,
		bool	forInput,
		bool	nowait
	)

Definition at line 521 of file localbuf.c.

{
    uint32      buf_state;
 
    /*
     * With AIO the buffer could have IO in progress, e.g. when there are two
     * scans of the same relation. Either wait for the other IO or return
     * false.
     */
    if (pgaio_wref_valid(&bufHdr->io_wref))
    {
        PgAioWaitRef iow = bufHdr->io_wref;
 
        if (nowait)
            return false;
 
        pgaio_wref_wait(&iow);
    }
 
    /* Once we get here, there is definitely no I/O active on this buffer */
 
    /* Check if someone else already did the I/O */
    buf_state = pg_atomic_read_u32(&bufHdr->state);
    if (forInput ? (buf_state & BM_VALID) : !(buf_state & BM_DIRTY))
    {
        return false;
    }
 
    /* BM_IO_IN_PROGRESS isn't currently used for local buffers */
 
    /* local buffers don't track IO using resowners */
 
    return true;
}

References BM_DIRTY, BM_VALID, BufferDesc::io_wref, pg_atomic_read_u32(), pgaio_wref_valid(), pgaio_wref_wait(), and BufferDesc::state.

Referenced by buffer_call_start_io(), ExtendBufferedRelLocal(), FlushLocalBuffer(), read_rel_block_ll(), ReadBuffersCanStartIOOnce(), and ZeroAndLockBuffer().

TerminateLocalBufferIO()

void TerminateLocalBufferIO	(	BufferDesc *	bufHdr,
		bool	clear_dirty,
		uint32	set_flag_bits,
		bool	release_aio
	)

Definition at line 560 of file localbuf.c.

{
    /* Only need to adjust flags */
    uint32      buf_state = pg_atomic_read_u32(&bufHdr->state);
 
    /* BM_IO_IN_PROGRESS isn't currently used for local buffers */
 
    /* Clear earlier errors, if this IO failed, it'll be marked again */
    buf_state &= ~BM_IO_ERROR;
 
    if (clear_dirty)
        buf_state &= ~BM_DIRTY;
 
    if (release_aio)
    {
        /* release pin held by IO subsystem, see also buffer_stage_common() */
        Assert(BUF_STATE_GET_REFCOUNT(buf_state) > 0);
        buf_state -= BUF_REFCOUNT_ONE;
        pgaio_wref_clear(&bufHdr->io_wref);
    }
 
    buf_state |= set_flag_bits;
    pg_atomic_unlocked_write_u32(&bufHdr->state, buf_state);
 
    /* local buffers don't track IO using resowners */
 
    /* local buffers don't use the IO CV, as no other process can see buffer */
 
    /* local buffers don't use BM_PIN_COUNT_WAITER, so no need to wake */
}

References Assert(), BUF_REFCOUNT_ONE, BUF_STATE_GET_REFCOUNT, BufferDesc::io_wref, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), pgaio_wref_clear(), and BufferDesc::state.

Referenced by buffer_call_terminate_io(), buffer_readv_complete_one(), FlushLocalBuffer(), and ZeroAndLockBuffer().

UnpinLocalBuffer()

void UnpinLocalBuffer

(

Buffer

buffer

)

Definition at line 832 of file localbuf.c.

{
    UnpinLocalBufferNoOwner(buffer);
    ResourceOwnerForgetBuffer(CurrentResourceOwner, buffer);
}

References CurrentResourceOwner, ResourceOwnerForgetBuffer(), and UnpinLocalBufferNoOwner().

Referenced by ExtendBufferedRelLocal(), FlushRelationBuffers(), ReleaseAndReadBuffer(), and ReleaseBuffer().

UnpinLocalBufferNoOwner()

void UnpinLocalBufferNoOwner

(

Buffer

buffer

)

Definition at line 839 of file localbuf.c.

{
    int         buffid = -buffer - 1;
 
    Assert(BufferIsLocal(buffer));
    Assert(LocalRefCount[buffid] > 0);
    Assert(NLocalPinnedBuffers > 0);
 
    if (--LocalRefCount[buffid] == 0)
    {
        BufferDesc *buf_hdr = GetLocalBufferDescriptor(buffid);
        uint32      buf_state;
 
        NLocalPinnedBuffers--;
 
        buf_state = pg_atomic_read_u32(&buf_hdr->state);
        Assert(BUF_STATE_GET_REFCOUNT(buf_state) > 0);
        buf_state -= BUF_REFCOUNT_ONE;
        pg_atomic_unlocked_write_u32(&buf_hdr->state, buf_state);
 
        /* see comment in UnpinBufferNoOwner */
        VALGRIND_MAKE_MEM_NOACCESS(LocalBufHdrGetBlock(buf_hdr), BLCKSZ);
    }
}

References Assert(), BUF_REFCOUNT_ONE, BUF_STATE_GET_REFCOUNT, BufferIsLocal, GetLocalBufferDescriptor(), LocalBufHdrGetBlock, LocalRefCount, NLocalPinnedBuffers, pg_atomic_read_u32(), pg_atomic_unlocked_write_u32(), BufferDesc::state, and VALGRIND_MAKE_MEM_NOACCESS.

Referenced by ResOwnerReleaseBufferPin(), and UnpinLocalBuffer().

Variable Documentation

LocalBufferBlockPointers

Block* LocalBufferBlockPointers = NULL

Definition at line 47 of file localbuf.c.

Referenced by BufferGetBlock(), and InitLocalBuffers().

LocalBufferDescriptors

BufferDesc* LocalBufferDescriptors = NULL

Definition at line 46 of file localbuf.c.

Referenced by GetLocalBufferDescriptor(), and InitLocalBuffers().

LocalBufHash

HTAB* LocalBufHash = NULL

static

Definition at line 52 of file localbuf.c.

Referenced by ExtendBufferedRelLocal(), InitLocalBuffers(), InvalidateLocalBuffer(), LocalBufferAlloc(), and PrefetchLocalBuffer().

LocalRefCount

int32* LocalRefCount = NULL

Definition at line 48 of file localbuf.c.

Referenced by CheckBufferIsPinnedOnce(), CheckForLocalBufferLeaks(), ConditionalLockBufferForCleanup(), DebugPrintBufferRefcount(), FlushLocalBuffer(), GetLocalVictimBuffer(), IncrBufferRefCount(), InitLocalBuffers(), InvalidateLocalBuffer(), IsBufferCleanupOK(), MarkLocalBufferDirty(), PinLocalBuffer(), and UnpinLocalBufferNoOwner().

nextFreeLocalBufId

int nextFreeLocalBufId = 0

static

Definition at line 50 of file localbuf.c.

Referenced by GetLocalVictimBuffer(), and InitLocalBuffers().

NLocalPinnedBuffers

int NLocalPinnedBuffers = 0

static

Definition at line 55 of file localbuf.c.

Referenced by GetAdditionalLocalPinLimit(), LimitAdditionalLocalPins(), PinLocalBuffer(), and UnpinLocalBufferNoOwner().

NLocBuffer

int NLocBuffer = 0

Definition at line 44 of file localbuf.c.

Referenced by BufferIsValid(), check_temp_buffers(), CheckForLocalBufferLeaks(), DropRelationAllLocalBuffers(), DropRelationLocalBuffers(), FlushRelationBuffers(), GetLocalBufferStorage(), GetLocalVictimBuffer(), hashbuild(), and InitLocalBuffers().