PostgreSQL Source Code git master
Data Structures | Macros | Typedefs | Functions
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:
#define TapeBlockGetTrailer(buf)
Definition: logtape.c:104
#define TapeBlockPayloadSize
Definition: logtape.c:103
#define TapeBlockIsLast(buf)
Definition: logtape.c:107
static char * buf
Definition: pg_test_fsync.c:72

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.

338{
339 return 2 * i + 1;
340}
i
int i
Definition: isn.c:77

References i.

Referenced by ltsGetFreeBlock().

◆ LogicalTapeBackspace()

size_t LogicalTapeBackspace ( LogicalTape *  lt,
size_t  size 
)

Definition at line 1062 of file logtape.c.

1063{
1064 size_t seekpos = 0;
1065
1066 Assert(lt->frozen);
1067 Assert(lt->buffer_size == BLCKSZ);
1068
1069 if (lt->buffer == NULL)
1071
1072 /*
1073 * Easy case for seek within current block.
1074 */
1075 if (size <= (size_t) lt->pos)
1076 {
1077 lt->pos -= (int) size;
1078 return size;
1079 }
1080
1081 /*
1082 * Not-so-easy case, have to walk back the chain of blocks. This
1083 * implementation would be pretty inefficient for long seeks, but we
1084 * really aren't doing that (a seek over one tuple is typical).
1085 */
1086 seekpos = (size_t) lt->pos; /* part within this block */
1087 while (size > seekpos)
1088 {
1089 int64 prev = TapeBlockGetTrailer(lt->buffer)->prev;
1090
1091 if (prev == -1L)
1092 {
1093 /* Tried to back up beyond the beginning of tape. */
1094 if (lt->curBlockNumber != lt->firstBlockNumber)
1095 elog(ERROR, "unexpected end of tape");
1096 lt->pos = 0;
1097 return seekpos;
1098 }
1099
1100 ltsReadBlock(lt->tapeSet, prev, lt->buffer);
1101
1102 if (TapeBlockGetTrailer(lt->buffer)->next != lt->curBlockNumber)
1103 elog(ERROR, "broken tape, next of block %" PRId64 " is %" PRId64 ", expected %" PRId64,
1104 prev,
1105 TapeBlockGetTrailer(lt->buffer)->next,
1106 lt->curBlockNumber);
1107
1109 lt->curBlockNumber = prev;
1111
1112 seekpos += TapeBlockPayloadSize;
1113 }
1114
1115 /*
1116 * 'seekpos' can now be greater than 'size', because it points to the
1117 * beginning the target block. The difference is the position within the
1118 * page.
1119 */
1120 lt->pos = seekpos - size;
1121 return size;
1122}
int64_t int64
Definition: c.h:535
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:226
Assert(PointerIsAligned(start, uint64))
static void ltsInitReadBuffer(LogicalTape *lt)
Definition: logtape.c:522
static void ltsReadBlock(LogicalTapeSet *lts, int64 blocknum, void *buffer)
Definition: logtape.c:282
while(p+4<=pend)
int64 curBlockNumber
Definition: logtape.c:157
int64 firstBlockNumber
Definition: logtape.c:156
int pos
Definition: logtape.c:167
char * buffer
Definition: logtape.c:164
int64 nextBlockNumber
Definition: logtape.c:158
bool frozen
Definition: logtape.c:142
int buffer_size
Definition: logtape.c:165
LogicalTapeSet * tapeSet
Definition: logtape.c:139
int nbytes
Definition: logtape.c:168

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.

734{
735 if (lt->buffer)
736 pfree(lt->buffer);
737 pfree(lt);
738}
void pfree(void *pointer)
Definition: mcxt.c:1594

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.

681{
682 /*
683 * The only thing that currently prevents creating new tapes in leader is
684 * the fact that BufFiles opened using BufFileOpenFileSet() are read-only
685 * by definition, but that could be changed if it seemed worthwhile. For
686 * now, writing to the leader tape will raise a "Bad file descriptor"
687 * error, so tuplesort must avoid writing to the leader tape altogether.
688 */
689 if (lts->fileset && lts->worker == -1)
690 elog(ERROR, "cannot create new tapes in leader process");
691
692 return ltsCreateTape(lts);
693}
static LogicalTape * ltsCreateTape(LogicalTapeSet *lts)
Definition: logtape.c:696
SharedFileSet * fileset
Definition: logtape.c:190
int worker
Definition: logtape.c:191

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.

