logtape.c File Reference

#include "postgres.h"
#include <fcntl.h>
#include "storage/buffile.h"
#include "utils/builtins.h"
#include "utils/logtape.h"
#include "utils/memdebug.h"
#include "utils/memutils.h"

Include dependency graph for logtape.c:

Go to the source code of this file.

Data Structures
struct	TapeBlockTrailer
struct	LogicalTape
struct	LogicalTapeSet

Macros
#define	TapeBlockPayloadSize   (BLCKSZ - sizeof(TapeBlockTrailer))
#define	TapeBlockGetTrailer(buf)    ((TapeBlockTrailer *) ((char *) buf + TapeBlockPayloadSize))
#define	TapeBlockIsLast(buf)   (TapeBlockGetTrailer(buf)->next < 0)
#define	TapeBlockGetNBytes(buf)
#define	TapeBlockSetNBytes(buf, nbytes)    (TapeBlockGetTrailer(buf)->next = -(nbytes))
#define	TAPE_WRITE_PREALLOC_MIN   8
#define	TAPE_WRITE_PREALLOC_MAX   128

Typedefs
typedef struct TapeBlockTrailer	TapeBlockTrailer

Functions
static LogicalTape *	ltsCreateTape (LogicalTapeSet *lts)
static void	ltsWriteBlock (LogicalTapeSet *lts, int64 blocknum, const void *buffer)
static void	ltsReadBlock (LogicalTapeSet *lts, int64 blocknum, void *buffer)
static int64	ltsGetBlock (LogicalTapeSet *lts, LogicalTape *lt)
static int64	ltsGetFreeBlock (LogicalTapeSet *lts)
static int64	ltsGetPreallocBlock (LogicalTapeSet *lts, LogicalTape *lt)
static void	ltsReleaseBlock (LogicalTapeSet *lts, int64 blocknum)
static void	ltsInitReadBuffer (LogicalTape *lt)
static bool	ltsReadFillBuffer (LogicalTape *lt)
static uint64	left_offset (uint64 i)
static uint64	right_offset (uint64 i)
static uint64	parent_offset (uint64 i)
LogicalTapeSet *	LogicalTapeSetCreate (bool preallocate, SharedFileSet *fileset, int worker)
LogicalTape *	LogicalTapeImport (LogicalTapeSet *lts, int worker, TapeShare *shared)
void	LogicalTapeSetClose (LogicalTapeSet *lts)
LogicalTape *	LogicalTapeCreate (LogicalTapeSet *lts)
void	LogicalTapeClose (LogicalTape *lt)
void	LogicalTapeSetForgetFreeSpace (LogicalTapeSet *lts)
void	LogicalTapeWrite (LogicalTape *lt, const void *ptr, size_t size)
void	LogicalTapeRewindForRead (LogicalTape *lt, size_t buffer_size)
size_t	LogicalTapeRead (LogicalTape *lt, void *ptr, size_t size)
void	LogicalTapeFreeze (LogicalTape *lt, TapeShare *share)
size_t	LogicalTapeBackspace (LogicalTape *lt, size_t size)
void	LogicalTapeSeek (LogicalTape *lt, int64 blocknum, int offset)
void	LogicalTapeTell (LogicalTape *lt, int64 *blocknum, int *offset)
int64	LogicalTapeSetBlocks (LogicalTapeSet *lts)

Macro Definition Documentation

TAPE_WRITE_PREALLOC_MAX

#define TAPE_WRITE_PREALLOC_MAX   128

Definition at line 125 of file logtape.c.

TAPE_WRITE_PREALLOC_MIN

#define TAPE_WRITE_PREALLOC_MIN   8

Definition at line 124 of file logtape.c.

TapeBlockGetNBytes

#define TapeBlockGetNBytes

(

buf

)

Value:

    (TapeBlockIsLast(buf) ? \
     (- TapeBlockGetTrailer(buf)->next) : TapeBlockPayloadSize)

Definition at line 108 of file logtape.c.

TapeBlockGetTrailer

#define TapeBlockGetTrailer

(

buf

)

((TapeBlockTrailer *) ((char *) buf + TapeBlockPayloadSize))

Definition at line 104 of file logtape.c.

TapeBlockIsLast

#define TapeBlockIsLast

(

buf

)

(TapeBlockGetTrailer(buf)->next < 0)

Definition at line 107 of file logtape.c.

TapeBlockPayloadSize

#define TapeBlockPayloadSize   (BLCKSZ - sizeof(TapeBlockTrailer))

Definition at line 103 of file logtape.c.

TapeBlockSetNBytes

#define TapeBlockSetNBytes	(		buf,
			nbytes
	)		(TapeBlockGetTrailer(buf)->next = -(nbytes))

Definition at line 111 of file logtape.c.

Typedef Documentation

TapeBlockTrailer

typedef struct TapeBlockTrailer TapeBlockTrailer

Function Documentation

left_offset()

static uint64 left_offset ( uint64  i )

inlinestatic

Definition at line 337 of file logtape.c.

{
    return 2 * i + 1;
}

