/*------------------------------------------------------------------------- * * logtape.c * Management of "logical tapes" within temporary files. * * This module exists to support sorting via multiple merge passes (see * tuplesort.c). Merging is an ideal algorithm for tape devices, but if * we implement it on disk by creating a separate file for each "tape", * there is an annoying problem: the peak space usage is at least twice * the volume of actual data to be sorted. (This must be so because each * datum will appear in both the input and output tapes of the final * merge pass.) * * We can work around this problem by recognizing that any one tape * dataset (with the possible exception of the final output) is written * and read exactly once in a perfectly sequential manner. Therefore, * a datum once read will not be required again, and we can recycle its * space for use by the new tape dataset(s) being generated. In this way, * the total space usage is essentially just the actual data volume, plus * insignificant bookkeeping and start/stop overhead. * * Few OSes allow arbitrary parts of a file to be released back to the OS, * so we have to implement this space-recycling ourselves within a single * logical file. logtape.c exists to perform this bookkeeping and provide * the illusion of N independent tape devices to tuplesort.c. Note that * logtape.c itself depends on buffile.c to provide a "logical file" of * larger size than the underlying OS may support. * * For simplicity, we allocate and release space in the underlying file * in BLCKSZ-size blocks. Space allocation boils down to keeping track * of which blocks in the underlying file belong to which logical tape, * plus any blocks that are free (recycled and not yet reused). * The blocks in each logical tape form a chain, with a prev- and next- * pointer in each block. * * The initial write pass is guaranteed to fill the underlying file * perfectly sequentially, no matter how data is divided into logical tapes. * Once we begin merge passes, the access pattern becomes considerably * less predictable --- but the seeking involved should be comparable to * what would happen if we kept each logical tape in a separate file, * so there's no serious performance penalty paid to obtain the space * savings of recycling. We try to localize the write accesses by always * writing to the lowest-numbered free block when we have a choice; it's * not clear this helps much, but it can't hurt. (XXX perhaps a LIFO * policy for free blocks would be better?) * * To further make the I/Os more sequential, we can use a larger buffer * when reading, and read multiple blocks from the same tape in one go, * whenever the buffer becomes empty. * * To support the above policy of writing to the lowest free block, the * freelist is a min heap. * * Since all the bookkeeping and buffer memory is allocated with palloc(), * and the underlying file(s) are made with OpenTemporaryFile, all resources * for a logical tape set are certain to be cleaned up even if processing * is aborted by ereport(ERROR). To avoid confusion, the caller should take * care that all calls for a single LogicalTapeSet are made in the same * palloc context. * * To support parallel sort operations involving coordinated callers to * tuplesort.c routines across multiple workers, it is necessary to * concatenate each worker BufFile/tapeset into one single logical tapeset * managed by the leader. Workers should have produced one final * materialized tape (their entire output) when this happens in leader. * There will always be the same number of runs as input tapes, and the same * number of input tapes as participants (worker Tuplesortstates). * * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * IDENTIFICATION * src/backend/utils/sort/logtape.c * *------------------------------------------------------------------------- */ #include "postgres.h" #include #include "storage/buffile.h" #include "utils/builtins.h" #include "utils/logtape.h" #include "utils/memdebug.h" #include "utils/memutils.h" /* * A TapeBlockTrailer is stored at the end of each BLCKSZ block. * * The first block of a tape has prev == -1. The last block of a tape * stores the number of valid bytes on the block, inverted, in 'next' * Therefore next < 0 indicates the last block. */ typedef struct TapeBlockTrailer { int64 prev; /* previous block on this tape, or -1 on first * block */ int64 next; /* next block on this tape, or # of valid * bytes on last block (if < 0) */ } TapeBlockTrailer; #define TapeBlockPayloadSize (BLCKSZ - sizeof(TapeBlockTrailer)) #define TapeBlockGetTrailer(buf) \ ((TapeBlockTrailer *) ((char *) buf + TapeBlockPayloadSize)) #define TapeBlockIsLast(buf) (TapeBlockGetTrailer(buf)->next < 0) #define TapeBlockGetNBytes(buf) \ (TapeBlockIsLast(buf) ? \ (- TapeBlockGetTrailer(buf)->next) : TapeBlockPayloadSize) #define TapeBlockSetNBytes(buf, nbytes) \ (TapeBlockGetTrailer(buf)->next = -(nbytes)) /* * When multiple tapes are being written to concurrently (as in HashAgg), * avoid excessive fragmentation by preallocating block numbers to individual * tapes. Each preallocation doubles in size starting at * TAPE_WRITE_PREALLOC_MIN blocks up to TAPE_WRITE_PREALLOC_MAX blocks. * * No filesystem operations are performed for preallocation; only the block * numbers are reserved. This may lead to sparse writes, which will cause * ltsWriteBlock() to fill in holes with zeros. */ #define TAPE_WRITE_PREALLOC_MIN 8 #define TAPE_WRITE_PREALLOC_MAX 128 /* * This data structure represents a single "logical tape" within the set * of logical tapes stored in the same file. * * While writing, we hold the current partially-written data block in the * buffer. While reading, we can hold multiple blocks in the buffer. Note * that we don't retain the trailers of a block when it's read into the * buffer. The buffer therefore contains one large contiguous chunk of data * from the tape. */ struct LogicalTape { LogicalTapeSet *tapeSet; /* tape set this tape is part of */ bool writing; /* T while in write phase */ bool frozen; /* T if blocks should not be freed when read */ bool dirty; /* does buffer need to be written? */ /* * Block numbers of the first, current, and next block of the tape. * * The "current" block number is only valid when writing, or reading from * a frozen tape. (When reading from an unfrozen tape, we use a larger * read buffer that holds multiple blocks, so the "current" block is * ambiguous.) * * When concatenation of worker tape BufFiles is performed, an offset to * the first block in the unified BufFile space is applied during reads. */ int64 firstBlockNumber; int64 curBlockNumber; int64 nextBlockNumber; int64 offsetBlockNumber; /* * Buffer for current data block(s). */ char *buffer; /* physical buffer (separately palloc'd) */ int buffer_size; /* allocated size of the buffer */ int max_size; /* highest useful, safe buffer_size */ int pos; /* next read/write position in buffer */ int nbytes; /* total # of valid bytes in buffer */ /* * Preallocated block numbers are held in an array sorted in descending * order; blocks are consumed from the end of the array (lowest block * numbers first). */ int64 *prealloc; int nprealloc; /* number of elements in list */ int prealloc_size; /* number of elements list can hold */ }; /* * This data structure represents a set of related "logical tapes" sharing * space in a single underlying file. (But that "file" may be multiple files * if needed to escape OS limits on file size; buffile.c handles that for us.) * Tapes belonging to a tape set can be created and destroyed on-the-fly, on * demand. */ struct LogicalTapeSet { BufFile *pfile; /* underlying file for whole tape set */ SharedFileSet *fileset; int worker; /* worker # if shared, -1 for leader/serial */ /* * File size tracking. nBlocksWritten is the size of the underlying file, * in BLCKSZ blocks. nBlocksAllocated is the number of blocks allocated * by ltsReleaseBlock(), and it is always greater than or equal to * nBlocksWritten. Blocks between nBlocksAllocated and nBlocksWritten are * blocks that have been allocated for a tape, but have not been written * to the underlying file yet. nHoleBlocks tracks the total number of * blocks that are in unused holes between worker spaces following BufFile * concatenation. */ int64 nBlocksAllocated; /* # of blocks allocated */ int64 nBlocksWritten; /* # of blocks used in underlying file */ int64 nHoleBlocks; /* # of "hole" blocks left */ /* * We store the numbers of recycled-and-available blocks in freeBlocks[]. * When there are no such blocks, we extend the underlying file. * * If forgetFreeSpace is true then any freed blocks are simply forgotten * rather than being remembered in freeBlocks[]. See notes for * LogicalTapeSetForgetFreeSpace(). */ bool forgetFreeSpace; /* are we remembering free blocks? */ int64 *freeBlocks; /* resizable array holding minheap */ int64 nFreeBlocks; /* # of currently free blocks */ Size freeBlocksLen; /* current allocated length of freeBlocks[] */ bool enable_prealloc; /* preallocate write blocks? */ }; 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); /* * Write a block-sized buffer to the specified block of the underlying file. * * No need for an error return convention; we ereport() on any error. */ static void ltsWriteBlock(LogicalTapeSet *lts, int64 blocknum, const void *buffer) { /* * 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 %" PRId64 " of temporary file", blocknum))); BufFileWrite(lts->pfile, buffer, BLCKSZ); /* Update nBlocksWritten, if we extended the file */ if (blocknum == lts->nBlocksWritten) lts->nBlocksWritten++; } /* * Read a block-sized buffer from the specified block of the underlying file. * * No need for an error return convention; we ereport() on any error. This * module should never attempt to read a block it doesn't know is there. */ static void ltsReadBlock(LogicalTapeSet *lts, int64 blocknum, void *buffer) { if (BufFileSeekBlock(lts->pfile, blocknum) != 0) ereport(ERROR, (errcode_for_file_access(), errmsg("could not seek to block %" PRId64 " of temporary file", blocknum))); BufFileReadExact(lts->pfile, buffer, BLCKSZ); } /* * Read as many blocks as we can into the per-tape buffer. * * Returns true if anything was read, 'false' on EOF. */ static bool ltsReadFillBuffer(LogicalTape *lt) { 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); } static inline uint64 left_offset(uint64 i) { return 2 * i + 1; } static inline uint64 right_offset(uint64 i) { return 2 * i + 2; } static inline uint64 parent_offset(uint64 i) { return (i - 1) / 2; } /* * Get the next block for writing. */ static int64 ltsGetBlock(LogicalTapeSet *lts, LogicalTape *lt) { if (lts->enable_prealloc) return ltsGetPreallocBlock(lts, lt); else return ltsGetFreeBlock(lts); } /* * Select the lowest currently unused block from the tape set's global free * list min heap. */ static int64 ltsGetFreeBlock(LogicalTapeSet *lts) { 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; } /* * Return the lowest free block number from the tape's preallocation list. * Refill the preallocation list with blocks from the tape set's free list if * necessary. */ static int64 ltsGetPreallocBlock(LogicalTapeSet *lts, LogicalTape *lt) { /* 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]; } /* * Return a block# to the freelist. */ static void ltsReleaseBlock(LogicalTapeSet *lts, int64 blocknum) { 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; } /* * Lazily allocate and initialize the read buffer. This avoids waste when many * tapes are open at once, but not all are active between rewinding and * reading. */ static void ltsInitReadBuffer(LogicalTape *lt) { 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); } /* * Create a tape set, backed by a temporary underlying file. * * The tape set is initially empty. Use LogicalTapeCreate() to create * tapes in it. * * In a single-process sort, pass NULL argument for fileset, and -1 for * worker. * * In a parallel sort, parallel workers pass the shared fileset handle and * their own worker number. After the workers have finished, create the * tape set in the leader, passing the shared fileset handle and -1 for * worker, and use LogicalTapeImport() to import the worker tapes into it. * * Currently, the leader will only import worker tapes into the set, it does * not create tapes of its own, although in principle that should work. * * If preallocate is true, blocks for each individual tape are allocated in * batches. This avoids fragmentation when writing multiple tapes at the * same time. */ LogicalTapeSet * LogicalTapeSetCreate(bool preallocate, SharedFileSet *fileset, int worker) { 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; } /* * Claim ownership of a logical tape from an existing shared BufFile. * * Caller should be leader process. Though tapes are marked as frozen in * workers, they are not frozen when opened within leader, since unfrozen tapes * use a larger read buffer. (Frozen tapes have smaller read buffer, optimized * for random access.) */ LogicalTape * LogicalTapeImport(LogicalTapeSet *lts, int worker, TapeShare *shared) { 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(<s->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; } /* * Close a logical tape set and release all resources. * * NOTE: This doesn't close any of the tapes! You must close them * first, or you can let them be destroyed along with the memory context. */ void LogicalTapeSetClose(LogicalTapeSet *lts) { BufFileClose(lts->pfile); pfree(lts->freeBlocks); pfree(lts); } /* * Create a logical tape in the given tapeset. * * The tape is initialized in write state. */ LogicalTape * LogicalTapeCreate(LogicalTapeSet *lts) { /* * 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); } static LogicalTape * ltsCreateTape(LogicalTapeSet *lts) { 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; } /* * Close a logical tape. * * Note: This doesn't return any blocks to the free list! You must read * the tape to the end first, to reuse the space. In current use, though, * we only close tapes after fully reading them. */ void LogicalTapeClose(LogicalTape *lt) { if (lt->buffer) pfree(lt->buffer); pfree(lt); } /* * Mark a logical tape set as not needing management of free space anymore. * * This should be called if the caller does not intend to write any more data * into the tape set, but is reading from un-frozen tapes. Since no more * writes are planned, remembering free blocks is no longer useful. Setting * this flag lets us avoid wasting time and space in ltsReleaseBlock(), which * is not designed to handle large numbers of free blocks. */ void LogicalTapeSetForgetFreeSpace(LogicalTapeSet *lts) { lts->forgetFreeSpace = true; } /* * Write to a logical tape. * * There are no error returns; we ereport() on failure. */ void LogicalTapeWrite(LogicalTape *lt, const void *ptr, size_t size) { 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; } } /* * Rewind logical tape and switch from writing to reading. * * The tape must currently be in writing state, or "frozen" in read state. * * 'buffer_size' specifies how much memory to use for the read buffer. * Regardless of the argument, the actual amount of memory used is between * BLCKSZ and MaxAllocSize, and is a multiple of BLCKSZ. The given value is * rounded down and truncated to fit those constraints, if necessary. If the * tape is frozen, the 'buffer_size' argument is ignored, and a small BLCKSZ * byte buffer is used. */ void LogicalTapeRewindForRead(LogicalTape *lt, size_t buffer_size) { 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; } } /* * Read from a logical tape. * * Early EOF is indicated by return value less than #bytes requested. */ size_t LogicalTapeRead(LogicalTape *lt, void *ptr, size_t size) { 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; } /* * "Freeze" the contents of a tape so that it can be read multiple times * and/or read backwards. Once a tape is frozen, its contents will not * be released until the LogicalTapeSet is destroyed. This is expected * to be used only for the final output pass of a merge. * * This *must* be called just at the end of a write pass, before the * tape is rewound (after rewind is too late!). It performs a rewind * and switch to read mode "for free". An immediately following rewind- * for-read call is OK but not necessary. * * share output argument is set with details of storage used for tape after * freezing, which may be passed to LogicalTapeSetCreate within leader * process later. This metadata is only of interest to worker callers * freezing their final output for leader (single materialized tape). * Serial sorts should set share to NULL. */ void LogicalTapeFreeze(LogicalTape *lt, TapeShare *share) { 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; } } /* * Backspace the tape a given number of bytes. (We also support a more * general seek interface, see below.) * * *Only* a frozen-for-read tape can be backed up; we don't support * random access during write, and an unfrozen read tape may have * already discarded the desired data! * * Returns the number of bytes backed up. It can be less than the * requested amount, if there isn't that much data before the current * position. The tape is positioned to the beginning of the tape in * that case. */ size_t LogicalTapeBackspace(LogicalTape *lt, size_t size) { 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 %" PRId64 " is %" PRId64 ", expected %" PRId64, prev, TapeBlockGetTrailer(lt->buffer)->next, 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; } /* * Seek to an arbitrary position in a logical tape. * * *Only* a frozen-for-read tape can be seeked. * * Must be called with a block/offset previously returned by * LogicalTapeTell(). */ void LogicalTapeSeek(LogicalTape *lt, int64 blocknum, int offset) { 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; } /* * Obtain current position in a form suitable for a later LogicalTapeSeek. * * NOTE: it'd be OK to do this during write phase with intention of using * the position for a seek after freezing. Not clear if anyone needs that. */ void LogicalTapeTell(LogicalTape *lt, int64 *blocknum, int *offset) { 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; } /* * Obtain total disk space currently used by a LogicalTapeSet, in blocks. Does * not account for open write buffer, if any. */ int64 LogicalTapeSetBlocks(LogicalTapeSet *lts) { return lts->nBlocksWritten - lts->nHoleBlocks; }