982{
983 LogicalTapeSet *lts = lt->tapeSet;
984
985 Assert(lt->writing);
986 Assert(lt->offsetBlockNumber == 0L);
987
988 /*
989 * Completion of a write phase. Flush last partial data block, and rewind
990 * for nondestructive read.
991 */
992 if (lt->dirty)
993 {
994 /*
995 * As long as we've filled the buffer at least once, its contents are
996 * entirely defined from valgrind's point of view, even though
997 * contents beyond the current end point may be stale. But it's
998 * possible - at least in the case of a parallel sort - to sort such
999 * small amount of data that we do not fill the buffer even once. Tell
1000 * valgrind that its contents are defined, so it doesn't bleat.
1001 */
1003 lt->buffer_size - lt->nbytes);
1004
1007 }
1008 lt->writing = false;
1009 lt->frozen = true;
1010
1011 /*
1012 * The seek and backspace functions assume a single block read buffer.
1013 * That's OK with current usage. A larger buffer is helpful to make the
1014 * read pattern of the backing file look more sequential to the OS, when
1015 * we're reading from multiple tapes. But at the end of a sort, when a
1016 * tape is frozen, we only read from a single tape anyway.
1017 */
1018 if (!lt->buffer || lt->buffer_size != BLCKSZ)
1019 {
1020 if (lt->buffer)
1021 pfree(lt->buffer);
1022 lt->buffer = palloc(BLCKSZ);
1023 lt->buffer_size = BLCKSZ;
1024 }
1025
1026 /* Read the first block, or reset if tape is empty */
1028 lt->pos = 0;
1029 lt->nbytes = 0;
1030
1031 if (lt->firstBlockNumber == -1L)
1032 lt->nextBlockNumber = -1L;
1034 if (TapeBlockIsLast(lt->buffer))
1035 lt->nextBlockNumber = -1L;
1036 else
1038 lt->nbytes = TapeBlockGetNBytes(lt->buffer);
1039
1040 /* Handle extra steps when caller is to share its tapeset */
1041 if (share)
1042 {
1045 }
1046}
void BufFileExportFileSet(BufFile *file)
Definition: buffile.c:394
static void ltsWriteBlock(LogicalTapeSet *lts, int64 blocknum, const void *buffer)
Definition: logtape.c:238
#define TapeBlockGetNBytes(buf)
Definition: logtape.c:108
#define TapeBlockSetNBytes(buf, nbytes)
Definition: logtape.c:111
void * palloc(Size size)
Definition: mcxt.c:1365
#define VALGRIND_MAKE_MEM_DEFINED(addr, size)
Definition: memdebug.h:26
BufFile * pfile
Definition: logtape.c:189
int64 offsetBlockNumber
Definition: logtape.c:159
bool writing
Definition: logtape.c:141
bool dirty
Definition: logtape.c:143
int64 firstblocknumber
Definition: logtape.h:54

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.

