The beauty of the polyphase merge algorithm is that it allows reusing
input tapes as output tapes efficiently. So if you have N tape drives,
you can keep them all busy throughout the merge.
That doesn't matter, when we can easily have as many "tape drives" as we
want. In PostgreSQL, a tape drive consumes just a few kB of memory, for
the buffers. With the patch being discussed to allow a tape to be
"paused" between write passes [1], we don't even keep the tape buffers
around, when a tape is not actively read written to, so all it consumes
is the memory needed for the LogicalTape struct.
The number of *input* tapes we can use in each merge pass is still
limited, by the memory needed for the tape buffers and the merge heap,
but only one output tape is active at any time. The inactive output
tapes consume very few resources. So the whole idea of trying to
efficiently reuse input tapes as output tapes is pointless.
Let's switch over to a simple k-way balanced merge. Because it's
simpler. If you're severely limited on memory, like when sorting 1GB of
data with work_mem='1MB' or less, it's also slightly faster. I'm not too
excited about the performance aspect, because in the typical case of a
single-pass merge, there's no difference. But it would be worth changing
on simplicity grounds, since we're mucking around in tuplesort.c anyway.
I came up with the attached patch to do that. This patch applies on top
of my latest "Logical tape pause/resume" patches [1]. It includes
changes to the logtape.c interface, that make it possible to create and
destroy LogicalTapes in a tapeset on the fly. I believe this will also
come handy for Peter's parallel tuplesort patch set.
[1] Logical tape pause/resume,
https://www.postgresql.org/message-id/55b3b7ae-8dec-b188-b8eb-e07604052351%40iki.fi
PS. I finally bit the bullet and got self a copy of The Art of Computer
Programming, Vol 3, 2nd edition. In section 5.4 on External Sorting,
Knuth says:
"
When this book was first written, magnetic tapes were abundant and disk
drives were expensive. But disks became enormously better during the
1980s, and by the late 1990s they had almost completely replaced
magnetic tape units on most of the world's computer systems. Therefore
the once-crucial topic of tape merging has become of limited relevance
to current needs.
Yet many of the patterns are quite beautiful, and the associated
algorithms reflect some of the best research done in computer science
during its early days; the techniques are just too nice to be discarded
abruptly onto the rubbish heap of history. Indeed, the ways in which
these methods blend theory with practice are especially instructive.
Therefore merging patterns are discussed carefully and completely below,
in what may be their last grand appearance before they accept the final
curtain call.
"
Yep, the polyphase and other merge patterns are beautiful. I enjoyed
reading through those sections. Now let's put them to rest in PostgreSQL.
- Heikki
>From 15169f32f99a69401d3565c8d0ff0d532d6b6638 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <heikki.linnakan...@iki.fi>
Date: Wed, 12 Oct 2016 20:04:17 +0300
Subject: [PATCH 1/1] Replace polyphase merge algorithm with a simple balanced
k-way merge.
The advantage of polyphase merge is that it can reuse the input tapes as
output tapes efficiently, but that is irrelevant on modern hardware, when
we can easily emulate any number of tape drives. The number of input tapes
we can/should use during merging is limited by work_mem, but output tapes
that we are not currently writing to only cost a little bit of memory, so
there is no need to skimp on them.
Refactor LogicalTapeSet/LogicalTape interface. All the tape functions,
like LogicalTapeRead and LogicalTapeWrite, take a LogicalTape as argument,
instead of LogicalTapeSet+tape number. You can create any number of
LogicalTapes in a single LogicalTapeSet, and you don't need to decide the
number upfront, when you create the tape set.
---
src/backend/utils/sort/logtape.c | 223 +++++--------
src/backend/utils/sort/tuplesort.c | 665 +++++++++++++++----------------------
src/include/utils/logtape.h | 37 ++-
3 files changed, 366 insertions(+), 559 deletions(-)
diff --git a/src/backend/utils/sort/logtape.c b/src/backend/utils/sort/logtape.c
index 1f540f0..2370fa5 100644
--- a/src/backend/utils/sort/logtape.c
+++ b/src/backend/utils/sort/logtape.c
@@ -118,6 +118,8 @@ typedef struct TapeBlockTrailer
*/
typedef struct LogicalTape
{
+ LogicalTapeSet *tapeSet; /* tape set this tape is part of */
+
bool writing; /* T while in write phase */
bool paused; /* T if the tape is paused */
bool frozen; /* T if blocks should not be freed when read */
@@ -173,10 +175,6 @@ struct LogicalTapeSet
long *freeBlocks; /* resizable array */
int nFreeBlocks; /* # of currently free blocks */
int freeBlocksLen; /* current allocated length of freeBlocks[] */
-
- /* The array of logical tapes. */
- int nTapes; /* # of logical tapes in set */
- LogicalTape tapes[FLEXIBLE_ARRAY_MEMBER]; /* has nTapes nentries */
};
static void ltsWriteBlock(LogicalTapeSet *lts, long blocknum, void *buffer);
@@ -228,7 +226,7 @@ ltsReadBlock(LogicalTapeSet *lts, long blocknum, void *buffer)
* Returns true if anything was read, 'false' on EOF.
*/
static bool
-ltsReadFillBuffer(LogicalTapeSet *lts, LogicalTape *lt)
+ltsReadFillBuffer(LogicalTape *lt)
{
lt->pos = 0;
lt->nbytes = 0;
@@ -242,9 +240,9 @@ ltsReadFillBuffer(LogicalTapeSet *lts, LogicalTape *lt)
break; /* EOF */
/* Read the block */
- ltsReadBlock(lts, lt->nextBlockNumber, (void *) thisbuf);
+ ltsReadBlock(lt->tapeSet, lt->nextBlockNumber, (void *) thisbuf);
if (!lt->frozen)
- ltsReleaseBlock(lts, lt->nextBlockNumber);
+ ltsReleaseBlock(lt->tapeSet, lt->nextBlockNumber);
lt->curBlockNumber = lt->nextBlockNumber;
lt->nbytes += TapeBlockGetNBytes(thisbuf);
@@ -348,18 +346,14 @@ ltsReleaseBlock(LogicalTapeSet *lts, long blocknum)
* Each tape is initialized in write state.
*/
LogicalTapeSet *
-LogicalTapeSetCreate(int ntapes)
+LogicalTapeSetCreate(void)
{
LogicalTapeSet *lts;
- LogicalTape *lt;
- int i;
/*
* Create top-level struct including per-tape LogicalTape structs.
*/
- Assert(ntapes > 0);
- lts = (LogicalTapeSet *) palloc(offsetof(LogicalTapeSet, tapes) +
- ntapes * sizeof(LogicalTape));
+ lts = (LogicalTapeSet *) palloc(sizeof(LogicalTapeSet));
lts->pfile = BufFileCreateTemp(false);
lts->nFileBlocks = 0L;
lts->forgetFreeSpace = false;
@@ -367,52 +361,71 @@ LogicalTapeSetCreate(int ntapes)
lts->freeBlocksLen = 32; /* reasonable initial guess */
lts->freeBlocks = (long *) palloc(lts->freeBlocksLen * sizeof(long));
lts->nFreeBlocks = 0;
- lts->nTapes = ntapes;
- /*
- * Initialize per-tape structs. Note we allocate the I/O buffer and the
- * first block for a tape only when it is first actually written to. This
- * avoids wasting memory space when tuplesort.c overestimates the number
- * of tapes needed.
- */
- for (i = 0; i < ntapes; i++)
- {
- lt = <s->tapes[i];
- lt->writing = true;
- lt->frozen = false;
- lt->dirty = false;
- lt->paused = false;
- lt->firstBlockNumber = -1L;
- lt->curBlockNumber = -1L;
- lt->buffer = NULL;
- lt->buffer_size = 0;
- lt->pos = 0;
- lt->nbytes = 0;
- }
return lts;
}
/*
* 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)
{
- LogicalTape *lt;
- int i;
-
BufFileClose(lts->pfile);
- for (i = 0; i < lts->nTapes; i++)
- {
- lt = <s->tapes[i];
- if (lt->buffer)
- pfree(lt->buffer);
- }
pfree(lts->freeBlocks);
pfree(lts);
}
/*
+ * Create a logical tape in the given tapeset.
+ */
+LogicalTape *
+LogicalTapeCreate(LogicalTapeSet *lts)
+{
+ LogicalTape *lt;
+
+ lt = palloc(sizeof(LogicalTape));
+
+ /*
+ * Initialize per-tape structs. Note we allocate the I/O buffer and the
+ * first block for a tape only when it is first actually written to. This
+ * avoids wasting memory space when tuplesort.c overestimates the number
+ * of tapes needed. (XXX: that doesn't really happen anymore).
+ */
+ lt->tapeSet = lts;
+ lt->writing = true;
+ lt->frozen = false;
+ lt->dirty = false;
+ lt->paused = false;
+ lt->firstBlockNumber = -1L;
+ lt->curBlockNumber = -1L;
+ lt->buffer = NULL;
+ lt->buffer_size = 0;
+ lt->pos = 0;
+ lt->nbytes = 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
@@ -428,19 +441,24 @@ LogicalTapeSetForgetFreeSpace(LogicalTapeSet *lts)
}
/*
+ * Obtain total disk space currently used by a LogicalTapeSet, in blocks.
+ */
+long
+LogicalTapeSetBlocks(LogicalTapeSet *lts)
+{
+ return lts->nFileBlocks;
+}
+
+/*
* Write to a logical tape.
*
* There are no error returns; we ereport() on failure.
*/
void
-LogicalTapeWrite(LogicalTapeSet *lts, int tapenum,
- void *ptr, size_t size)
+LogicalTapeWrite(LogicalTape *lt, void *ptr, size_t size)
{
- LogicalTape *lt;
size_t nthistime;
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
Assert(lt->writing);
/* Allocate data buffer and first block on first write */
@@ -452,7 +470,7 @@ LogicalTapeWrite(LogicalTapeSet *lts, int tapenum,
/* if the tape was paused, resume it now. */
if (lt->paused)
{
- ltsReadBlock(lts, lt->curBlockNumber, (void *) lt->buffer);
+ ltsReadBlock(lt->tapeSet, lt->curBlockNumber, (void *) lt->buffer);
/* 'pos' and 'nbytes' should still be valid */
Assert(lt->nbytes == TapeBlockGetNBytes(lt->buffer));
@@ -470,7 +488,7 @@ LogicalTapeWrite(LogicalTapeSet *lts, int tapenum,
Assert(lt->firstBlockNumber == -1);
Assert(lt->pos == 0);
- lt->curBlockNumber = ltsGetFreeBlock(lts);
+ lt->curBlockNumber = ltsGetFreeBlock(lt->tapeSet);
lt->firstBlockNumber = lt->curBlockNumber;
TapeBlockGetTrailer(lt->buffer)->prev = -1L;
@@ -488,11 +506,11 @@ LogicalTapeWrite(LogicalTapeSet *lts, int tapenum,
* First allocate the next block, so that we can store it in the
* 'next' pointer of this block.
*/
- nextBlockNumber = ltsGetFreeBlock(lts);
+ nextBlockNumber = ltsGetFreeBlock(lt->tapeSet);
/* set the next-pointer and dump the current block. */
TapeBlockGetTrailer(lt->buffer)->next = nextBlockNumber;
- ltsWriteBlock(lts, lt->curBlockNumber, (void *) lt->buffer);
+ ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, (void *) lt->buffer);
/* initialize the prev-pointer of the next block */
TapeBlockGetTrailer(lt->buffer)->prev = lt->curBlockNumber;
@@ -530,13 +548,8 @@ LogicalTapeWrite(LogicalTapeSet *lts, int tapenum,
* byte buffer is used.
*/
void
-LogicalTapeRewindForRead(LogicalTapeSet *lts, int tapenum, size_t buffer_size)
+LogicalTapeRewindForRead(LogicalTape *lt, size_t buffer_size)
{
- LogicalTape *lt;
-
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
-
/*
* Round and cap buffer_size if needed.
*/
@@ -578,7 +591,7 @@ LogicalTapeRewindForRead(LogicalTapeSet *lts, int tapenum, size_t buffer_size)
if (lt->dirty)
{
TapeBlockSetNBytes(lt->buffer, lt->nbytes);
- ltsWriteBlock(lts, lt->curBlockNumber, (void *) lt->buffer);
+ ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, (void *) lt->buffer);
}
}
else
@@ -606,36 +619,7 @@ LogicalTapeRewindForRead(LogicalTapeSet *lts, int tapenum, size_t buffer_size)
lt->nextBlockNumber = lt->firstBlockNumber;
lt->pos = 0;
lt->nbytes = 0;
- ltsReadFillBuffer(lts, lt);
-}
-
-/*
- * Rewind logical tape and switch from reading to writing.
- *
- * NOTE: we assume the caller has read the tape to the end; otherwise
- * untouched data and indirect blocks will not have been freed. We
- * could add more code to free any unread blocks, but in current usage
- * of this module it'd be useless code.
- */
-void
-LogicalTapeRewindForWrite(LogicalTapeSet *lts, int tapenum)
-{
- LogicalTape *lt;
-
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
-
- Assert(!lt->writing && !lt->frozen);
- lt->writing = true;
- lt->dirty = false;
- lt->firstBlockNumber = -1L;
- lt->curBlockNumber = -1L;
- lt->pos = 0;
- lt->nbytes = 0;
- if (lt->buffer)
- pfree(lt->buffer);
- lt->buffer = NULL;
- lt->buffer_size = 0;
+ ltsReadFillBuffer(lt);
}
/*
@@ -647,18 +631,13 @@ LogicalTapeRewindForWrite(LogicalTapeSet *lts, int tapenum)
* and the last partial block is read back into memory.
*/
void
-LogicalTapePause(LogicalTapeSet *lts, int tapenum)
+LogicalTapePause(LogicalTape *lt)
{
- LogicalTape *lt;
-
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
-
Assert(lt->writing);
/* Flush last partial data block. */
TapeBlockSetNBytes(lt->buffer, lt->nbytes);
- ltsWriteBlock(lts, lt->curBlockNumber, (void *) lt->buffer);
+ ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, (void *) lt->buffer);
lt->dirty = false;
pfree(lt->buffer);
@@ -674,15 +653,11 @@ LogicalTapePause(LogicalTapeSet *lts, int tapenum)
* Early EOF is indicated by return value less than #bytes requested.
*/
size_t
-LogicalTapeRead(LogicalTapeSet *lts, int tapenum,
- void *ptr, size_t size)
+LogicalTapeRead(LogicalTape *lt, void *ptr, size_t size)
{
- LogicalTape *lt;
size_t nread = 0;
size_t nthistime;
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
Assert(!lt->writing);
while (size > 0)
@@ -690,7 +665,7 @@ LogicalTapeRead(LogicalTapeSet *lts, int tapenum,
if (lt->pos >= lt->nbytes)
{
/* Try to load more data into buffer. */
- if (!ltsReadFillBuffer(lts, lt))
+ if (!ltsReadFillBuffer(lt))
break; /* EOF */
}
@@ -722,12 +697,8 @@ LogicalTapeRead(LogicalTapeSet *lts, int tapenum,
* for-read call is OK but not necessary.
*/
void
-LogicalTapeFreeze(LogicalTapeSet *lts, int tapenum)
+LogicalTapeFreeze(LogicalTape *lt)
{
- LogicalTape *lt;
-
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
Assert(lt->writing);
/*
@@ -737,8 +708,7 @@ LogicalTapeFreeze(LogicalTapeSet *lts, int tapenum)
if (lt->dirty)
{
TapeBlockSetNBytes(lt->buffer, lt->nbytes);
- ltsWriteBlock(lts, lt->curBlockNumber, (void *) lt->buffer);
- lt->writing = false;
+ ltsWriteBlock(lt->tapeSet, lt->curBlockNumber, (void *) lt->buffer);
}
lt->writing = false;
lt->paused = false;
@@ -766,7 +736,7 @@ LogicalTapeFreeze(LogicalTapeSet *lts, int tapenum)
if (lt->firstBlockNumber == -1L)
lt->nextBlockNumber = -1L;
- ltsReadBlock(lts, lt->curBlockNumber, (void *) lt->buffer);
+ ltsReadBlock(lt->tapeSet, lt->curBlockNumber, (void *) lt->buffer);
if (TapeBlockIsLast(lt->buffer))
lt->nextBlockNumber = -1L;
else
@@ -788,13 +758,10 @@ LogicalTapeFreeze(LogicalTapeSet *lts, int tapenum)
* that case.
*/
size_t
-LogicalTapeBackspace(LogicalTapeSet *lts, int tapenum, size_t size)
+LogicalTapeBackspace(LogicalTape *lt, size_t size)
{
- LogicalTape *lt;
size_t seeked = 0;
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
Assert(lt->frozen);
Assert(lt->buffer_size == BLCKSZ);
@@ -826,7 +793,7 @@ LogicalTapeBackspace(LogicalTapeSet *lts, int tapenum, size_t size)
return seeked;
}
- ltsReadBlock(lts, prev, (void *) lt->buffer);
+ ltsReadBlock(lt->tapeSet, prev, (void *) lt->buffer);
if (TapeBlockGetTrailer(lt->buffer)->next != lt->curBlockNumber)
elog(ERROR, "broken tape, next of block %ld is %ld, expected %ld",
@@ -859,20 +826,15 @@ LogicalTapeBackspace(LogicalTapeSet *lts, int tapenum, size_t size)
* LogicalTapeTell().
*/
void
-LogicalTapeSeek(LogicalTapeSet *lts, int tapenum,
- long blocknum, int offset)
+LogicalTapeSeek(LogicalTape *lt, long blocknum, int offset)
{
- LogicalTape *lt;
-
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
Assert(lt->frozen);
Assert(offset >= 0 && offset <= TapeBlockPayloadSize);
Assert(lt->buffer_size == BLCKSZ);
if (blocknum != lt->curBlockNumber)
{
- ltsReadBlock(lts, blocknum, (void *) lt->buffer);
+ ltsReadBlock(lt->tapeSet, blocknum, (void *) lt->buffer);
lt->curBlockNumber = blocknum;
lt->nbytes = TapeBlockPayloadSize;
lt->nextBlockNumber = TapeBlockGetTrailer(lt->buffer)->next;
@@ -890,26 +852,11 @@ LogicalTapeSeek(LogicalTapeSet *lts, int tapenum,
* the position for a seek after freezing. Not clear if anyone needs that.
*/
void
-LogicalTapeTell(LogicalTapeSet *lts, int tapenum,
- long *blocknum, int *offset)
+LogicalTapeTell(LogicalTape *lt, long *blocknum, int *offset)
{
- LogicalTape *lt;
-
- Assert(tapenum >= 0 && tapenum < lts->nTapes);
- lt = <s->tapes[tapenum];
-
/* 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.
- */
-long
-LogicalTapeSetBlocks(LogicalTapeSet *lts)
-{
- return lts->nFileBlocks;
-}
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index eebfb3d..fd4c4a8 100644
--- a/src/backend/utils/sort/tuplesort.c
+++ b/src/backend/utils/sort/tuplesort.c
@@ -10,16 +10,22 @@
* amounts are sorted using temporary files and a standard external sort
* algorithm.
*
- * See Knuth, volume 3, for more than you want to know about the external
- * sorting algorithm. Historically, we divided the input into sorted runs
- * using replacement selection, in the form of a priority tree implemented
- * as a heap (essentially his Algorithm 5.2.3H), but now we only do that
- * for the first run, and only if the run would otherwise end up being very
- * short. We merge the runs using polyphase merge, Knuth's Algorithm
- * 5.4.2D. The logical "tapes" used by Algorithm D are implemented by
- * logtape.c, which avoids space wastage by recycling disk space as soon
- * as each block is read from its "tape".
+ * See Knuth, volume 3, for more than you want to know about external
+ * sorting algorithms. The algorithm we use is a balanced k-way merge.
+ * Before PostgreSQL 10, we used the polyphase merge algorithm (Knuth's
+ * Algorithm 5.4.2D), but with modern hardware, a straightforward
+ * balanced merge is better. Knuth is assuming that tape drives are
+ * expensive beasts, and in particular that there will always be many more
+ * runs than tape drives. The polyphase merge algorithm was good at keeping
+ * all the tape drives busy, but in our implementation a "tape drive"
+ * doesn't cost much more than a few Kb of memory buffers, so we can afford
+ * to have lots of them. In particular, if we can have as many tape drives
+ * as sorted runs, we can eliminate any repeated I/O at all.
*
+ * Historically, we divided the input into sorted runs using replacement
+ * selection, in the form of a priority tree implemented as a heap
+ * (essentially Knuth's Algorithm 5.2.3H), but now we only do that for the
+ * first run, and only if the run would otherwise end up being very short.
* We do not use Knuth's recommended data structure (Algorithm 5.4.1R) for
* the replacement selection, because it uses a fixed number of records
* in memory at all times. Since we are dealing with tuples that may vary
@@ -63,10 +69,9 @@
* tuples just by scanning the tuple array sequentially. If we do exceed
* workMem, we begin to emit tuples into sorted runs in temporary tapes.
* When tuples are dumped in batch after quicksorting, we begin a new run
- * with a new output tape (selected per Algorithm D). After the end of the
- * input is reached, we dump out remaining tuples in memory into a final run
- * (or two, when replacement selection is still used), then merge the runs
- * using Algorithm D.
+ * with a new output tape. After the end of the input is reached, we dump
+ * out remaining tuples in memory into a final run (or two, when replacement
+ * selection is still used), then merge the runs.
*
* When merging runs, we use a heap containing just the frontmost tuple from
* each source run; we repeatedly output the smallest tuple and replace it
@@ -89,6 +94,14 @@
* accesses. The pre-reading is handled by logtape.c, we just tell it how
* much memory to use for the buffers.
*
+ * In the current code we determine the number of input tapes M on the basis
+ * of workMem: we want workMem/M to be large enough that we read a fair
+ * amount of data each time we read from a tape, so as to maintain the
+ * locality of access described above. Nonetheless, with large workMem we
+ * can have many tapes. The logical "tapes" are implemented by logtape.c,
+ * which avoids space wastage by recycling disk space as soon as each block
+ * is read from its "tape".
+ *
* When the caller requests random access to the sort result, we form
* the final sorted run on a logical tape which is then "frozen", so
* that we can access it randomly. When the caller does not need random
@@ -97,19 +110,6 @@
* on-the-fly as the caller repeatedly calls tuplesort_getXXX; this
* saves one cycle of writing all the data out to disk and reading it in.
*
- * Before Postgres 8.2, we always used a seven-tape polyphase merge, on the
- * grounds that 7 is the "sweet spot" on the tapes-to-passes curve according
- * to Knuth's figure 70 (section 5.4.2). However, Knuth is assuming that
- * tape drives are expensive beasts, and in particular that there will always
- * be many more runs than tape drives. In our implementation a "tape drive"
- * doesn't cost much more than a few Kb of memory buffers, so we can afford
- * to have lots of them. In particular, if we can have as many tape drives
- * as sorted runs, we can eliminate any repeated I/O at all. In the current
- * code we determine the number of tapes M on the basis of workMem: we want
- * workMem/M to be large enough that we read a fair amount of data each time
- * we preread from a tape, so as to maintain the locality of access described
- * above. Nonetheless, with large workMem we can have many tapes.
- *
*
* Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
@@ -277,8 +277,7 @@ struct Tuplesortstate
bool tuples; /* Can SortTuple.tuple ever be set? */
int64 availMem; /* remaining memory available, in bytes */
int64 allowedMem; /* total memory allowed, in bytes */
- int maxTapes; /* number of tapes (Knuth's T) */
- int tapeRange; /* maxTapes-1 (Knuth's P) */
+ int maxInputTapes; /* max number of input tapes */
MemoryContext sortcontext; /* memory context holding most sort data */
MemoryContext tuplecontext; /* sub-context of sortcontext for tuple data */
LogicalTapeSet *tapeset; /* logtape.c object for tapes in a temp file */
@@ -310,7 +309,7 @@ struct Tuplesortstate
* SortTuple struct!), and increase state->availMem by the amount of
* memory space thereby released.
*/
- void (*writetup) (Tuplesortstate *state, int tapenum,
+ void (*writetup) (Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
/*
@@ -319,7 +318,7 @@ struct Tuplesortstate
* from the slab memory arena, or is palloc'd, see readtup_alloc().
*/
void (*readtup) (Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len);
+ LogicalTape *tape, unsigned int len);
/*
* This array holds the tuples now in sort memory. If we are in state
@@ -366,8 +365,8 @@ struct Tuplesortstate
char *slabMemoryEnd; /* end of slab memory arena */
SlabSlot *slabFreeHead; /* head of free list */
- /* Buffer size to use for reading input tapes, during merge. */
- size_t read_buffer_size;
+ /* Memory to use for input tape buffers, during merge. */
+ size_t read_buffer_mem;
/*
* When we return a tuple to the caller in tuplesort_gettuple_XXX, that
@@ -392,36 +391,30 @@ struct Tuplesortstate
int currentRun;
/*
- * Unless otherwise noted, all pointer variables below are pointers to
- * arrays of length maxTapes, holding per-tape data.
+ * Logical tapes.
+ *
+ * The initial runs are written in the output tapes. In each merge pass,
+ * the output tapes of the previous pass become the input tapes, and
+ * new output tapes are allocated as needed. If the number of input runs
+ * is equal to the number of input tapes, there is only one merge pass
+ * left.
*/
+ LogicalTape **inputTapes;
+ int nInputTapes;
+ int nInputRuns;
- /*
- * This variable is only used during merge passes. mergeactive[i] is true
- * if we are reading an input run from (actual) tape number i and have not
- * yet exhausted that run.
- */
- bool *mergeactive; /* active input run source? */
+ LogicalTape **outputTapes;
+ int nOutputTapes;
+ int nOutputRuns;
- /*
- * Variables for Algorithm D. Note that destTape is a "logical" tape
- * number, ie, an index into the tp_xxx[] arrays. Be careful to keep
- * "logical" and "actual" tape numbers straight!
- */
- int Level; /* Knuth's l */
- int destTape; /* current output tape (Knuth's j, less 1) */
- int *tp_fib; /* Target Fibonacci run counts (A[]) */
- int *tp_runs; /* # of real runs on each tape */
- int *tp_dummy; /* # of dummy runs for each tape (D[]) */
- int *tp_tapenum; /* Actual tape numbers (TAPE[]) */
- int activeTapes; /* # of active input tapes in merge pass */
+ LogicalTape *destTape; /* current output tape */
/*
* These variables are used after completion of sorting to keep track of
* the next tuple to return. (In the tape case, the tape's current read
* position is also critical state.)
*/
- int result_tape; /* actual tape number of finished output */
+ LogicalTape *result_tape; /* actual tape of finished output */
int current; /* array index (only used if SORTEDINMEM) */
bool eof_reached; /* reached EOF (needed for cursors) */
@@ -568,9 +561,9 @@ struct Tuplesortstate
*/
/* When using this macro, beware of double evaluation of len */
-#define LogicalTapeReadExact(tapeset, tapenum, ptr, len) \
+#define LogicalTapeReadExact(tape, ptr, len) \
do { \
- if (LogicalTapeRead(tapeset, tapenum, ptr, len) != (size_t) (len)) \
+ if (LogicalTapeRead(tape, ptr, len) != (size_t) (len)) \
elog(ERROR, "unexpected end of data"); \
} while(0)
@@ -585,7 +578,7 @@ static void init_slab_allocator(Tuplesortstate *state, int numSlots);
static void mergeruns(Tuplesortstate *state);
static void mergeonerun(Tuplesortstate *state);
static void beginmerge(Tuplesortstate *state);
-static bool mergereadnext(Tuplesortstate *state, int srcTape, SortTuple *stup);
+static bool mergereadnext(Tuplesortstate *state, LogicalTape *srcTape, SortTuple *stup);
static void dumptuples(Tuplesortstate *state, bool alltuples);
static void dumpbatch(Tuplesortstate *state, bool alltuples);
static void make_bounded_heap(Tuplesortstate *state);
@@ -597,39 +590,39 @@ static void tuplesort_heap_replace_top(Tuplesortstate *state, SortTuple *tuple,
bool checkIndex);
static void tuplesort_heap_delete_top(Tuplesortstate *state, bool checkIndex);
static void reversedirection(Tuplesortstate *state);
-static unsigned int getlen(Tuplesortstate *state, int tapenum, bool eofOK);
-static void markrunend(Tuplesortstate *state, int tapenum);
+static unsigned int getlen(Tuplesortstate *state, LogicalTape *tape, bool eofOK);
+static void markrunend(Tuplesortstate *state, LogicalTape *tape);
static void *readtup_alloc(Tuplesortstate *state, Size tuplen);
static int comparetup_heap(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static void copytup_heap(Tuplesortstate *state, SortTuple *stup, void *tup);
-static void writetup_heap(Tuplesortstate *state, int tapenum,
+static void writetup_heap(Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
static void readtup_heap(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len);
+ LogicalTape *tape, unsigned int len);
static int comparetup_cluster(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static void copytup_cluster(Tuplesortstate *state, SortTuple *stup, void *tup);
-static void writetup_cluster(Tuplesortstate *state, int tapenum,
+static void writetup_cluster(Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
static void readtup_cluster(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len);
+ LogicalTape *tape, unsigned int len);
static int comparetup_index_btree(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static int comparetup_index_hash(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static void copytup_index(Tuplesortstate *state, SortTuple *stup, void *tup);
-static void writetup_index(Tuplesortstate *state, int tapenum,
+static void writetup_index(Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
static void readtup_index(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len);
+ LogicalTape *tape, unsigned int len);
static int comparetup_datum(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static void copytup_datum(Tuplesortstate *state, SortTuple *stup, void *tup);
-static void writetup_datum(Tuplesortstate *state, int tapenum,
+static void writetup_datum(Tuplesortstate *state, LogicalTape *tape,
SortTuple *stup);
static void readtup_datum(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len);
+ LogicalTape *tape, unsigned int len);
static void free_sort_tuple(Tuplesortstate *state, SortTuple *stup);
/*
@@ -736,11 +729,11 @@ tuplesort_begin_common(int workMem, bool randomAccess)
state->currentRun = RUN_FIRST;
/*
- * maxTapes, tapeRange, and Algorithm D variables will be initialized by
- * inittapes(), if needed
+ * Tape variables (inputTapes, outputTapes, etc.) will be initialized by
+ * inittapes(), if needed.
*/
- state->result_tape = -1; /* flag that result tape has not been formed */
+ state->result_tape = NULL; /* flag that result tape has not been formed */
MemoryContextSwitchTo(oldcontext);
@@ -1171,6 +1164,9 @@ tuplesort_end(Tuplesortstate *state)
*
* Note: want to include this in reported total cost of sort, hence need
* for two #ifdef TRACE_SORT sections.
+ *
+ * We don't bother to destroy the individual tapes here, they will go away
+ * with the sortcontext.
*/
if (state->tapeset)
LogicalTapeSetClose(state->tapeset);
@@ -1825,7 +1821,7 @@ tuplesort_performsort(Tuplesortstate *state)
{
if (state->status == TSS_FINALMERGE)
elog(LOG, "performsort done (except %d-way final merge): %s",
- state->activeTapes,
+ state->nInputTapes,
pg_rusage_show(&state->ru_start));
else
elog(LOG, "performsort done: %s",
@@ -1949,8 +1945,7 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
* end of file; back up to fetch last tuple's ending length
* word. If seek fails we must have a completely empty file.
*/
- seeked = LogicalTapeBackspace(state->tapeset,
- state->result_tape,
+ seeked = LogicalTapeBackspace(state->result_tape,
2 * sizeof(unsigned int));
if (seeked == 0)
return false;
@@ -1964,8 +1959,7 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
* Back up and fetch previously-returned tuple's ending length
* word. If seek fails, assume we are at start of file.
*/
- seeked = LogicalTapeBackspace(state->tapeset,
- state->result_tape,
+ seeked = LogicalTapeBackspace(state->result_tape,
sizeof(unsigned int));
if (seeked == 0)
return false;
@@ -1976,8 +1970,7 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
/*
* Back up to get ending length word of tuple before it.
*/
- seeked = LogicalTapeBackspace(state->tapeset,
- state->result_tape,
+ seeked = LogicalTapeBackspace(state->result_tape,
tuplen + 2 * sizeof(unsigned int));
if (seeked == tuplen + sizeof(unsigned int))
{
@@ -2001,8 +1994,7 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
* Note: READTUP expects we are positioned after the initial
* length word of the tuple, so back up to that point.
*/
- if (!LogicalTapeBackspace(state->tapeset,
- state->result_tape,
+ if (!LogicalTapeBackspace(state->result_tape,
tuplen))
elog(ERROR, "bogus tuple length in backward scan");
READTUP(state, stup, state->result_tape, tuplen);
@@ -2037,7 +2029,8 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
*/
if (state->memtupcount > 0)
{
- int srcTape = state->memtuples[0].tupindex;
+ int srcTapeIndex = state->memtuples[0].tupindex;
+ LogicalTape *srcTape = state->inputTapes[srcTapeIndex];
SortTuple newtup;
*stup = state->memtuples[0];
@@ -2059,16 +2052,16 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
* Remove the top node from the heap.
*/
tuplesort_heap_delete_top(state, false);
+ state->nInputRuns--;
/*
- * Rewind to free the read buffer. It'd go away at the
- * end of the sort anyway, but better to release the
- * memory early.
+ * Close the tape. It'd go away at the end of the sort
+ * anyway, but better to release the memory early.
*/
- LogicalTapeRewindForWrite(state->tapeset, srcTape);
+ LogicalTapeClose(srcTape);
return true;
}
- newtup.tupindex = srcTape;
+ newtup.tupindex = srcTapeIndex;
tuplesort_heap_replace_top(state, &newtup, false);
return true;
}
@@ -2347,20 +2340,16 @@ useselection(Tuplesortstate *state)
static void
inittapes(Tuplesortstate *state)
{
- int maxTapes,
- j;
+ int j;
int64 tapeSpace;
- /* Compute number of tapes to use: merge order plus 1 */
- maxTapes = tuplesort_merge_order(state->allowedMem) + 1;
-
- state->maxTapes = maxTapes;
- state->tapeRange = maxTapes - 1;
+ /* Compute number of input tapes to use (merge order) */
+ state->maxInputTapes = tuplesort_merge_order(state->allowedMem);
#ifdef TRACE_SORT
if (trace_sort)
elog(LOG, "switching to external sort with %d tapes: %s",
- maxTapes, pg_rusage_show(&state->ru_start));
+ state->maxInputTapes, pg_rusage_show(&state->ru_start));
#endif
/*
@@ -2384,15 +2373,10 @@ inittapes(Tuplesortstate *state)
PrepareTempTablespaces();
/*
- * Create the tape set and allocate the per-tape data arrays.
+ * Create the tape set. It is initially empty, the tapes are created as
+ * needed.
*/
- state->tapeset = LogicalTapeSetCreate(maxTapes);
-
- state->mergeactive = (bool *) palloc0(maxTapes * sizeof(bool));
- state->tp_fib = (int *) palloc0(maxTapes * sizeof(int));
- state->tp_runs = (int *) palloc0(maxTapes * sizeof(int));
- state->tp_dummy = (int *) palloc0(maxTapes * sizeof(int));
- state->tp_tapenum = (int *) palloc0(maxTapes * sizeof(int));
+ state->tapeset = LogicalTapeSetCreate();
/*
* Give replacement selection a try based on user setting. There will be
@@ -2434,63 +2418,50 @@ inittapes(Tuplesortstate *state)
state->currentRun = RUN_FIRST;
/*
- * Initialize variables of Algorithm D (step D1).
+ * Initialize logical tape variables.
*/
- for (j = 0; j < maxTapes; j++)
- {
- state->tp_fib[j] = 1;
- state->tp_runs[j] = 0;
- state->tp_dummy[j] = 1;
- state->tp_tapenum[j] = j;
- }
- state->tp_fib[state->tapeRange] = 0;
- state->tp_dummy[state->tapeRange] = 0;
+ state->inputTapes = NULL;
+ state->nInputTapes = 0;
+ state->nInputRuns = 0;
+
+ state->outputTapes = palloc0(state->maxInputTapes * sizeof(LogicalTape *));
+ state->nOutputTapes = 0;
+ state->nOutputRuns = 0;
- state->Level = 1;
- state->destTape = 0;
+ state->destTape = NULL;
state->status = TSS_BUILDRUNS;
+
+ selectnewtape(state);
}
/*
- * selectnewtape -- select new tape for new initial run.
+ * selectnewtape -- select next tape to output to.
*
* This is called after finishing a run when we know another run
- * must be started. This implements steps D3, D4 of Algorithm D.
+ * must be started. This is used both when building the initial
+ * runs, and during merge passes.
*/
static void
selectnewtape(Tuplesortstate *state)
{
- int j;
- int a;
-
- /*
- * Pause the old tape, to release the memory that was used for its buffer,
- * so that we can use it for building the next run.
- */
- LogicalTapePause(state->tapeset, state->destTape);
-
- /* Step D3: advance j (destTape) */
- if (state->tp_dummy[state->destTape] < state->tp_dummy[state->destTape + 1])
- {
- state->destTape++;
- return;
- }
- if (state->tp_dummy[state->destTape] != 0)
+ if (state->nOutputRuns < state->maxInputTapes)
{
- state->destTape = 0;
- return;
+ /* Create a new tape to hold the next run */
+ Assert(state->outputTapes[state->nOutputRuns] == NULL);
+ Assert(state->nOutputRuns == state->nOutputTapes);
+ state->outputTapes[state->nOutputRuns] = LogicalTapeCreate(state->tapeset);
+ state->nOutputTapes++;
}
-
- /* Step D4: increase level */
- state->Level++;
- a = state->tp_fib[0];
- for (j = 0; j < state->tapeRange; j++)
+ else
{
- state->tp_dummy[j] = a + state->tp_fib[j + 1] - state->tp_fib[j];
- state->tp_fib[j] = a + state->tp_fib[j + 1];
+ /*
+ * We have reached the max number of tapes. Append to an existing
+ * tape.
+ */
}
- state->destTape = 0;
+ state->destTape = state->outputTapes[state->nOutputRuns % state->nOutputTapes];
+ state->nOutputRuns++;
}
/*
@@ -2529,18 +2500,14 @@ init_slab_allocator(Tuplesortstate *state, int numSlots)
/*
* mergeruns -- merge all the completed initial runs.
*
- * This implements steps D5, D6 of Algorithm D. All input data has
+ * This implements the Balanced k-Way Merge Algorithm. All input data has
* already been written to initial runs on tape (see dumptuples).
*/
static void
mergeruns(Tuplesortstate *state)
{
- int tapenum,
- svTape,
- svRuns,
- svDummy;
- int numTapes;
- int numInputTapes;
+ int tapenum;
+ int64 tape_buffer_mem;
Assert(state->status == TSS_BUILDRUNS);
Assert(state->memtupcount == 0);
@@ -2577,37 +2544,19 @@ mergeruns(Tuplesortstate *state)
state->memtuples = NULL;
/*
- * If we had fewer runs than tapes, refund the memory that we imagined we
- * would need for the tape buffers of the unused tapes.
- *
- * numTapes and numInputTapes reflect the actual number of tapes we will
- * use. Note that the output tape's tape number is maxTapes - 1, so the
- * tape numbers of the used tapes are not consecutive, and you cannot just
- * loop from 0 to numTapes to visit all used tapes!
- */
- if (state->Level == 1)
- {
- numInputTapes = state->currentRun;
- numTapes = numInputTapes + 1;
- FREEMEM(state, (state->maxTapes - numTapes) * TAPE_BUFFER_OVERHEAD);
- }
- else
- {
- numInputTapes = state->tapeRange;
- numTapes = state->maxTapes;
- }
-
- /*
* Initialize the slab allocator. We need one slab slot per input tape,
* for the tuples in the heap, plus one to hold the tuple last returned
* from tuplesort_gettuple. (If we're sorting pass-by-val Datums,
* however, we don't need to do allocate anything.)
*
+ * In a multi-pass merge, we could shrink this allocation between each
+ * pass, when the number of tapes is reduced, but we don't bother.
+ *
* From this point on, we no longer use the USEMEM()/LACKMEM() mechanism
* to track memory usage of individual tuples.
*/
if (state->tuples)
- init_slab_allocator(state, numInputTapes + 1);
+ init_slab_allocator(state, state->nOutputTapes + 1);
else
init_slab_allocator(state, 0);
@@ -2616,13 +2565,13 @@ mergeruns(Tuplesortstate *state)
* volume is between 1X and 2X workMem when replacement selection is used,
* but something we particular count on when input is presorted), we can
* just use that tape as the finished output, rather than doing a useless
- * merge. (This obvious optimization is not in Knuth's algorithm.)
+ * merge.
*/
if (state->currentRun == RUN_SECOND)
{
- state->result_tape = state->tp_tapenum[state->destTape];
+ state->result_tape = state->outputTapes[0];
/* must freeze and rewind the finished output tape */
- LogicalTapeFreeze(state->tapeset, state->result_tape);
+ LogicalTapeFreeze(state->result_tape);
state->status = TSS_SORTEDONTAPE;
return;
}
@@ -2630,65 +2579,73 @@ mergeruns(Tuplesortstate *state)
/*
* Use all the spare memory we have available for read buffers among the
* input tapes.
- *
- * We do this only after checking for the case that we produced only one
- * initial run, because there is no need to use a large read buffer when
- * we're reading from a single tape. With one tape, the I/O pattern will
- * be the same regardless of the buffer size.
- *
- * We don't try to "rebalance" the memory among tapes, when we start a new
- * merge phase, even if some tapes are inactive in the new phase. That
- * would be hard, because logtape.c doesn't know where one run ends and
- * another begins. When a new merge phase begins, and a tape doesn't
- * participate in it, its buffer nevertheless already contains tuples from
- * the next run on same tape, so we cannot release the buffer. That's OK
- * in practice, merge performance isn't that sensitive to the amount of
- * buffers used, and most merge phases use all or almost all tapes,
- * anyway.
*/
#ifdef TRACE_SORT
if (trace_sort)
elog(LOG, "using " INT64_FORMAT " KB of memory for read buffers among %d input tapes",
- (state->availMem) / 1024, numInputTapes);
+ (state->availMem) / 1024, state->nOutputTapes);
#endif
- state->read_buffer_size = state->availMem / numInputTapes;
+ state->read_buffer_mem = state->availMem;
USEMEM(state, state->availMem);
/*
* Allocate a new 'memtuples' array, for the heap. It will hold one tuple
* from each input tape.
+ *
+ * We could shrink this, too, between passes in a multi-pass merge, but
+ * we don't bother. (The initial input tapes are still in outputTapes.
+ * The number of input tapes will not increase between passes.)
*/
- state->memtupsize = numInputTapes;
- state->memtuples = (SortTuple *) palloc(numInputTapes * sizeof(SortTuple));
+ state->memtupsize = state->nOutputTapes;
+ state->memtuples = (SortTuple *) palloc(state->nOutputTapes * sizeof(SortTuple));
USEMEM(state, GetMemoryChunkSpace(state->memtuples));
- /* End of step D2: rewind all output tapes to prepare for merging */
- for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
- LogicalTapeRewindForRead(state->tapeset, tapenum, state->read_buffer_size);
+ /* We will use all remaining memory for read buffers */
+ tape_buffer_mem = state->availMem;
+ USEMEM(state, tape_buffer_mem);
for (;;)
{
/*
- * At this point we know that tape[T] is empty. If there's just one
- * (real or dummy) run left on each input tape, then only one merge
- * pass remains. If we don't have to produce a materialized sorted
- * tape, we can stop at this point and do the final merge on-the-fly.
+ * On the first iteration, or if we have read all the runs from the input tapes in
+ * a multi-pass merge, it's time to start a new pass. Rewind all the output tapes,
+ * and make them inputs for the next pass.
*/
- if (!state->randomAccess)
+ if (state->nInputRuns == 0)
{
- bool allOneRun = true;
-
- Assert(state->tp_runs[state->tapeRange] == 0);
- for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
+ /* Close the old, emptied, input tapes */
+ if (state->nInputTapes > 0)
{
- if (state->tp_runs[tapenum] + state->tp_dummy[tapenum] != 1)
- {
- allOneRun = false;
- break;
- }
+ for (tapenum = 0; tapenum < state->nInputTapes; tapenum++)
+ LogicalTapeClose(state->inputTapes[tapenum]);
+ pfree(state->inputTapes);
}
- if (allOneRun)
+
+ /* Previous pass's outputs become next pass's inputs. */
+ state->inputTapes = state->outputTapes;
+ state->nInputTapes = state->nOutputTapes;
+ state->nInputRuns = state->nOutputRuns;
+
+ /*
+ * Reset output tape variables. (The actual LogicalTapes will be created
+ * as needed, we just allocate a large-enough array for them here.)
+ */
+ state->outputTapes = palloc0(state->nInputTapes * sizeof(LogicalTape *));
+ state->nOutputTapes = 0;
+ state->nOutputRuns = 0;
+
+ /* Prepare the new input tapes for merge pass. */
+ for (tapenum = 0; tapenum < state->nInputTapes; tapenum++)
+ LogicalTapeRewindForRead(state->inputTapes[tapenum],
+ state->read_buffer_mem / state->nInputTapes);
+
+ /*
+ * If there's just one run left on each input tape, then only one merge pass
+ * remains. If we don't have to produce a materialized sorted tape, we can
+ * stop at this point and do the final merge on-the-fly.
+ */
+ if (!state->randomAccess && state->nInputRuns <= state->nInputTapes)
{
/* Tell logtape.c we won't be writing anymore */
LogicalTapeSetForgetFreeSpace(state->tapeset);
@@ -2699,93 +2656,46 @@ mergeruns(Tuplesortstate *state)
}
}
- /* Step D5: merge runs onto tape[T] until tape[P] is empty */
- while (state->tp_runs[state->tapeRange - 1] ||
- state->tp_dummy[state->tapeRange - 1])
- {
- bool allDummy = true;
-
- for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
- {
- if (state->tp_dummy[tapenum] == 0)
- {
- allDummy = false;
- break;
- }
- }
+ /* Select an output tape */
+ selectnewtape(state);
- if (allDummy)
- {
- state->tp_dummy[state->tapeRange]++;
- for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
- state->tp_dummy[tapenum]--;
- }
- else
- mergeonerun(state);
- }
+ /* Merge one run from each input tape. */
+ mergeonerun(state);
- /* Step D6: decrease level */
- if (--state->Level == 0)
- break;
- /* rewind output tape T to use as new input */
- LogicalTapeRewindForRead(state->tapeset, state->tp_tapenum[state->tapeRange],
- state->read_buffer_size);
- /* rewind used-up input tape P, and prepare it for write pass */
- LogicalTapeRewindForWrite(state->tapeset, state->tp_tapenum[state->tapeRange - 1]);
- state->tp_runs[state->tapeRange - 1] = 0;
+ LogicalTapePause(state->destTape);
/*
- * reassign tape units per step D6; note we no longer care about A[]
+ * If the input tapes are empty, and we output only one output run,
+ * we're done. The current output tape contains the final result.
*/
- svTape = state->tp_tapenum[state->tapeRange];
- svDummy = state->tp_dummy[state->tapeRange];
- svRuns = state->tp_runs[state->tapeRange];
- for (tapenum = state->tapeRange; tapenum > 0; tapenum--)
- {
- state->tp_tapenum[tapenum] = state->tp_tapenum[tapenum - 1];
- state->tp_dummy[tapenum] = state->tp_dummy[tapenum - 1];
- state->tp_runs[tapenum] = state->tp_runs[tapenum - 1];
- }
- state->tp_tapenum[0] = svTape;
- state->tp_dummy[0] = svDummy;
- state->tp_runs[0] = svRuns;
+ if (state->nInputRuns == 0 && state->nOutputRuns <= 1)
+ break;
}
/*
- * Done. Knuth says that the result is on TAPE[1], but since we exited
- * the loop without performing the last iteration of step D6, we have not
- * rearranged the tape unit assignment, and therefore the result is on
- * TAPE[T]. We need to do it this way so that we can freeze the final
- * output tape while rewinding it. The last iteration of step D6 would be
- * a waste of cycles anyway...
+ * Done.
*/
- state->result_tape = state->tp_tapenum[state->tapeRange];
- LogicalTapeFreeze(state->tapeset, state->result_tape);
+ state->result_tape = state->outputTapes[0];
+ LogicalTapeFreeze(state->result_tape);
state->status = TSS_SORTEDONTAPE;
- /* Release the read buffers of all the other tapes, by rewinding them. */
- for (tapenum = 0; tapenum < state->maxTapes; tapenum++)
- {
- if (tapenum != state->result_tape)
- LogicalTapeRewindForWrite(state->tapeset, tapenum);
- }
+ /* Release the read buffers of all the now-empty input tapes. */
+ for (tapenum = 0; tapenum < state->nInputTapes; tapenum++)
+ LogicalTapeClose(state->inputTapes[tapenum]);
}
/*
- * Merge one run from each input tape, except ones with dummy runs.
- *
- * This is the inner loop of Algorithm D step D5. We know that the
- * output tape is TAPE[T].
+ * Merge one run from each input tape.
*/
static void
mergeonerun(Tuplesortstate *state)
{
- int destTape = state->tp_tapenum[state->tapeRange];
- int srcTape;
+ int srcTapeIndex;
+ LogicalTape *srcTape;
/*
* Start the merge by loading one tuple from each active source tape into
- * the heap. We can also decrease the input run/dummy run counts.
+ * the heap.
*/
beginmerge(state);
@@ -2799,8 +2709,9 @@ mergeonerun(Tuplesortstate *state)
SortTuple stup;
/* write the tuple to destTape */
- srcTape = state->memtuples[0].tupindex;
- WRITETUP(state, destTape, &state->memtuples[0]);
+ srcTapeIndex = state->memtuples[0].tupindex;
+ srcTape = state->inputTapes[srcTapeIndex];
+ WRITETUP(state, state->destTape, &state->memtuples[0]);
/* recycle the slot of the tuple we just wrote out, for the next read */
RELEASE_SLAB_SLOT(state, state->memtuples[0].tuple);
@@ -2811,24 +2722,26 @@ mergeonerun(Tuplesortstate *state)
*/
if (mergereadnext(state, srcTape, &stup))
{
- stup.tupindex = srcTape;
+ stup.tupindex = srcTapeIndex;
tuplesort_heap_replace_top(state, &stup, false);
}
else
+ {
tuplesort_heap_delete_top(state, false);
+ state->nInputRuns--;
+ }
}
/*
* When the heap empties, we're done. Write an end-of-run marker on the
- * output tape, and increment its count of real runs.
+ * output tape.
*/
- markrunend(state, destTape);
- state->tp_runs[state->tapeRange]++;
+ markrunend(state, state->destTape);
#ifdef TRACE_SORT
if (trace_sort)
- elog(LOG, "finished %d-way merge step: %s", state->activeTapes,
+ elog(LOG, "finished %d-way merge step: %s", state->nInputTapes,
pg_rusage_show(&state->ru_start));
#endif
}
@@ -2836,48 +2749,26 @@ mergeonerun(Tuplesortstate *state)
/*
* beginmerge - initialize for a merge pass
*
- * We decrease the counts of real and dummy runs for each tape, and mark
- * which tapes contain active input runs in mergeactive[]. Then, fill the
- * merge heap with the first tuple from each active tape.
+ * Fill the merge heap with the first tuple from each input tape.
*/
static void
beginmerge(Tuplesortstate *state)
{
int activeTapes;
- int tapenum;
- int srcTape;
+ int srcTapeIndex;
/* Heap should be empty here */
Assert(state->memtupcount == 0);
- /* Adjust run counts and mark the active tapes */
- memset(state->mergeactive, 0,
- state->maxTapes * sizeof(*state->mergeactive));
- activeTapes = 0;
- for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
- {
- if (state->tp_dummy[tapenum] > 0)
- state->tp_dummy[tapenum]--;
- else
- {
- Assert(state->tp_runs[tapenum] > 0);
- state->tp_runs[tapenum]--;
- srcTape = state->tp_tapenum[tapenum];
- state->mergeactive[srcTape] = true;
- activeTapes++;
- }
- }
- Assert(activeTapes > 0);
- state->activeTapes = activeTapes;
+ activeTapes = Min(state->nInputTapes, state->nInputRuns);
- /* Load the merge heap with the first tuple from each input tape */
- for (srcTape = 0; srcTape < state->maxTapes; srcTape++)
+ for (srcTapeIndex = 0; srcTapeIndex < activeTapes; srcTapeIndex++)
{
SortTuple tup;
- if (mergereadnext(state, srcTape, &tup))
+ if (mergereadnext(state, state->inputTapes[srcTapeIndex], &tup))
{
- tup.tupindex = srcTape;
+ tup.tupindex = srcTapeIndex;
tuplesort_heap_insert(state, &tup, false);
}
}
@@ -2889,19 +2780,13 @@ beginmerge(Tuplesortstate *state)
* Returns false on EOF.
*/
static bool
-mergereadnext(Tuplesortstate *state, int srcTape, SortTuple *stup)
+mergereadnext(Tuplesortstate *state, LogicalTape *srcTape, SortTuple *stup)
{
unsigned int tuplen;
- if (!state->mergeactive[srcTape])
- return false; /* tape's run is already exhausted */
-
/* read next tuple, if any */
if ((tuplen = getlen(state, srcTape, true)) == 0)
- {
- state->mergeactive[srcTape] = false;
return false;
- }
READTUP(state, stup, srcTape, tuplen);
return true;
@@ -2944,7 +2829,7 @@ dumptuples(Tuplesortstate *state, bool alltuples)
* Dump the heap's frontmost entry, and remove it from the heap.
*/
Assert(state->memtupcount > 0);
- WRITETUP(state, state->tp_tapenum[state->destTape],
+ WRITETUP(state, state->destTape,
&state->memtuples[0]);
tuplesort_heap_delete_top(state, true);
}
@@ -2964,17 +2849,21 @@ dumptuples(Tuplesortstate *state, bool alltuples)
if (state->memtupcount == 0 ||
state->memtuples[0].tupindex == HEAP_RUN_NEXT)
{
- markrunend(state, state->tp_tapenum[state->destTape]);
+ markrunend(state, state->destTape);
Assert(state->currentRun == RUN_FIRST);
state->currentRun++;
- state->tp_runs[state->destTape]++;
- state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
+
+ /*
+ * Pause the old tape, to release the memory that was used for
+ * its buffer, so that we can use it for building the next run.
+ */
+ LogicalTapePause(state->destTape);
#ifdef TRACE_SORT
if (trace_sort)
elog(LOG, "finished incrementally writing %s run %d to tape %d: %s",
(state->memtupcount == 0) ? "only" : "first",
- state->currentRun, state->destTape,
+ state->currentRun, state->nOutputTapes % state->nOutputRuns,
pg_rusage_show(&state->ru_start));
#endif
@@ -3035,16 +2924,10 @@ dumpbatch(Tuplesortstate *state, bool alltuples)
* In general, short final runs are quite possible. Rather than allowing
* a special case where there was a superfluous selectnewtape() call (i.e.
* a call with no subsequent run actually written to destTape), we prefer
- * to write out a 0 tuple run.
- *
- * mergereadnext() is prepared for 0 tuple runs, and will reliably mark
- * the tape inactive for the merge when called from beginmerge(). This
- * case is therefore similar to the case where mergeonerun() finds a dummy
- * run for the tape, and so doesn't need to merge a run from the tape (or
- * conceptually "merges" the dummy run, if you prefer). According to
- * Knuth, Algorithm D "isn't strictly optimal" in its method of
- * distribution and dummy run assignment; this edge case seems very
- * unlikely to make that appreciably worse.
+ * to write out a 0 tuple run. In the worst case, that could add another
+ * merge pass, if that pushes us over the threshold, but it's unlikely
+ * enough to not warrant a special case. (XXX: Actually, I think some
+ * refactoring to avoid that would be in order...)
*/
Assert(state->status == TSS_BUILDRUNS);
@@ -3081,8 +2964,7 @@ dumpbatch(Tuplesortstate *state, bool alltuples)
memtupwrite = state->memtupcount;
for (i = 0; i < memtupwrite; i++)
{
- WRITETUP(state, state->tp_tapenum[state->destTape],
- &state->memtuples[i]);
+ WRITETUP(state, state->destTape, &state->memtuples[i]);
state->memtupcount--;
}
@@ -3095,14 +2977,18 @@ dumpbatch(Tuplesortstate *state, bool alltuples)
*/
MemoryContextReset(state->tuplecontext);
- markrunend(state, state->tp_tapenum[state->destTape]);
- state->tp_runs[state->destTape]++;
- state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
+ markrunend(state, state->destTape);
+
+ /*
+ * Pause the old tape, to release the memory that was used for its buffer,
+ * so that we can use it for building the next run.
+ */
+ LogicalTapePause(state->destTape);
#ifdef TRACE_SORT
if (trace_sort)
elog(LOG, "finished writing run %d to tape %d: %s",
- state->currentRun, state->destTape,
+ state->currentRun, (state->currentRun - 1) % state->nOutputTapes + 1,
pg_rusage_show(&state->ru_start));
#endif
@@ -3129,9 +3015,7 @@ tuplesort_rescan(Tuplesortstate *state)
state->markpos_eof = false;
break;
case TSS_SORTEDONTAPE:
- LogicalTapeRewindForRead(state->tapeset,
- state->result_tape,
- 0);
+ LogicalTapeRewindForRead(state->result_tape, 0);
state->eof_reached = false;
state->markpos_block = 0L;
state->markpos_offset = 0;
@@ -3162,8 +3046,7 @@ tuplesort_markpos(Tuplesortstate *state)
state->markpos_eof = state->eof_reached;
break;
case TSS_SORTEDONTAPE:
- LogicalTapeTell(state->tapeset,
- state->result_tape,
+ LogicalTapeTell(state->result_tape,
&state->markpos_block,
&state->markpos_offset);
state->markpos_eof = state->eof_reached;
@@ -3194,8 +3077,7 @@ tuplesort_restorepos(Tuplesortstate *state)
state->eof_reached = state->markpos_eof;
break;
case TSS_SORTEDONTAPE:
- LogicalTapeSeek(state->tapeset,
- state->result_tape,
+ LogicalTapeSeek(state->result_tape,
state->markpos_block,
state->markpos_offset);
state->eof_reached = state->markpos_eof;
@@ -3527,11 +3409,11 @@ reversedirection(Tuplesortstate *state)
*/
static unsigned int
-getlen(Tuplesortstate *state, int tapenum, bool eofOK)
+getlen(Tuplesortstate *state, LogicalTape *tape, bool eofOK)
{
unsigned int len;
- if (LogicalTapeRead(state->tapeset, tapenum,
+ if (LogicalTapeRead(tape,
&len, sizeof(len)) != sizeof(len))
elog(ERROR, "unexpected end of tape");
if (len == 0 && !eofOK)
@@ -3540,11 +3422,11 @@ getlen(Tuplesortstate *state, int tapenum, bool eofOK)
}
static void
-markrunend(Tuplesortstate *state, int tapenum)
+markrunend(Tuplesortstate *state, LogicalTape *tape)
{
unsigned int len = 0;
- LogicalTapeWrite(state->tapeset, tapenum, (void *) &len, sizeof(len));
+ LogicalTapeWrite(tape, (void *) &len, sizeof(len));
}
/*
@@ -3722,7 +3604,7 @@ copytup_heap(Tuplesortstate *state, SortTuple *stup, void *tup)
}
static void
-writetup_heap(Tuplesortstate *state, int tapenum, SortTuple *stup)
+writetup_heap(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
{
MinimalTuple tuple = (MinimalTuple) stup->tuple;
@@ -3733,13 +3615,10 @@ writetup_heap(Tuplesortstate *state, int tapenum, SortTuple *stup)
/* total on-disk footprint: */
unsigned int tuplen = tupbodylen + sizeof(int);
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) &tuplen, sizeof(tuplen));
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) tupbody, tupbodylen);
+ LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, (void *) tupbody, tupbodylen);
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
if (!state->slabAllocatorUsed)
{
@@ -3750,7 +3629,7 @@ writetup_heap(Tuplesortstate *state, int tapenum, SortTuple *stup)
static void
readtup_heap(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len)
+ LogicalTape *tape, unsigned int len)
{
unsigned int tupbodylen = len - sizeof(int);
unsigned int tuplen = tupbodylen + MINIMAL_TUPLE_DATA_OFFSET;
@@ -3760,11 +3639,9 @@ readtup_heap(Tuplesortstate *state, SortTuple *stup,
/* read in the tuple proper */
tuple->t_len = tuplen;
- LogicalTapeReadExact(state->tapeset, tapenum,
- tupbody, tupbodylen);
+ LogicalTapeReadExact(tape, tupbody, tupbodylen);
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeReadExact(state->tapeset, tapenum,
- &tuplen, sizeof(tuplen));
+ LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
stup->tuple = (void *) tuple;
/* set up first-column key value */
htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
@@ -3965,21 +3842,17 @@ copytup_cluster(Tuplesortstate *state, SortTuple *stup, void *tup)
}
static void
-writetup_cluster(Tuplesortstate *state, int tapenum, SortTuple *stup)
+writetup_cluster(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
{
HeapTuple tuple = (HeapTuple) stup->tuple;
unsigned int tuplen = tuple->t_len + sizeof(ItemPointerData) + sizeof(int);
/* We need to store t_self, but not other fields of HeapTupleData */
- LogicalTapeWrite(state->tapeset, tapenum,
- &tuplen, sizeof(tuplen));
- LogicalTapeWrite(state->tapeset, tapenum,
- &tuple->t_self, sizeof(ItemPointerData));
- LogicalTapeWrite(state->tapeset, tapenum,
- tuple->t_data, tuple->t_len);
+ LogicalTapeWrite(tape, &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, &tuple->t_self, sizeof(ItemPointerData));
+ LogicalTapeWrite(tape, tuple->t_data, tuple->t_len);
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeWrite(state->tapeset, tapenum,
- &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, &tuplen, sizeof(tuplen));
if (!state->slabAllocatorUsed)
{
@@ -3990,7 +3863,7 @@ writetup_cluster(Tuplesortstate *state, int tapenum, SortTuple *stup)
static void
readtup_cluster(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int tuplen)
+ LogicalTape *tape, unsigned int tuplen)
{
unsigned int t_len = tuplen - sizeof(ItemPointerData) - sizeof(int);
HeapTuple tuple = (HeapTuple) readtup_alloc(state,
@@ -3999,16 +3872,13 @@ readtup_cluster(Tuplesortstate *state, SortTuple *stup,
/* Reconstruct the HeapTupleData header */
tuple->t_data = (HeapTupleHeader) ((char *) tuple + HEAPTUPLESIZE);
tuple->t_len = t_len;
- LogicalTapeReadExact(state->tapeset, tapenum,
- &tuple->t_self, sizeof(ItemPointerData));
+ LogicalTapeReadExact(tape, &tuple->t_self, sizeof(ItemPointerData));
/* We don't currently bother to reconstruct t_tableOid */
tuple->t_tableOid = InvalidOid;
/* Read in the tuple body */
- LogicalTapeReadExact(state->tapeset, tapenum,
- tuple->t_data, tuple->t_len);
+ LogicalTapeReadExact(tape, tuple->t_data, tuple->t_len);
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeReadExact(state->tapeset, tapenum,
- &tuplen, sizeof(tuplen));
+ LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
stup->tuple = (void *) tuple;
/* set up first-column key value, if it's a simple column */
if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
@@ -4271,19 +4141,16 @@ copytup_index(Tuplesortstate *state, SortTuple *stup, void *tup)
}
static void
-writetup_index(Tuplesortstate *state, int tapenum, SortTuple *stup)
+writetup_index(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
{
IndexTuple tuple = (IndexTuple) stup->tuple;
unsigned int tuplen;
tuplen = IndexTupleSize(tuple) + sizeof(tuplen);
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) &tuplen, sizeof(tuplen));
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) tuple, IndexTupleSize(tuple));
+ LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, (void *) tuple, IndexTupleSize(tuple));
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) &tuplen, sizeof(tuplen));
+ LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
if (!state->slabAllocatorUsed)
{
@@ -4294,16 +4161,14 @@ writetup_index(Tuplesortstate *state, int tapenum, SortTuple *stup)
static void
readtup_index(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len)
+ LogicalTape *tape, unsigned int len)
{
unsigned int tuplen = len - sizeof(unsigned int);
IndexTuple tuple = (IndexTuple) readtup_alloc(state, tuplen);
- LogicalTapeReadExact(state->tapeset, tapenum,
- tuple, tuplen);
+ LogicalTapeReadExact(tape, tuple, tuplen);
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeReadExact(state->tapeset, tapenum,
- &tuplen, sizeof(tuplen));
+ LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
stup->tuple = (void *) tuple;
/* set up first-column key value */
stup->datum1 = index_getattr(tuple,
@@ -4345,7 +4210,7 @@ copytup_datum(Tuplesortstate *state, SortTuple *stup, void *tup)
}
static void
-writetup_datum(Tuplesortstate *state, int tapenum, SortTuple *stup)
+writetup_datum(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
{
void *waddr;
unsigned int tuplen;
@@ -4370,13 +4235,10 @@ writetup_datum(Tuplesortstate *state, int tapenum, SortTuple *stup)
writtenlen = tuplen + sizeof(unsigned int);
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) &writtenlen, sizeof(writtenlen));
- LogicalTapeWrite(state->tapeset, tapenum,
- waddr, tuplen);
+ LogicalTapeWrite(tape, (void *) &writtenlen, sizeof(writtenlen));
+ LogicalTapeWrite(tape, waddr, tuplen);
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeWrite(state->tapeset, tapenum,
- (void *) &writtenlen, sizeof(writtenlen));
+ LogicalTapeWrite(tape, (void *) &writtenlen, sizeof(writtenlen));
if (!state->slabAllocatorUsed && stup->tuple)
{
@@ -4387,7 +4249,7 @@ writetup_datum(Tuplesortstate *state, int tapenum, SortTuple *stup)
static void
readtup_datum(Tuplesortstate *state, SortTuple *stup,
- int tapenum, unsigned int len)
+ LogicalTape *tape, unsigned int len)
{
unsigned int tuplen = len - sizeof(unsigned int);
@@ -4401,8 +4263,7 @@ readtup_datum(Tuplesortstate *state, SortTuple *stup,
else if (!state->tuples)
{
Assert(tuplen == sizeof(Datum));
- LogicalTapeReadExact(state->tapeset, tapenum,
- &stup->datum1, tuplen);
+ LogicalTapeReadExact(tape, &stup->datum1, tuplen);
stup->isnull1 = false;
stup->tuple = NULL;
}
@@ -4410,16 +4271,14 @@ readtup_datum(Tuplesortstate *state, SortTuple *stup,
{
void *raddr = readtup_alloc(state, tuplen);
- LogicalTapeReadExact(state->tapeset, tapenum,
- raddr, tuplen);
+ LogicalTapeReadExact(tape, raddr, tuplen);
stup->datum1 = PointerGetDatum(raddr);
stup->isnull1 = false;
stup->tuple = raddr;
}
if (state->randomAccess) /* need trailing length word? */
- LogicalTapeReadExact(state->tapeset, tapenum,
- &tuplen, sizeof(tuplen));
+ LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
}
/*
diff --git a/src/include/utils/logtape.h b/src/include/utils/logtape.h
index 3a51cfb..6add0a5 100644
--- a/src/include/utils/logtape.h
+++ b/src/include/utils/logtape.h
@@ -16,32 +16,33 @@
#ifndef LOGTAPE_H
#define LOGTAPE_H
-/* LogicalTapeSet is an opaque type whose details are not known outside logtape.c. */
-
+/*
+ * LogicalTapeSet and LogicalTape are opaque types whose details are not
+ * known outside logtape.c.
+ */
typedef struct LogicalTapeSet LogicalTapeSet;
+typedef struct LogicalTape LogicalTape;
+
/*
* prototypes for functions in logtape.c
*/
-extern LogicalTapeSet *LogicalTapeSetCreate(int ntapes);
+extern LogicalTapeSet *LogicalTapeSetCreate(void);
extern void LogicalTapeSetClose(LogicalTapeSet *lts);
extern void LogicalTapeSetForgetFreeSpace(LogicalTapeSet *lts);
-extern size_t LogicalTapeRead(LogicalTapeSet *lts, int tapenum,
- void *ptr, size_t size);
-extern void LogicalTapeWrite(LogicalTapeSet *lts, int tapenum,
- void *ptr, size_t size);
-extern void LogicalTapeRewindForRead(LogicalTapeSet *lts, int tapenum,
- size_t buffer_size);
-extern void LogicalTapeRewindForWrite(LogicalTapeSet *lts, int tapenum);
-extern void LogicalTapePause(LogicalTapeSet *lts, int tapenum);
-extern void LogicalTapeFreeze(LogicalTapeSet *lts, int tapenum);
-extern size_t LogicalTapeBackspace(LogicalTapeSet *lts, int tapenum,
- size_t size);
-extern void LogicalTapeSeek(LogicalTapeSet *lts, int tapenum,
- long blocknum, int offset);
-extern void LogicalTapeTell(LogicalTapeSet *lts, int tapenum,
- long *blocknum, int *offset);
extern long LogicalTapeSetBlocks(LogicalTapeSet *lts);
+extern LogicalTape *LogicalTapeCreate(LogicalTapeSet *lts);
+extern void LogicalTapeClose(LogicalTape *lt);
+extern size_t LogicalTapeRead(LogicalTape *lt, void *ptr, size_t size);
+extern void LogicalTapeWrite(LogicalTape *lt, void *ptr, size_t size);
+extern void LogicalTapeRewindForRead(LogicalTape *lt, size_t buffer_size);
+extern void LogicalTapePause(LogicalTape *lt);
+extern void LogicalTapeFreeze(LogicalTape *lt);
+extern size_t LogicalTapeBackspace(LogicalTape *lt, size_t size);
+extern void LogicalTapeSeek(LogicalTape *lt, long blocknum, int offset);
+extern void LogicalTapeTell(LogicalTape *lt, long *blocknum, int *offset);
+extern void LogicalTapeAssignReadBufferSize(LogicalTape *lt, size_t bufsize);
+
#endif /* LOGTAPE_H */
--
2.9.3
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
