logtape.c

PostgreSQL 8.1.4的源码 适用于Linux下的开源数据库系统

	 * Create top-level struct.  First LogicalTape pointer is already counted
	 * in sizeof(LogicalTapeSet).
	 */
	Assert(ntapes > 0);
	lts = (LogicalTapeSet *) palloc(sizeof(LogicalTapeSet) +
									(ntapes - 1) *sizeof(LogicalTape *));
	lts->pfile = BufFileCreateTemp(false);
	lts->nFileBlocks = 0L;
	lts->freeBlocksLen = 32;	/* reasonable initial guess */
	lts->freeBlocks = (long *) palloc(lts->freeBlocksLen * sizeof(long));
	lts->nFreeBlocks = 0;
	lts->nTapes = ntapes;

	/*
	 * Create per-tape structs, including first-level indirect blocks.
	 */
	for (i = 0; i < ntapes; i++)
	{
		lt = (LogicalTape *) palloc(sizeof(LogicalTape));
		lts->tapes[i] = lt;
		lt->indirect = (IndirectBlock *) palloc(sizeof(IndirectBlock));
		lt->indirect->nextSlot = 0;
		lt->indirect->nextup = NULL;
		lt->writing = true;
		lt->frozen = false;
		lt->dirty = false;
		lt->numFullBlocks = 0L;
		lt->lastBlockBytes = 0;
		lt->curBlockNumber = 0L;
		lt->pos = 0;
		lt->nbytes = 0;
	}
	return lts;
}

/*
 * Close a logical tape set and release all resources.
 */
void
LogicalTapeSetClose(LogicalTapeSet *lts)
{
	LogicalTape *lt;
	IndirectBlock *ib,
			   *nextib;
	int			i;

	BufFileClose(lts->pfile);
	for (i = 0; i < lts->nTapes; i++)
	{
		lt = lts->tapes[i];
		for (ib = lt->indirect; ib != NULL; ib = nextib)
		{
			nextib = ib->nextup;
			pfree(ib);
		}
		pfree(lt);
	}
	pfree(lts->freeBlocks);
	pfree(lts);
}

/*
 * Dump the dirty buffer of a logical tape.
 */
static void
ltsDumpBuffer(LogicalTapeSet *lts, LogicalTape *lt)
{
	long		datablock = ltsGetFreeBlock(lts);

	Assert(lt->dirty);
	ltsWriteBlock(lts, datablock, (void *) lt->buffer);
	ltsRecordBlockNum(lts, lt->indirect, datablock);
	lt->dirty = false;
	/* Caller must do other state update as needed */
}

/*
 * 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)
{
	LogicalTape *lt;
	size_t		nthistime;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];
	Assert(lt->writing);

	while (size > 0)
	{
		if (lt->pos >= BLCKSZ)
		{
			/* Buffer full, dump it out */
			if (lt->dirty)
				ltsDumpBuffer(lts, lt);
			else
			{
				/* Hmm, went directly from reading to writing? */
				elog(ERROR, "invalid logtape state: should be dirty");
			}
			lt->numFullBlocks++;
			lt->curBlockNumber++;
			lt->pos = 0;
			lt->nbytes = 0;
		}

		nthistime = BLCKSZ - lt->pos;
		if (nthistime > size)
			nthistime = size;
		Assert(nthistime > 0);

		memcpy(lt->buffer + lt->pos, ptr, nthistime);

		lt->dirty = true;
		lt->pos += nthistime;
		if (lt->nbytes < lt->pos)
			lt->nbytes = lt->pos;
		ptr = (void *) ((char *) ptr + nthistime);
		size -= nthistime;
	}
}

/*
 * Rewind logical tape and switch from writing to reading or vice versa.
 *
 * Unless the tape has been "frozen" in read state, forWrite must be the
 * opposite of the previous tape state.
 */