610{
611 LogicalTape *lt;
612 int64 tapeblocks;
613 char filename[MAXPGPATH];
614 BufFile *file;
615 int64 filesize;
616
617 lt = ltsCreateTape(lts);
618
619 /*
620 * build concatenated view of all buffiles, remembering the block number
621 * where each source file begins.
622 */
623 pg_itoa(worker, filename);
624 file = BufFileOpenFileSet(&lts->fileset->fs, filename, O_RDONLY, false);
625 filesize = BufFileSize(file);
626
627 /*
628 * Stash first BufFile, and concatenate subsequent BufFiles to that. Store
629 * block offset into each tape as we go.
630 */
632 if (lts->pfile == NULL)
633 {
634 lts->pfile = file;
635 lt->offsetBlockNumber = 0L;
636 }
637 else
638 {
639 lt->offsetBlockNumber = BufFileAppend(lts->pfile, file);
640 }
641 /* Don't allocate more for read buffer than could possibly help */
642 lt->max_size = Min(MaxAllocSize, filesize);
643 tapeblocks = filesize / BLCKSZ;
644
645 /*
646 * Update # of allocated blocks and # blocks written to reflect the
647 * imported BufFile. Allocated/written blocks include space used by holes
648 * left between concatenated BufFiles. Also track the number of hole
649 * blocks so that we can later work backwards to calculate the number of
650 * physical blocks for instrumentation.
651 */
653
654 lts->nBlocksAllocated = lt->offsetBlockNumber + tapeblocks;
656
657 return lt;
658}
BufFile * BufFileOpenFileSet(FileSet *fileset, const char *name, int mode, bool missing_ok)
Definition: buffile.c:291
int64 BufFileSize(BufFile *file)
Definition: buffile.c:866
int64 BufFileAppend(BufFile *target, BufFile *source)
Definition: buffile.c:902
#define Min(x, y)
Definition: c.h:1003
#define MaxAllocSize
Definition: fe_memutils.h:22
int pg_itoa(int16 i, char *a)
Definition: numutils.c:1042
#define MAXPGPATH
static char * filename
Definition: pg_dumpall.c:120
Definition: buffile.c:71
int64 nHoleBlocks
Definition: logtape.c:205
int64 nBlocksAllocated
Definition: logtape.c:203
int64 nBlocksWritten
Definition: logtape.c:204
int max_size
Definition: logtape.c:166
FileSet fs
Definition: sharedfileset.h:28

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.

929{
930 size_t nread = 0;
931 size_t nthistime;
932
933 Assert(!lt->writing);
934
935 if (lt->buffer == NULL)
937
938 while (size > 0)
939 {
940 if (lt->pos >= lt->nbytes)
941 {
942 /* Try to load more data into buffer. */
943 if (!ltsReadFillBuffer(lt))
944 break; /* EOF */
945 }
946
947 nthistime = lt->nbytes - lt->pos;
948 if (nthistime > size)
949 nthistime = size;
950 Assert(nthistime > 0);
951
952 memcpy(ptr, lt->buffer + lt->pos, nthistime);
953
954 lt->pos += nthistime;
955 ptr = (char *) ptr + nthistime;
956 size -= nthistime;
957 nread += nthistime;
958 }
959
960 return nread;
961}
static bool ltsReadFillBuffer(LogicalTape *lt)
Definition: logtape.c:298

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.

847{
848 LogicalTapeSet *lts = lt->tapeSet;
849
850 /*
851 * Round and cap buffer_size if needed.
852 */
853 if (lt->frozen)
854 buffer_size = BLCKSZ;
855 else
856 {
857 /* need at least one block */
858 if (buffer_size < BLCKSZ)
859 buffer_size = BLCKSZ;
860
861 /* palloc() larger than max_size is unlikely to be helpful */
862 if (buffer_size > lt->max_size)
863 buffer_size = lt->max_size;
864
865 /* round down to BLCKSZ boundary */
866 buffer_size -= buffer_size % BLCKSZ;
867 }
868
869 if (lt->writing)
870 {
871 /*
872 * Completion of a write phase. Flush last partial data block, and
873 * rewind for normal (destructive) read.
874 */
875 if (lt->dirty)
876 {
877 /*
878 * As long as we've filled the buffer at least once, its contents
879 * are entirely defined from valgrind's point of view, even though
880 * contents beyond the current end point may be stale. But it's
881 * possible - at least in the case of a parallel sort - to sort
882 * such small amount of data that we do not fill the buffer even
883 * once. Tell valgrind that its contents are defined, so it
884 * doesn't bleat.
885 */
887 lt->buffer_size - lt->nbytes);
888
891 }
892 lt->writing = false;
893 }
894 else
895 {
896 /*
897 * This is only OK if tape is frozen; we rewind for (another) read
898 * pass.
899 */
900 Assert(lt->frozen);
901 }
902
903 if (lt->buffer)
904 pfree(lt->buffer);
905
906 /* the buffer is lazily allocated, but set the size here */
907 lt->buffer = NULL;
908 lt->buffer_size = buffer_size;
909
910 /* free the preallocation list, and return unused block numbers */
911 if (lt->prealloc != NULL)
912 {
913 for (int i = lt->nprealloc; i > 0; i--)
914 ltsReleaseBlock(lts, lt->prealloc[i - 1]);
915 pfree(lt->prealloc);
916 lt->prealloc = NULL;
917 lt->nprealloc = 0;
918 lt->prealloc_size = 0;
919 }
920}
static void ltsReleaseBlock(LogicalTapeSet *lts, int64 blocknum)
Definition: logtape.c:469
int64 * prealloc
Definition: logtape.c:175
int nprealloc
Definition: logtape.c:176
int prealloc_size
Definition: logtape.c:177

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.

