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 1 of file logtape.h.

LogicalTapeSet

typedef struct LogicalTapeSet LogicalTapeSet

Definition at line 1 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, 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().