void
LogicalTapeRewind(LogicalTapeSet *lts, int tapenum, bool forWrite)
{
	LogicalTape *lt;
	long		datablocknum;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];

	if (!forWrite)
	{
		if (lt->writing)
		{
			/*
			 * Completion of a write phase.  Flush last partial data block,
			 * flush any partial indirect blocks, rewind for normal
			 * (destructive) read.
			 */
			if (lt->dirty)
				ltsDumpBuffer(lts, lt);
			lt->lastBlockBytes = lt->nbytes;
			lt->writing = false;
			datablocknum = ltsRewindIndirectBlock(lts, lt->indirect, false);
		}
		else
		{
			/*
			 * This is only OK if tape is frozen; we rewind for (another) read
			 * pass.
			 */
			Assert(lt->frozen);
			datablocknum = ltsRewindFrozenIndirectBlock(lts, lt->indirect);
		}
		/* Read the first block, or reset if tape is empty */
		lt->curBlockNumber = 0L;
		lt->pos = 0;
		lt->nbytes = 0;
		if (datablocknum != -1L)
		{
			ltsReadBlock(lts, datablocknum, (void *) lt->buffer);
			if (!lt->frozen)
				ltsReleaseBlock(lts, datablocknum);
			lt->nbytes = (lt->curBlockNumber < lt->numFullBlocks) ?
				BLCKSZ : lt->lastBlockBytes;
		}
	}
	else
	{
		/*
		 * Completion of a read phase.	Rewind and prepare for write.
		 *
		 * 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.
		 */
		IndirectBlock *ib,
				   *nextib;

		Assert(!lt->writing && !lt->frozen);
		/* Must truncate the indirect-block hierarchy down to one level. */
		for (ib = lt->indirect->nextup; ib != NULL; ib = nextib)
		{
			nextib = ib->nextup;
			pfree(ib);
		}
		lt->indirect->nextSlot = 0;
		lt->indirect->nextup = NULL;
		lt->writing = true;
		lt->dirty = false;
		lt->numFullBlocks = 0L;
		lt->lastBlockBytes = 0;
		lt->curBlockNumber = 0L;
		lt->pos = 0;
		lt->nbytes = 0;
	}
}

/*
 * Read from a logical tape.
 *
 * Early EOF is indicated by return value less than #bytes requested.
 */
size_t
LogicalTapeRead(LogicalTapeSet *lts, int tapenum,
				void *ptr, size_t size)
{
	LogicalTape *lt;
	size_t		nread = 0;
	size_t		nthistime;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];
	Assert(!lt->writing);

	while (size > 0)
	{
		if (lt->pos >= lt->nbytes)
		{
			/* Try to load more data into buffer. */
			long		datablocknum = ltsRecallNextBlockNum(lts, lt->indirect,
															 lt->frozen);

			if (datablocknum == -1L)
				break;			/* EOF */
			lt->curBlockNumber++;
			lt->pos = 0;
			ltsReadBlock(lts, datablocknum, (void *) lt->buffer);
			if (!lt->frozen)
				ltsReleaseBlock(lts, datablocknum);
			lt->nbytes = (lt->curBlockNumber < lt->numFullBlocks) ?
				BLCKSZ : lt->lastBlockBytes;
			if (lt->nbytes <= 0)
				break;			/* EOF (possible here?) */
		}

		nthistime = lt->nbytes - lt->pos;
		if (nthistime > size)
			nthistime = size;
		Assert(nthistime > 0);

		memcpy(ptr, lt->buffer + lt->pos, nthistime);

		lt->pos += nthistime;
		ptr = (void *) ((char *) ptr + nthistime);
		size -= nthistime;
		nread += nthistime;
	}

	return nread;
}

/*
 * "Freeze" the contents of a tape so that it can be read multiple times
 * and/or read backwards.  Once a tape is frozen, its contents will not
 * be released until the LogicalTapeSet is destroyed.  This is expected
 * to be used only for the final output pass of a merge.
 *
 * This *must* be called just at the end of a write pass, before the
 * tape is rewound (after rewind is too late!).  It performs a rewind
 * and switch to read mode "for free".	An immediately following rewind-
 * for-read call is OK but not necessary.
 */
void
LogicalTapeFreeze(LogicalTapeSet *lts, int tapenum)
{
	LogicalTape *lt;
	long		datablocknum;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];
	Assert(lt->writing);

	/*
	 * Completion of a write phase.  Flush last partial data block, flush any
	 * partial indirect blocks, rewind for nondestructive read.
	 */
	if (lt->dirty)
		ltsDumpBuffer(lts, lt);
	lt->lastBlockBytes = lt->nbytes;
	lt->writing = false;
	lt->frozen = true;
	datablocknum = ltsRewindIndirectBlock(lts, lt->indirect, true);
	/* Read the first block, or reset if tape is empty */
	lt->curBlockNumber = 0L;
	lt->pos = 0;
	lt->nbytes = 0;
	if (datablocknum != -1L)
	{
		ltsReadBlock(lts, datablocknum, (void *) lt->buffer);
		lt->nbytes = (lt->curBlockNumber < lt->numFullBlocks) ?
			BLCKSZ : lt->lastBlockBytes;
	}
}