1134{
1135 Assert(lt->frozen);
1136 Assert(offset >= 0 && offset <= TapeBlockPayloadSize);
1137 Assert(lt->buffer_size == BLCKSZ);
1138
1139 if (lt->buffer == NULL)
1141
1142 if (blocknum != lt->curBlockNumber)
1143 {
1144 ltsReadBlock(lt->tapeSet, blocknum, lt->buffer);
1145 lt->curBlockNumber = blocknum;
1148 }
1149
1150 if (offset > lt->nbytes)
1151 elog(ERROR, "invalid tape seek position");
1152 lt->pos = offset;
1153}

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.

1182{
1183 return lts->nBlocksWritten - lts->nHoleBlocks;
1184}

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.

668{
669 BufFileClose(lts->pfile);
670 pfree(lts->freeBlocks);
671 pfree(lts);
672}
void BufFileClose(BufFile *file)
Definition: buffile.c:412
int64 * freeBlocks
Definition: logtape.c:216

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.

557{
558 LogicalTapeSet *lts;
559
560 /*
561 * Create top-level struct including per-tape LogicalTape structs.
562 */
563 lts = (LogicalTapeSet *) palloc(sizeof(LogicalTapeSet));
564 lts->nBlocksAllocated = 0L;
565 lts->nBlocksWritten = 0L;
566 lts->nHoleBlocks = 0L;
567 lts->forgetFreeSpace = false;
568 lts->freeBlocksLen = 32; /* reasonable initial guess */
569 lts->freeBlocks = (int64 *) palloc(lts->freeBlocksLen * sizeof(int64));
570 lts->nFreeBlocks = 0;
571 lts->enable_prealloc = preallocate;
572
573 lts->fileset = fileset;
574 lts->worker = worker;
575
576 /*
577 * Create temp BufFile storage as required.
578 *
579 * In leader, we hijack the BufFile of the first tape that's imported, and
580 * concatenate the BufFiles of any subsequent tapes to that. Hence don't
581 * create a BufFile here. Things are simpler for the worker case and the
582 * serial case, though. They are generally very similar -- workers use a
583 * shared fileset, whereas serial sorts use a conventional serial BufFile.
584 */
585 if (fileset && worker == -1)
586 lts->pfile = NULL;
587 else if (fileset)
588 {
589 char filename[MAXPGPATH];
590
591 pg_itoa(worker, filename);
592 lts->pfile = BufFileCreateFileSet(&fileset->fs, filename);
593 }
594 else
595 lts->pfile = BufFileCreateTemp(false);
596
597 return lts;
598}
BufFile * BufFileCreateTemp(bool interXact)
Definition: buffile.c:193
BufFile * BufFileCreateFileSet(FileSet *fileset, const char *name)
Definition: buffile.c:267
bool forgetFreeSpace
Definition: logtape.c:215
Size freeBlocksLen
Definition: logtape.c:218
int64 nFreeBlocks
Definition: logtape.c:217
bool enable_prealloc
Definition: logtape.c:219

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.

751{
752 lts->forgetFreeSpace = true;
753}

References LogicalTapeSet::forgetFreeSpace.

Referenced by mergeruns().

◆ LogicalTapeTell()

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

Definition at line 1162 of file logtape.c.