References i.

Referenced by ltsGetFreeBlock().

LogicalTapeBackspace()

size_t LogicalTapeBackspace	(	LogicalTape *	lt,
		size_t	size
	)

Definition at line 1062 of file logtape.c.

{
    size_t      seekpos = 0;
 
    Assert(lt->frozen);
    Assert(lt->buffer_size == BLCKSZ);
 
    if (lt->buffer == NULL)
        ltsInitReadBuffer(lt);
 
    /*
     * Easy case for seek within current block.
     */
    if (size <= (size_t) lt->pos)
    {
        lt->pos -= (int) size;
        return size;
    }
 
    /*
     * Not-so-easy case, have to walk back the chain of blocks.  This
     * implementation would be pretty inefficient for long seeks, but we
     * really aren't doing that (a seek over one tuple is typical).
     */
    seekpos = (size_t) lt->pos; /* part within this block */
    while (size > seekpos)
    {
        int64       prev = TapeBlockGetTrailer(lt->buffer)->prev;
 
        if (prev == -1L)
        {
            /* Tried to back up beyond the beginning of tape. */
            if (lt->curBlockNumber != lt->firstBlockNumber)
                elog(ERROR, "unexpected end of tape");
            lt->pos = 0;
            return seekpos;
        }
 
        ltsReadBlock(lt->tapeSet, prev, lt->buffer);
 
        if (TapeBlockGetTrailer(lt->buffer)->next != lt->curBlockNumber)
            elog(ERROR, "broken tape, next of block %lld is %lld, expected %lld",
                 (long long) prev,
                 (long long) (TapeBlockGetTrailer(lt->buffer)->next),
                 (long long) lt->curBlockNumber);
 
        lt->nbytes = TapeBlockPayloadSize;
        lt->curBlockNumber = prev;
        lt->nextBlockNumber = TapeBlockGetTrailer(lt->buffer)->next;
 
        seekpos += TapeBlockPayloadSize;
    }
 
    /*
     * 'seekpos' can now be greater than 'size', because it points to the
     * beginning the target block.  The difference is the position within the
     * page.
     */
    lt->pos = seekpos - size;
    return size;
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, elog, ERROR, LogicalTape::firstBlockNumber, LogicalTape::frozen, ltsInitReadBuffer(), ltsReadBlock(), LogicalTape::nbytes, LogicalTape::nextBlockNumber, LogicalTape::pos, size, TapeBlockGetTrailer, TapeBlockPayloadSize, LogicalTape::tapeSet, and while().

Referenced by tuplesort_gettuple_common().

LogicalTapeClose()

void LogicalTapeClose

(

LogicalTape *

lt

)

Definition at line 733 of file logtape.c.

{
    if (lt->buffer)
        pfree(lt->buffer);
    pfree(lt);
}

References LogicalTape::buffer, and pfree().

Referenced by agg_refill_hash_table(), mergeruns(), and tuplesort_gettuple_common().

LogicalTapeCreate()

LogicalTape* LogicalTapeCreate

(

LogicalTapeSet *

lts

)

Definition at line 680 of file logtape.c.

{
    /*
     * The only thing that currently prevents creating new tapes in leader is
     * the fact that BufFiles opened using BufFileOpenShared() are read-only
     * by definition, but that could be changed if it seemed worthwhile.  For
     * now, writing to the leader tape will raise a "Bad file descriptor"
     * error, so tuplesort must avoid writing to the leader tape altogether.
     */
    if (lts->fileset && lts->worker == -1)
        elog(ERROR, "cannot create new tapes in leader process");
 
    return ltsCreateTape(lts);
}

References elog, ERROR, LogicalTapeSet::fileset, ltsCreateTape(), and LogicalTapeSet::worker.

Referenced by hashagg_spill_init(), and selectnewtape().

LogicalTapeFreeze()

void LogicalTapeFreeze	(	LogicalTape *	lt,
		TapeShare *	share
	)

Definition at line 981 of file logtape.c.

{
    LogicalTapeSet *lts = lt->tapeSet;
 
    Assert(lt->writing);
    Assert(lt->offsetBlockNumber == 0L);
 
    /*
     * Completion of a write phase.  Flush last partial data block, and rewind
     * for nondestructive read.
     */
    if (lt->dirty)
    {
        /*
         * As long as we've filled the buffer at least once, its contents are
         * entirely defined from valgrind's point of view, even though
         * contents beyond the current end point may be stale.  But it's
         * possible - at least in the case of a parallel sort - to sort such
         * small amount of data that we do not fill the buffer even once. Tell
         * valgrind that its contents are defined, so it doesn't bleat.
         */
        VALGRIND_MAKE_MEM_DEFINED(lt->buffer + lt->nbytes,
                                  lt->buffer_size - lt->nbytes);
 
        TapeBlockSetNBytes(lt->buffer, lt->nbytes);
        ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, lt->buffer);
    }
    lt->writing = false;
    lt->frozen = true;
 
    /*
     * The seek and backspace functions assume a single block read buffer.
     * That's OK with current usage.  A larger buffer is helpful to make the
     * read pattern of the backing file look more sequential to the OS, when
     * we're reading from multiple tapes.  But at the end of a sort, when a
     * tape is frozen, we only read from a single tape anyway.
     */
    if (!lt->buffer || lt->buffer_size != BLCKSZ)
    {
        if (lt->buffer)
            pfree(lt->buffer);
        lt->buffer = palloc(BLCKSZ);
        lt->buffer_size = BLCKSZ;
    }
 
    /* Read the first block, or reset if tape is empty */
    lt->curBlockNumber = lt->firstBlockNumber;
    lt->pos = 0;
    lt->nbytes = 0;
 
    if (lt->firstBlockNumber == -1L)
        lt->nextBlockNumber = -1L;
    ltsReadBlock(lt->tapeSet, lt->curBlockNumber, lt->buffer);
    if (TapeBlockIsLast(lt->buffer))
        lt->nextBlockNumber = -1L;
    else
        lt->nextBlockNumber = TapeBlockGetTrailer(lt->buffer)->next;
    lt->nbytes = TapeBlockGetNBytes(lt->buffer);
 
    /* Handle extra steps when caller is to share its tapeset */
    if (share)
    {
        BufFileExportFileSet(lts->pfile);
        share->firstblocknumber = lt->firstBlockNumber;
    }
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, BufFileExportFileSet(), LogicalTape::curBlockNumber, LogicalTape::dirty, LogicalTape::firstBlockNumber, TapeShare::firstblocknumber, LogicalTape::frozen, ltsReadBlock(), ltsWriteBlock(), LogicalTape::nbytes, LogicalTape::nextBlockNumber, LogicalTape::offsetBlockNumber, palloc(), LogicalTapeSet::pfile, pfree(), LogicalTape::pos, TapeBlockGetNBytes, TapeBlockGetTrailer, TapeBlockIsLast, TapeBlockSetNBytes, LogicalTape::tapeSet, VALGRIND_MAKE_MEM_DEFINED, and LogicalTape::writing.

Referenced by mergeruns(), and worker_freeze_result_tape().

LogicalTapeImport()

LogicalTape* LogicalTapeImport	(	LogicalTapeSet *	lts,
		int	worker,
		TapeShare *	shared
	)

Definition at line 609 of file logtape.c.

{
    LogicalTape *lt;
    int64       tapeblocks;
    char        filename[MAXPGPATH];
    BufFile    *file;
    int64       filesize;
 
    lt = ltsCreateTape(lts);
 
    /*
     * build concatenated view of all buffiles, remembering the block number
     * where each source file begins.
     */
    pg_itoa(worker, filename);
    file = BufFileOpenFileSet(&lts->fileset->fs, filename, O_RDONLY, false);
    filesize = BufFileSize(file);
 
    /*
     * Stash first BufFile, and concatenate subsequent BufFiles to that. Store
     * block offset into each tape as we go.
     */
    lt->firstBlockNumber = shared->firstblocknumber;
    if (lts->pfile == NULL)
    {
        lts->pfile = file;
        lt->offsetBlockNumber = 0L;
    }
    else
    {
        lt->offsetBlockNumber = BufFileAppend(lts->pfile, file);
    }
    /* Don't allocate more for read buffer than could possibly help */
    lt->max_size = Min(MaxAllocSize, filesize);
    tapeblocks = filesize / BLCKSZ;
 
    /*
     * Update # of allocated blocks and # blocks written to reflect the
     * imported BufFile.  Allocated/written blocks include space used by holes
     * left between concatenated BufFiles.  Also track the number of hole
     * blocks so that we can later work backwards to calculate the number of
     * physical blocks for instrumentation.
     */
    lts->nHoleBlocks += lt->offsetBlockNumber - lts->nBlocksAllocated;
 
    lts->nBlocksAllocated = lt->offsetBlockNumber + tapeblocks;
    lts->nBlocksWritten = lts->nBlocksAllocated;
 
    return lt;
}

References BufFileAppend(), BufFileOpenFileSet(), BufFileSize(), filename, LogicalTapeSet::fileset, LogicalTape::firstBlockNumber, TapeShare::firstblocknumber, SharedFileSet::fs, ltsCreateTape(), LogicalTape::max_size, MaxAllocSize, MAXPGPATH, Min, LogicalTapeSet::nBlocksAllocated, LogicalTapeSet::nBlocksWritten, LogicalTapeSet::nHoleBlocks, LogicalTape::offsetBlockNumber, LogicalTapeSet::pfile, and pg_itoa().

Referenced by leader_takeover_tapes().

LogicalTapeRead()

size_t LogicalTapeRead	(	LogicalTape *	lt,
		void *	ptr,
		size_t	size
	)

Definition at line 928 of file logtape.c.

{
    size_t      nread = 0;
    size_t      nthistime;
 
    Assert(!lt->writing);
 
    if (lt->buffer == NULL)
        ltsInitReadBuffer(lt);
 
    while (size > 0)
    {
        if (lt->pos >= lt->nbytes)
        {
            /* Try to load more data into buffer. */
            if (!ltsReadFillBuffer(lt))
                break;          /* EOF */
        }
 
        nthistime = lt->nbytes - lt->pos;
        if (nthistime > size)
            nthistime = size;
        Assert(nthistime > 0);
 
        memcpy(ptr, lt->buffer + lt->pos, nthistime);
 
        lt->pos += nthistime;
        ptr = (char *) ptr + nthistime;
        size -= nthistime;
        nread += nthistime;
    }
 
    return nread;
}

References Assert, LogicalTape::buffer, ltsInitReadBuffer(), ltsReadFillBuffer(), LogicalTape::nbytes, LogicalTape::pos, size, and LogicalTape::writing.

Referenced by getlen(), and hashagg_batch_read().

LogicalTapeRewindForRead()

void LogicalTapeRewindForRead	(	LogicalTape *	lt,
		size_t	buffer_size
	)

Definition at line 846 of file logtape.c.

{
    LogicalTapeSet *lts = lt->tapeSet;
 
    /*
     * Round and cap buffer_size if needed.
     */
    if (lt->frozen)
        buffer_size = BLCKSZ;
    else
    {
        /* need at least one block */
        if (buffer_size < BLCKSZ)
            buffer_size = BLCKSZ;
 
        /* palloc() larger than max_size is unlikely to be helpful */
        if (buffer_size > lt->max_size)
            buffer_size = lt->max_size;
 
        /* round down to BLCKSZ boundary */
        buffer_size -= buffer_size % BLCKSZ;
    }
 
    if (lt->writing)
    {
        /*
         * Completion of a write phase.  Flush last partial data block, and
         * rewind for normal (destructive) read.
         */
        if (lt->dirty)
        {
            /*
             * As long as we've filled the buffer at least once, its contents
             * are entirely defined from valgrind's point of view, even though
             * contents beyond the current end point may be stale.  But it's
             * possible - at least in the case of a parallel sort - to sort
             * such small amount of data that we do not fill the buffer even
             * once.  Tell valgrind that its contents are defined, so it
             * doesn't bleat.
             */
            VALGRIND_MAKE_MEM_DEFINED(lt->buffer + lt->nbytes,
                                      lt->buffer_size - lt->nbytes);
 
            TapeBlockSetNBytes(lt->buffer, lt->nbytes);
            ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, lt->buffer);
        }
        lt->writing = false;
    }
    else
    {
        /*
         * This is only OK if tape is frozen; we rewind for (another) read
         * pass.
         */
        Assert(lt->frozen);
    }
 
    if (lt->buffer)
        pfree(lt->buffer);
 
    /* the buffer is lazily allocated, but set the size here */
    lt->buffer = NULL;
    lt->buffer_size = buffer_size;
 
    /* free the preallocation list, and return unused block numbers */
    if (lt->prealloc != NULL)
    {
        for (int i = lt->nprealloc; i > 0; i--)
            ltsReleaseBlock(lts, lt->prealloc[i - 1]);
        pfree(lt->prealloc);
        lt->prealloc = NULL;
        lt->nprealloc = 0;
        lt->prealloc_size = 0;
    }
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, LogicalTape::dirty, LogicalTape::frozen, i, ltsReleaseBlock(), ltsWriteBlock(), LogicalTape::max_size, LogicalTape::nbytes, LogicalTape::nprealloc, pfree(), LogicalTape::prealloc, LogicalTape::prealloc_size, TapeBlockSetNBytes, LogicalTape::tapeSet, VALGRIND_MAKE_MEM_DEFINED, and LogicalTape::writing.

Referenced by hashagg_spill_finish(), mergeruns(), and tuplesort_rescan().

LogicalTapeSeek()

void LogicalTapeSeek	(	LogicalTape *	lt,
		int64	blocknum,
		int	offset
	)

Definition at line 1133 of file logtape.c.

{
    Assert(lt->frozen);
    Assert(offset >= 0 && offset <= TapeBlockPayloadSize);
    Assert(lt->buffer_size == BLCKSZ);
 
    if (lt->buffer == NULL)
        ltsInitReadBuffer(lt);
 
    if (blocknum != lt->curBlockNumber)
    {
        ltsReadBlock(lt->tapeSet, blocknum, lt->buffer);
        lt->curBlockNumber = blocknum;
        lt->nbytes = TapeBlockPayloadSize;
        lt->nextBlockNumber = TapeBlockGetTrailer(lt->buffer)->next;
    }
 
    if (offset > lt->nbytes)
        elog(ERROR, "invalid tape seek position");
    lt->pos = offset;
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, elog, ERROR, LogicalTape::frozen, ltsInitReadBuffer(), ltsReadBlock(), LogicalTape::nbytes, LogicalTape::nextBlockNumber, LogicalTape::pos, TapeBlockGetTrailer, TapeBlockPayloadSize, and LogicalTape::tapeSet.

Referenced by tuplesort_restorepos().

LogicalTapeSetBlocks()

int64 LogicalTapeSetBlocks

(

LogicalTapeSet *

lts

)

Definition at line 1181 of file logtape.c.

{
    return lts->nBlocksWritten - lts->nHoleBlocks;
}

References LogicalTapeSet::nBlocksWritten, and LogicalTapeSet::nHoleBlocks.

Referenced by hash_agg_update_metrics(), tuplesort_free(), and tuplesort_updatemax().

LogicalTapeSetClose()

void LogicalTapeSetClose

(

LogicalTapeSet *

lts

)

Definition at line 667 of file logtape.c.

{
    BufFileClose(lts->pfile);
    pfree(lts->freeBlocks);
    pfree(lts);
}

References BufFileClose(), LogicalTapeSet::freeBlocks, LogicalTapeSet::pfile, and pfree().

Referenced by hashagg_reset_spill_state(), and tuplesort_free().

LogicalTapeSetCreate()

LogicalTapeSet* LogicalTapeSetCreate	(	bool	preallocate,
		SharedFileSet *	fileset,
		int	worker
	)

Definition at line 556 of file logtape.c.

{
    LogicalTapeSet *lts;
 
    /*
     * Create top-level struct including per-tape LogicalTape structs.
     */
    lts = (LogicalTapeSet *) palloc(sizeof(LogicalTapeSet));
    lts->nBlocksAllocated = 0L;
    lts->nBlocksWritten = 0L;
    lts->nHoleBlocks = 0L;
    lts->forgetFreeSpace = false;
    lts->freeBlocksLen = 32;    /* reasonable initial guess */
    lts->freeBlocks = (int64 *) palloc(lts->freeBlocksLen * sizeof(int64));
    lts->nFreeBlocks = 0;
    lts->enable_prealloc = preallocate;
 
    lts->fileset = fileset;
    lts->worker = worker;
 
    /*
     * Create temp BufFile storage as required.
     *
     * In leader, we hijack the BufFile of the first tape that's imported, and
     * concatenate the BufFiles of any subsequent tapes to that. Hence don't
     * create a BufFile here. Things are simpler for the worker case and the
     * serial case, though.  They are generally very similar -- workers use a
     * shared fileset, whereas serial sorts use a conventional serial BufFile.
     */
    if (fileset && worker == -1)
        lts->pfile = NULL;
    else if (fileset)
    {
        char        filename[MAXPGPATH];
 
        pg_itoa(worker, filename);
        lts->pfile = BufFileCreateFileSet(&fileset->fs, filename);
    }
    else
        lts->pfile = BufFileCreateTemp(false);
 
    return lts;
}

References BufFileCreateFileSet(), BufFileCreateTemp(), LogicalTapeSet::enable_prealloc, filename, LogicalTapeSet::fileset, LogicalTapeSet::forgetFreeSpace, LogicalTapeSet::freeBlocks, LogicalTapeSet::freeBlocksLen, SharedFileSet::fs, MAXPGPATH, LogicalTapeSet::nBlocksAllocated, LogicalTapeSet::nBlocksWritten, LogicalTapeSet::nFreeBlocks, LogicalTapeSet::nHoleBlocks, palloc(), LogicalTapeSet::pfile, pg_itoa(), and LogicalTapeSet::worker.

Referenced by hash_agg_enter_spill_mode(), inittapes(), and leader_takeover_tapes().

LogicalTapeSetForgetFreeSpace()

void LogicalTapeSetForgetFreeSpace

(

LogicalTapeSet *

lts

)

Definition at line 750 of file logtape.c.

{
    lts->forgetFreeSpace = true;
}

References LogicalTapeSet::forgetFreeSpace.

Referenced by mergeruns().

LogicalTapeTell()

void LogicalTapeTell	(	LogicalTape *	lt,
		int64 *	blocknum,
		int *	offset
	)

Definition at line 1162 of file logtape.c.

{
    if (lt->buffer == NULL)
        ltsInitReadBuffer(lt);
 
    Assert(lt->offsetBlockNumber == 0L);
 
    /* With a larger buffer, 'pos' wouldn't be the same as offset within page */
    Assert(lt->buffer_size == BLCKSZ);
 
    *blocknum = lt->curBlockNumber;
    *offset = lt->pos;
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, ltsInitReadBuffer(), LogicalTape::offsetBlockNumber, and LogicalTape::pos.

Referenced by tuplesort_markpos().

LogicalTapeWrite()

void LogicalTapeWrite	(	LogicalTape *	lt,
		const void *	ptr,
		size_t	size
	)

Definition at line 761 of file logtape.c.

{
    LogicalTapeSet *lts = lt->tapeSet;
    size_t      nthistime;
 
    Assert(lt->writing);
    Assert(lt->offsetBlockNumber == 0L);
 
    /* Allocate data buffer and first block on first write */
    if (lt->buffer == NULL)
    {
        lt->buffer = (char *) palloc(BLCKSZ);
        lt->buffer_size = BLCKSZ;
    }
    if (lt->curBlockNumber == -1)
    {
        Assert(lt->firstBlockNumber == -1);
        Assert(lt->pos == 0);
 
        lt->curBlockNumber = ltsGetBlock(lts, lt);
        lt->firstBlockNumber = lt->curBlockNumber;
 
        TapeBlockGetTrailer(lt->buffer)->prev = -1L;
    }
 
    Assert(lt->buffer_size == BLCKSZ);
    while (size > 0)
    {
        if (lt->pos >= (int) TapeBlockPayloadSize)
        {
            /* Buffer full, dump it out */
            int64       nextBlockNumber;
 
            if (!lt->dirty)
            {
                /* Hmm, went directly from reading to writing? */
                elog(ERROR, "invalid logtape state: should be dirty");
            }
 
            /*
             * First allocate the next block, so that we can store it in the
             * 'next' pointer of this block.
             */
            nextBlockNumber = ltsGetBlock(lt->tapeSet, lt);
 
            /* set the next-pointer and dump the current block. */
            TapeBlockGetTrailer(lt->buffer)->next = nextBlockNumber;
            ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, lt->buffer);
 
            /* initialize the prev-pointer of the next block */
            TapeBlockGetTrailer(lt->buffer)->prev = lt->curBlockNumber;
            lt->curBlockNumber = nextBlockNumber;
            lt->pos = 0;
            lt->nbytes = 0;
        }
 
        nthistime = TapeBlockPayloadSize - lt->pos;
        if (nthistime > size)
            nthistime = size;
        Assert(nthistime > 0);
 
        memcpy(lt->buffer + lt->pos, ptr, nthistime);
 
        lt->dirty = true;
        lt->pos += nthistime;
        if (lt->nbytes < lt->pos)
            lt->nbytes = lt->pos;
        ptr = (const char *) ptr + nthistime;
        size -= nthistime;
    }
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, LogicalTape::dirty, elog, ERROR, LogicalTape::firstBlockNumber, ltsGetBlock(), ltsWriteBlock(), LogicalTape::nbytes, LogicalTape::offsetBlockNumber, palloc(), LogicalTape::pos, size, TapeBlockGetTrailer, TapeBlockPayloadSize, LogicalTape::tapeSet, and LogicalTape::writing.

Referenced by hashagg_spill_tuple(), markrunend(), writetup_cluster(), writetup_datum(), writetup_heap(), writetup_index(), and writetup_index_brin().

ltsCreateTape()

static LogicalTape * ltsCreateTape ( LogicalTapeSet *  lts )

static

Definition at line 696 of file logtape.c.

{
    LogicalTape *lt;
 
    /*
     * Create per-tape struct.  Note we allocate the I/O buffer lazily.
     */
    lt = palloc(sizeof(LogicalTape));
    lt->tapeSet = lts;
    lt->writing = true;
    lt->frozen = false;
    lt->dirty = false;
    lt->firstBlockNumber = -1L;
    lt->curBlockNumber = -1L;
    lt->nextBlockNumber = -1L;
    lt->offsetBlockNumber = 0L;
    lt->buffer = NULL;
    lt->buffer_size = 0;
    /* palloc() larger than MaxAllocSize would fail */
    lt->max_size = MaxAllocSize;
    lt->pos = 0;
    lt->nbytes = 0;
    lt->prealloc = NULL;
    lt->nprealloc = 0;
    lt->prealloc_size = 0;
 
    return lt;
}

References LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, LogicalTape::dirty, LogicalTape::firstBlockNumber, LogicalTape::frozen, LogicalTape::max_size, MaxAllocSize, LogicalTape::nbytes, LogicalTape::nextBlockNumber, LogicalTape::nprealloc, LogicalTape::offsetBlockNumber, palloc(), LogicalTape::pos, LogicalTape::prealloc, LogicalTape::prealloc_size, LogicalTape::tapeSet, and LogicalTape::writing.

Referenced by LogicalTapeCreate(), and LogicalTapeImport().

ltsGetBlock()

static int64 ltsGetBlock ( LogicalTapeSet *  lts, LogicalTape *  lt  )

static

Definition at line 358 of file logtape.c.

{
    if (lts->enable_prealloc)
        return ltsGetPreallocBlock(lts, lt);
    else
        return ltsGetFreeBlock(lts);
}

References LogicalTapeSet::enable_prealloc, ltsGetFreeBlock(), and ltsGetPreallocBlock().

Referenced by LogicalTapeWrite().

ltsGetFreeBlock()

static int64 ltsGetFreeBlock ( LogicalTapeSet *  lts )

static

Definition at line 371 of file logtape.c.

{
    int64      *heap = lts->freeBlocks;
    int64       blocknum;
    int64       heapsize;
    int64       holeval;
    uint64      holepos;
 
    /* freelist empty; allocate a new block */
    if (lts->nFreeBlocks == 0)
        return lts->nBlocksAllocated++;
 
    /* easy if heap contains one element */
    if (lts->nFreeBlocks == 1)
    {
        lts->nFreeBlocks--;
        return lts->freeBlocks[0];
    }
 
    /* remove top of minheap */
    blocknum = heap[0];
 
    /* we'll replace it with end of minheap array */
    holeval = heap[--lts->nFreeBlocks];
 
    /* sift down */
    holepos = 0;                /* holepos is where the "hole" is */
    heapsize = lts->nFreeBlocks;
    while (true)
    {
        uint64      left = left_offset(holepos);
        uint64      right = right_offset(holepos);
        uint64      min_child;
 
        if (left < heapsize && right < heapsize)
            min_child = (heap[left] < heap[right]) ? left : right;
        else if (left < heapsize)
            min_child = left;
        else if (right < heapsize)
            min_child = right;
        else
            break;
 
        if (heap[min_child] >= holeval)
            break;
 
        heap[holepos] = heap[min_child];
        holepos = min_child;
    }
    heap[holepos] = holeval;
 
    return blocknum;
}

References LogicalTapeSet::freeBlocks, left_offset(), LogicalTapeSet::nBlocksAllocated, LogicalTapeSet::nFreeBlocks, and right_offset().

Referenced by ltsGetBlock(), and ltsGetPreallocBlock().

ltsGetPreallocBlock()

static int64 ltsGetPreallocBlock ( LogicalTapeSet *  lts, LogicalTape *  lt  )

static

Definition at line 431 of file logtape.c.

{
    /* sorted in descending order, so return the last element */
    if (lt->nprealloc > 0)
        return lt->prealloc[--lt->nprealloc];
 
    if (lt->prealloc == NULL)
    {
        lt->prealloc_size = TAPE_WRITE_PREALLOC_MIN;
        lt->prealloc = (int64 *) palloc(sizeof(int64) * lt->prealloc_size);
    }
    else if (lt->prealloc_size < TAPE_WRITE_PREALLOC_MAX)
    {
        /* when the preallocation list runs out, double the size */
        lt->prealloc_size *= 2;
        if (lt->prealloc_size > TAPE_WRITE_PREALLOC_MAX)
            lt->prealloc_size = TAPE_WRITE_PREALLOC_MAX;
        lt->prealloc = (int64 *) repalloc(lt->prealloc,
                                          sizeof(int64) * lt->prealloc_size);
    }
 
    /* refill preallocation list */
    lt->nprealloc = lt->prealloc_size;
    for (int i = lt->nprealloc; i > 0; i--)
    {
        lt->prealloc[i - 1] = ltsGetFreeBlock(lts);
 
        /* verify descending order */
        Assert(i == lt->nprealloc || lt->prealloc[i - 1] > lt->prealloc[i]);
    }
 
    return lt->prealloc[--lt->nprealloc];
}

References Assert, i, ltsGetFreeBlock(), LogicalTape::nprealloc, palloc(), LogicalTape::prealloc, LogicalTape::prealloc_size, repalloc(), TAPE_WRITE_PREALLOC_MAX, and TAPE_WRITE_PREALLOC_MIN.

Referenced by ltsGetBlock().

ltsInitReadBuffer()

static void ltsInitReadBuffer ( LogicalTape *  lt )

static

Definition at line 522 of file logtape.c.

{
    Assert(lt->buffer_size > 0);
    lt->buffer = palloc(lt->buffer_size);
 
    /* Read the first block, or reset if tape is empty */
    lt->nextBlockNumber = lt->firstBlockNumber;
    lt->pos = 0;
    lt->nbytes = 0;
    ltsReadFillBuffer(lt);
}

References Assert, LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::firstBlockNumber, ltsReadFillBuffer(), LogicalTape::nbytes, LogicalTape::nextBlockNumber, palloc(), and LogicalTape::pos.

Referenced by LogicalTapeBackspace(), LogicalTapeRead(), LogicalTapeSeek(), and LogicalTapeTell().

ltsReadBlock()

static void ltsReadBlock ( LogicalTapeSet *  lts, int64  blocknum, void *  buffer  )

static

Definition at line 282 of file logtape.c.

{
    if (BufFileSeekBlock(lts->pfile, blocknum) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not seek to block %lld of temporary file",
                        (long long) blocknum)));
    BufFileReadExact(lts->pfile, buffer, BLCKSZ);
}

References BufFileReadExact(), BufFileSeekBlock(), ereport, errcode_for_file_access(), errmsg(), ERROR, and LogicalTapeSet::pfile.

Referenced by LogicalTapeBackspace(), LogicalTapeFreeze(), LogicalTapeSeek(), and ltsReadFillBuffer().

ltsReadFillBuffer()

static bool ltsReadFillBuffer ( LogicalTape *  lt )

static

Definition at line 298 of file logtape.c.

{
    lt->pos = 0;
    lt->nbytes = 0;
 
    do
    {
        char       *thisbuf = lt->buffer + lt->nbytes;
        int64       datablocknum = lt->nextBlockNumber;
 
        /* Fetch next block number */
        if (datablocknum == -1L)
            break;              /* EOF */
        /* Apply worker offset, needed for leader tapesets */
        datablocknum += lt->offsetBlockNumber;
 
        /* Read the block */
        ltsReadBlock(lt->tapeSet, datablocknum, thisbuf);
        if (!lt->frozen)
            ltsReleaseBlock(lt->tapeSet, datablocknum);
        lt->curBlockNumber = lt->nextBlockNumber;
 
        lt->nbytes += TapeBlockGetNBytes(thisbuf);
        if (TapeBlockIsLast(thisbuf))
        {
            lt->nextBlockNumber = -1L;
            /* EOF */
            break;
        }
        else
            lt->nextBlockNumber = TapeBlockGetTrailer(thisbuf)->next;
 
        /* Advance to next block, if we have buffer space left */
    } while (lt->buffer_size - lt->nbytes > BLCKSZ);
 
    return (lt->nbytes > 0);
}

References LogicalTape::buffer, LogicalTape::buffer_size, LogicalTape::curBlockNumber, LogicalTape::frozen, ltsReadBlock(), ltsReleaseBlock(), LogicalTape::nbytes, LogicalTape::nextBlockNumber, LogicalTape::offsetBlockNumber, LogicalTape::pos, TapeBlockGetNBytes, TapeBlockGetTrailer, TapeBlockIsLast, and LogicalTape::tapeSet.

Referenced by LogicalTapeRead(), and ltsInitReadBuffer().

ltsReleaseBlock()

static void ltsReleaseBlock ( LogicalTapeSet *  lts, int64  blocknum  )

static

Definition at line 469 of file logtape.c.

{
    int64      *heap;
    uint64      holepos;
 
    /*
     * Do nothing if we're no longer interested in remembering free space.
     */
    if (lts->forgetFreeSpace)
        return;
 
    /*
     * Enlarge freeBlocks array if full.
     */
    if (lts->nFreeBlocks >= lts->freeBlocksLen)
    {
        /*
         * If the freelist becomes very large, just return and leak this free
         * block.
         */
        if (lts->freeBlocksLen * 2 * sizeof(int64) > MaxAllocSize)
            return;
 
        lts->freeBlocksLen *= 2;
        lts->freeBlocks = (int64 *) repalloc(lts->freeBlocks,
                                             lts->freeBlocksLen * sizeof(int64));
    }
 
    /* create a "hole" at end of minheap array */
    heap = lts->freeBlocks;
    holepos = lts->nFreeBlocks;
    lts->nFreeBlocks++;
 
    /* sift up to insert blocknum */
    while (holepos != 0)
    {
        uint64      parent = parent_offset(holepos);
 
        if (heap[parent] < blocknum)
            break;
 
        heap[holepos] = heap[parent];
        holepos = parent;
    }
    heap[holepos] = blocknum;
}

References LogicalTapeSet::forgetFreeSpace, LogicalTapeSet::freeBlocks, LogicalTapeSet::freeBlocksLen, MaxAllocSize, LogicalTapeSet::nFreeBlocks, parent_offset(), and repalloc().

Referenced by LogicalTapeRewindForRead(), and ltsReadFillBuffer().

ltsWriteBlock()

static void ltsWriteBlock ( LogicalTapeSet *  lts, int64  blocknum, const void *  buffer  )

static

Definition at line 238 of file logtape.c.

{
    /*
     * BufFile does not support "holes", so if we're about to write a block
     * that's past the current end of file, fill the space between the current
     * end of file and the target block with zeros.
     *
     * This can happen either when tapes preallocate blocks; or for the last
     * block of a tape which might not have been flushed.
     *
     * Note that BufFile concatenation can leave "holes" in BufFile between
     * worker-owned block ranges.  These are tracked for reporting purposes
     * only.  We never read from nor write to these hole blocks, and so they
     * are not considered here.
     */
    while (blocknum > lts->nBlocksWritten)
    {
        PGIOAlignedBlock zerobuf;
 
        MemSet(zerobuf.data, 0, sizeof(zerobuf));
 
        ltsWriteBlock(lts, lts->nBlocksWritten, zerobuf.data);
    }
 
    /* Write the requested block */
    if (BufFileSeekBlock(lts->pfile, blocknum) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not seek to block %lld of temporary file",
                        (long long) blocknum)));
    BufFileWrite(lts->pfile, buffer, BLCKSZ);
 
    /* Update nBlocksWritten, if we extended the file */
    if (blocknum == lts->nBlocksWritten)
        lts->nBlocksWritten++;
}

References BufFileSeekBlock(), BufFileWrite(), PGIOAlignedBlock::data, ereport, errcode_for_file_access(), errmsg(), ERROR, MemSet, LogicalTapeSet::nBlocksWritten, and LogicalTapeSet::pfile.

Referenced by LogicalTapeFreeze(), LogicalTapeRewindForRead(), and LogicalTapeWrite().

parent_offset()

static uint64 parent_offset ( uint64  i )

inlinestatic

Definition at line 349 of file logtape.c.

{
    return (i - 1) / 2;
}

References i.

Referenced by ltsReleaseBlock().

right_offset()

static uint64 right_offset ( uint64  i )

inlinestatic

Definition at line 343 of file logtape.c.

{
    return 2 * i + 2;
}

References i.

Referenced by ltsGetFreeBlock().