logtape.h File Reference

#include "storage/sharedfileset.h"

Include dependency graph for logtape.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	TapeShare

Typedefs
typedef struct LogicalTapeSet	LogicalTapeSet
typedef struct LogicalTape	LogicalTape
typedef struct TapeShare	TapeShare

Functions
LogicalTapeSet *	LogicalTapeSetCreate (bool preallocate, SharedFileSet *fileset, int worker)
void	LogicalTapeClose (LogicalTape *lt)
void	LogicalTapeSetClose (LogicalTapeSet *lts)
LogicalTape *	LogicalTapeCreate (LogicalTapeSet *lts)
LogicalTape *	LogicalTapeImport (LogicalTapeSet *lts, int worker, TapeShare *shared)
void	LogicalTapeSetForgetFreeSpace (LogicalTapeSet *lts)
size_t	LogicalTapeRead (LogicalTape *lt, void *ptr, size_t size)
void	LogicalTapeWrite (LogicalTape *lt, const void *ptr, size_t size)
void	LogicalTapeRewindForRead (LogicalTape *lt, size_t buffer_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)

Typedef Documentation

LogicalTape

typedef struct LogicalTape LogicalTape

Definition at line 26 of file logtape.h.

LogicalTapeSet

typedef struct LogicalTapeSet LogicalTapeSet

Definition at line 25 of file logtape.h.

TapeShare

typedef struct TapeShare TapeShare

Function Documentation

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, 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, 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, TapeBlockGetTrailer, TapeBlockPayloadSize, LogicalTape::tapeSet, and LogicalTape::writing.

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