1163{
1164 if (lt->buffer == NULL)
1166
1167 Assert(lt->offsetBlockNumber == 0L);
1168
1169 /* With a larger buffer, 'pos' wouldn't be the same as offset within page */
1170 Assert(lt->buffer_size == BLCKSZ);
1171
1172 *blocknum = lt->curBlockNumber;
1173 *offset = lt->pos;
1174}

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.

762{
763 LogicalTapeSet *lts = lt->tapeSet;
764 size_t nthistime;
765
766 Assert(lt->writing);
767 Assert(lt->offsetBlockNumber == 0L);
768
769 /* Allocate data buffer and first block on first write */
770 if (lt->buffer == NULL)
771 {
772 lt->buffer = (char *) palloc(BLCKSZ);
773 lt->buffer_size = BLCKSZ;
774 }
775 if (lt->curBlockNumber == -1)
776 {
777 Assert(lt->firstBlockNumber == -1);
778 Assert(lt->pos == 0);
779
780 lt->curBlockNumber = ltsGetBlock(lts, lt);
782
783 TapeBlockGetTrailer(lt->buffer)->prev = -1L;
784 }
785
786 Assert(lt->buffer_size == BLCKSZ);
787 while (size > 0)
788 {
789 if (lt->pos >= (int) TapeBlockPayloadSize)
790 {
791 /* Buffer full, dump it out */
792 int64 nextBlockNumber;
793
794 if (!lt->dirty)
795 {
796 /* Hmm, went directly from reading to writing? */
797 elog(ERROR, "invalid logtape state: should be dirty");
798 }
799
800 /*
801 * First allocate the next block, so that we can store it in the
802 * 'next' pointer of this block.
803 */
804 nextBlockNumber = ltsGetBlock(lt->tapeSet, lt);
805
806 /* set the next-pointer and dump the current block. */
807 TapeBlockGetTrailer(lt->buffer)->next = nextBlockNumber;
809
810 /* initialize the prev-pointer of the next block */
812 lt->curBlockNumber = nextBlockNumber;
813 lt->pos = 0;
814 lt->nbytes = 0;
815 }
816
817 nthistime = TapeBlockPayloadSize - lt->pos;
818 if (nthistime > size)
819 nthistime = size;
820 Assert(nthistime > 0);
821
822 memcpy(lt->buffer + lt->pos, ptr, nthistime);
823
824 lt->dirty = true;
825 lt->pos += nthistime;
826 if (lt->nbytes < lt->pos)
827 lt->nbytes = lt->pos;
828 ptr = (const char *) ptr + nthistime;
829 size -= nthistime;
830 }
831}
static int64 ltsGetBlock(LogicalTapeSet *lts, LogicalTape *lt)
Definition: logtape.c:358

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().

◆ ltsCreateTape()

static LogicalTape * ltsCreateTape ( LogicalTapeSet *  lts )
static

Definition at line 696 of file logtape.c.

697{
698 LogicalTape *lt;
699
700 /*
701 * Create per-tape struct. Note we allocate the I/O buffer lazily.
702 */
703 lt = palloc(sizeof(LogicalTape));
704 lt->tapeSet = lts;
705 lt->writing = true;
706 lt->frozen = false;
707 lt->dirty = false;
708 lt->firstBlockNumber = -1L;
709 lt->curBlockNumber = -1L;
710 lt->nextBlockNumber = -1L;
711 lt->offsetBlockNumber = 0L;
712 lt->buffer = NULL;
713 lt->buffer_size = 0;
714 /* palloc() larger than MaxAllocSize would fail */
716 lt->pos = 0;
717 lt->nbytes = 0;
718 lt->prealloc = NULL;
719 lt->nprealloc = 0;
720 lt->prealloc_size = 0;
721
722 return lt;
723}

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.

359{
360 if (lts->enable_prealloc)
361 return ltsGetPreallocBlock(lts, lt);
362 else
363 return ltsGetFreeBlock(lts);
364}
static int64 ltsGetPreallocBlock(LogicalTapeSet *lts, LogicalTape *lt)
Definition: logtape.c:431
static int64 ltsGetFreeBlock(LogicalTapeSet *lts)
Definition: logtape.c:371

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.