/*
 * Backspace the tape a given number of bytes.	(We also support a more
 * general seek interface, see below.)
 *
 * *Only* a frozen-for-read tape can be backed up; we don't support
 * random access during write, and an unfrozen read tape may have
 * already discarded the desired data!
 *
 * Return value is TRUE if seek successful, FALSE if there isn't that much
 * data before the current point (in which case there's no state change).
 */
bool
LogicalTapeBackspace(LogicalTapeSet *lts, int tapenum, size_t size)
{
	LogicalTape *lt;
	long		nblocks;
	int			newpos;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];
	Assert(lt->frozen);

	/*
	 * Easy case for seek within current block.
	 */
	if (size <= (size_t) lt->pos)
	{
		lt->pos -= (int) size;
		return true;
	}

	/*
	 * Not-so-easy case.  Figure out whether it's possible at all.
	 */
	size -= (size_t) lt->pos;	/* part within this block */
	nblocks = size / BLCKSZ;
	size = size % BLCKSZ;
	if (size)
	{
		nblocks++;
		newpos = (int) (BLCKSZ - size);
	}
	else
		newpos = 0;
	if (nblocks > lt->curBlockNumber)
		return false;			/* a seek too far... */

	/*
	 * OK, we need to back up nblocks blocks.  This implementation would be
	 * pretty inefficient for long seeks, but we really aren't expecting that
	 * (a seek over one tuple is typical).
	 */
	while (nblocks-- > 0)
	{
		long		datablocknum = ltsRecallPrevBlockNum(lts, lt->indirect);

		if (datablocknum == -1L)
			elog(ERROR, "unexpected end of tape");
		lt->curBlockNumber--;
		if (nblocks == 0)
		{
			ltsReadBlock(lts, datablocknum, (void *) lt->buffer);
			lt->nbytes = BLCKSZ;
		}
	}
	lt->pos = newpos;
	return true;
}

/*
 * Seek to an arbitrary position in a logical tape.
 *
 * *Only* a frozen-for-read tape can be seeked.
 *
 * Return value is TRUE if seek successful, FALSE if there isn't that much
 * data in the tape (in which case there's no state change).
 */
bool
LogicalTapeSeek(LogicalTapeSet *lts, int tapenum,
				long blocknum, int offset)
{
	LogicalTape *lt;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];
	Assert(lt->frozen);
	Assert(offset >= 0 && offset <= BLCKSZ);

	/*
	 * Easy case for seek within current block.
	 */
	if (blocknum == lt->curBlockNumber && offset <= lt->nbytes)
	{
		lt->pos = offset;
		return true;
	}

	/*
	 * Not-so-easy case.  Figure out whether it's possible at all.
	 */
	if (blocknum < 0 || blocknum > lt->numFullBlocks ||
		(blocknum == lt->numFullBlocks && offset > lt->lastBlockBytes))
		return false;

	/*
	 * OK, advance or back up to the target block.	This implementation would
	 * be pretty inefficient for long seeks, but we really aren't expecting
	 * that (a seek over one tuple is typical).
	 */
	while (lt->curBlockNumber > blocknum)
	{
		long		datablocknum = ltsRecallPrevBlockNum(lts, lt->indirect);

		if (datablocknum == -1L)
			elog(ERROR, "unexpected end of tape");
		if (--lt->curBlockNumber == blocknum)
			ltsReadBlock(lts, datablocknum, (void *) lt->buffer);
	}
	while (lt->curBlockNumber < blocknum)
	{
		long		datablocknum = ltsRecallNextBlockNum(lts, lt->indirect,
														 lt->frozen);

		if (datablocknum == -1L)
			elog(ERROR, "unexpected end of tape");
		if (++lt->curBlockNumber == blocknum)
			ltsReadBlock(lts, datablocknum, (void *) lt->buffer);
	}
	lt->nbytes = (lt->curBlockNumber < lt->numFullBlocks) ?
		BLCKSZ : lt->lastBlockBytes;
	lt->pos = offset;
	return true;
}

/*
 * Obtain current position in a form suitable for a later LogicalTapeSeek.
 *
 * NOTE: it'd be OK to do this during write phase with intention of using
 * the position for a seek after freezing.	Not clear if anyone needs that.
 */
void
LogicalTapeTell(LogicalTapeSet *lts, int tapenum,
				long *blocknum, int *offset)
{
	LogicalTape *lt;

	Assert(tapenum >= 0 && tapenum < lts->nTapes);
	lt = lts->tapes[tapenum];
	*blocknum = lt->curBlockNumber;
	*offset = lt->pos;
}

/*
 * Obtain total disk space currently used by a LogicalTapeSet, in blocks.
 */
long
LogicalTapeSetBlocks(LogicalTapeSet *lts)
{
	return lts->nFileBlocks;
}