372{
373 int64 *heap = lts->freeBlocks;
374 int64 blocknum;
375 int64 heapsize;
376 int64 holeval;
377 uint64 holepos;
378
379 /* freelist empty; allocate a new block */
380 if (lts->nFreeBlocks == 0)
381 return lts->nBlocksAllocated++;
382
383 /* easy if heap contains one element */
384 if (lts->nFreeBlocks == 1)
385 {
386 lts->nFreeBlocks--;
387 return lts->freeBlocks[0];
388 }
389
390 /* remove top of minheap */
391 blocknum = heap[0];
392
393 /* we'll replace it with end of minheap array */
394 holeval = heap[--lts->nFreeBlocks];
395
396 /* sift down */
397 holepos = 0; /* holepos is where the "hole" is */
398 heapsize = lts->nFreeBlocks;
399 while (true)
400 {
401 uint64 left = left_offset(holepos);
402 uint64 right = right_offset(holepos);
403 uint64 min_child;
404
405 if (left < heapsize && right < heapsize)
406 min_child = (heap[left] < heap[right]) ? left : right;
407 else if (left < heapsize)
408 min_child = left;
409 else if (right < heapsize)
410 min_child = right;
411 else
412 break;
413
414 if (heap[min_child] >= holeval)
415 break;
416
417 heap[holepos] = heap[min_child];
418 holepos = min_child;
419 }
420 heap[holepos] = holeval;
421
422 return blocknum;
423}
uint64_t uint64
Definition: c.h:539
static uint64 left_offset(uint64 i)
Definition: logtape.c:337
static uint64 right_offset(uint64 i)
Definition: logtape.c:343

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.

432{
433 /* sorted in descending order, so return the last element */
434 if (lt->nprealloc > 0)
435 return lt->prealloc[--lt->nprealloc];
436
437 if (lt->prealloc == NULL)
438 {
440 lt->prealloc = (int64 *) palloc(sizeof(int64) * lt->prealloc_size);
441 }
443 {
444 /* when the preallocation list runs out, double the size */
445 lt->prealloc_size *= 2;
448 lt->prealloc = (int64 *) repalloc(lt->prealloc,
449 sizeof(int64) * lt->prealloc_size);
450 }
451
452 /* refill preallocation list */
453 lt->nprealloc = lt->prealloc_size;
454 for (int i = lt->nprealloc; i > 0; i--)
455 {
456 lt->prealloc[i - 1] = ltsGetFreeBlock(lts);
457
458 /* verify descending order */
459 Assert(i == lt->nprealloc || lt->prealloc[i - 1] > lt->prealloc[i]);
460 }
461
462 return lt->prealloc[--lt->nprealloc];
463}
#define TAPE_WRITE_PREALLOC_MIN
Definition: logtape.c:124
#define TAPE_WRITE_PREALLOC_MAX
Definition: logtape.c:125
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1610

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.

523{
524 Assert(lt->buffer_size > 0);
525 lt->buffer = palloc(lt->buffer_size);
526
527 /* Read the first block, or reset if tape is empty */
529 lt->pos = 0;
530 lt->nbytes = 0;
532}

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.

283{
284 if (BufFileSeekBlock(lts->pfile, blocknum) != 0)
287 errmsg("could not seek to block %" PRId64 " of temporary file",
288 blocknum)));
289 BufFileReadExact(lts->pfile, buffer, BLCKSZ);
290}
int BufFileSeekBlock(BufFile *file, int64 blknum)
Definition: buffile.c:851
void BufFileReadExact(BufFile *file, void *ptr, size_t size)
Definition: buffile.c:654
int errcode_for_file_access(void)
Definition: elog.c:877
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ereport(elevel,...)
Definition: elog.h:150

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.

299{
300 lt->pos = 0;
301 lt->nbytes = 0;
302
303 do
304 {
305 char *thisbuf = lt->buffer + lt->nbytes;
306 int64 datablocknum = lt->nextBlockNumber;
307
308 /* Fetch next block number */
309 if (datablocknum == -1L)
310 break; /* EOF */
311 /* Apply worker offset, needed for leader tapesets */
312 datablocknum += lt->offsetBlockNumber;
313
314 /* Read the block */
315 ltsReadBlock(lt->tapeSet, datablocknum, thisbuf);
316 if (!lt->frozen)
317 ltsReleaseBlock(lt->tapeSet, datablocknum);
319
320 lt->nbytes += TapeBlockGetNBytes(thisbuf);
321 if (TapeBlockIsLast(thisbuf))
322 {
323 lt->nextBlockNumber = -1L;
324 /* EOF */
325 break;
326 }
327 else
328 lt->nextBlockNumber = TapeBlockGetTrailer(thisbuf)->next;
329
330 /* Advance to next block, if we have buffer space left */
331 } while (lt->buffer_size - lt->nbytes > BLCKSZ);
332
333 return (lt->nbytes > 0);
334}

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.

470{
471 int64 *heap;
472 uint64 holepos;
473
474 /*
475 * Do nothing if we're no longer interested in remembering free space.
476 */
477 if (lts->forgetFreeSpace)
478 return;
479
480 /*
481 * Enlarge freeBlocks array if full.
482 */
483 if (lts->nFreeBlocks >= lts->freeBlocksLen)
484 {
485 /*
486 * If the freelist becomes very large, just return and leak this free
487 * block.
488 */
489 if (lts->freeBlocksLen * 2 * sizeof(int64) > MaxAllocSize)
490 return;
491
492 lts->freeBlocksLen *= 2;
493 lts->freeBlocks = (int64 *) repalloc(lts->freeBlocks,
494 lts->freeBlocksLen * sizeof(int64));
495 }
496
497 /* create a "hole" at end of minheap array */
498 heap = lts->freeBlocks;
499 holepos = lts->nFreeBlocks;
500 lts->nFreeBlocks++;
501
502 /* sift up to insert blocknum */
503 while (holepos != 0)
504 {
505 uint64 parent = parent_offset(holepos);
506
507 if (heap[parent] < blocknum)
508 break;
509
510 heap[holepos] = heap[parent];
511 holepos = parent;
512 }
513 heap[holepos] = blocknum;
514}
static uint64 parent_offset(uint64 i)
Definition: logtape.c:349

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.

239{
240 /*
241 * BufFile does not support "holes", so if we're about to write a block
242 * that's past the current end of file, fill the space between the current
243 * end of file and the target block with zeros.
244 *
245 * This can happen either when tapes preallocate blocks; or for the last
246 * block of a tape which might not have been flushed.
247 *
248 * Note that BufFile concatenation can leave "holes" in BufFile between
249 * worker-owned block ranges. These are tracked for reporting purposes
250 * only. We never read from nor write to these hole blocks, and so they
251 * are not considered here.
252 */
253 while (blocknum > lts->nBlocksWritten)
254 {
255 PGIOAlignedBlock zerobuf;
256
257 MemSet(zerobuf.data, 0, sizeof(zerobuf));
258
259 ltsWriteBlock(lts, lts->nBlocksWritten, zerobuf.data);
260 }
261
262 /* Write the requested block */
263 if (BufFileSeekBlock(lts->pfile, blocknum) != 0)
266 errmsg("could not seek to block %" PRId64 " of temporary file",
267 blocknum)));
268 BufFileWrite(lts->pfile, buffer, BLCKSZ);
269
270 /* Update nBlocksWritten, if we extended the file */
271 if (blocknum == lts->nBlocksWritten)
272 lts->nBlocksWritten++;
273}
void BufFileWrite(BufFile *file, const void *ptr, size_t size)
Definition: buffile.c:676
#define MemSet(start, val, len)
Definition: c.h:1019
char data[BLCKSZ]
Definition: c.h:1136

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

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

◆ parent_offset()

static uint64 parent_offset ( uint64  i )
inlinestatic

Definition at line 349 of file logtape.c.

350{
351 return (i - 1) / 2;
352}

References i.

Referenced by ltsReleaseBlock().

◆ right_offset()

static uint64 right_offset ( uint64  i )
inlinestatic

Definition at line 343 of file logtape.c.

344{
345 return 2 * i + 2;
346}

References i.

Referenced by ltsGetFreeBlock().

AltStyle によって変換されたページ (->オリジナル) /