xact.c

postgresql8.3.4源码,开源数据库

TimestampTz
GetCurrentTransactionStartTimestamp(void)
{
	return xactStartTimestamp;
}

/*
 *	GetCurrentStatementStartTimestamp
 */
TimestampTz
GetCurrentStatementStartTimestamp(void)
{
	return stmtStartTimestamp;
}

/*
 *	GetCurrentTransactionStopTimestamp
 *
 * We return current time if the transaction stop time hasn't been set
 * (which can happen if we decide we don't need to log an XLOG record).
 */
TimestampTz
GetCurrentTransactionStopTimestamp(void)
{
	if (xactStopTimestamp != 0)
		return xactStopTimestamp;
	return GetCurrentTimestamp();
}

/*
 *	SetCurrentStatementStartTimestamp
 */
void
SetCurrentStatementStartTimestamp(void)
{
	stmtStartTimestamp = GetCurrentTimestamp();
}

/*
 *	SetCurrentTransactionStopTimestamp
 */
static inline void
SetCurrentTransactionStopTimestamp(void)
{
	xactStopTimestamp = GetCurrentTimestamp();
}

/*
 *	GetCurrentTransactionNestLevel
 *
 * Note: this will return zero when not inside any transaction, one when
 * inside a top-level transaction, etc.
 */
int
GetCurrentTransactionNestLevel(void)
{
	TransactionState s = CurrentTransactionState;

	return s->nestingLevel;
}


/*
 *	TransactionIdIsCurrentTransactionId
 */
bool
TransactionIdIsCurrentTransactionId(TransactionId xid)
{
	TransactionState s;

	/*
	 * We always say that BootstrapTransactionId is "not my transaction ID"
	 * even when it is (ie, during bootstrap).	Along with the fact that
	 * transam.c always treats BootstrapTransactionId as already committed,
	 * this causes the tqual.c routines to see all tuples as committed, which
	 * is what we need during bootstrap.  (Bootstrap mode only inserts tuples,
	 * it never updates or deletes them, so all tuples can be presumed good
	 * immediately.)
	 *
	 * Likewise, InvalidTransactionId and FrozenTransactionId are certainly
	 * not my transaction ID, so we can just return "false" immediately for
	 * any non-normal XID.
	 */
	if (!TransactionIdIsNormal(xid))
		return false;

	/*
	 * We will return true for the Xid of the current subtransaction, any of
	 * its subcommitted children, any of its parents, or any of their
	 * previously subcommitted children.  However, a transaction being aborted
	 * is no longer "current", even though it may still have an entry on the
	 * state stack.
	 */
	for (s = CurrentTransactionState; s != NULL; s = s->parent)
	{
		int low, high;

		if (s->state == TRANS_ABORT)
			continue;
		if (!TransactionIdIsValid(s->transactionId))
			continue;			/* it can't have any child XIDs either */
		if (TransactionIdEquals(xid, s->transactionId))
			return true;
		/* As the childXids array is ordered, we can use binary search */
		low = 0;
		high = s->nChildXids - 1;
		while (low <= high)
		{
			int				middle;
			TransactionId	probe;

			middle = low + (high - low) / 2;
			probe = s->childXids[middle];
			if (TransactionIdEquals(probe, xid))
				return true;
			else if (TransactionIdPrecedes(probe, xid))
				low = middle + 1;
			else
				high = middle - 1;
		}
	}

	return false;
}


/*
 *	CommandCounterIncrement
 */
void
CommandCounterIncrement(void)
{
	/*
	 * If the current value of the command counter hasn't been "used" to
	 * mark tuples, we need not increment it, since there's no need to
	 * distinguish a read-only command from others.  This helps postpone
	 * command counter overflow, and keeps no-op CommandCounterIncrement
	 * operations cheap.
	 */
	if (currentCommandIdUsed)
	{
		currentCommandId += 1;
		if (currentCommandId == FirstCommandId)	/* check for overflow */
		{
			currentCommandId -= 1;
			ereport(ERROR,
					(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
		  errmsg("cannot have more than 2^32-1 commands in a transaction")));
		}
		currentCommandIdUsed = false;

		/* Propagate new command ID into static snapshots, if set */
		if (SerializableSnapshot)
			SerializableSnapshot->curcid = currentCommandId;
		if (LatestSnapshot)
			LatestSnapshot->curcid = currentCommandId;

		/*
		 * Make any catalog changes done by the just-completed command
		 * visible in the local syscache.  We obviously don't need to do
		 * this after a read-only command.  (But see hacks in inval.c
		 * to make real sure we don't think a command that queued inval
		 * messages was read-only.)
		 */
		AtCommit_LocalCache();
	}

	/*
	 * Make any other backends' catalog changes visible to me.
	 *
	 * XXX this is probably in the wrong place: CommandCounterIncrement
	 * should be purely a local operation, most likely.  However fooling
	 * with this will affect asynchronous cross-backend interactions,
	 * which doesn't seem like a wise thing to do in late beta, so save
	 * improving this for another day - tgl 2007-11-30
	 */
	AtStart_Cache();
}

/*
 * ForceSyncCommit
 *
 * Interface routine to allow commands to force a synchronous commit of the
 * current top-level transaction
 */
void
ForceSyncCommit(void)
{
	forceSyncCommit = true;
}


/* ----------------------------------------------------------------
 *						StartTransaction stuff
 * ----------------------------------------------------------------
 */

/*
 *	AtStart_Cache
 */
static void
AtStart_Cache(void)
{
	AcceptInvalidationMessages();
}

/*
 *	AtStart_Memory
 */
static void
AtStart_Memory(void)
{
	TransactionState s = CurrentTransactionState;

	/*
	 * If this is the first time through, create a private context for
	 * AbortTransaction to work in.  By reserving some space now, we can
	 * insulate AbortTransaction from out-of-memory scenarios.	Like
	 * ErrorContext, we set it up with slow growth rate and a nonzero minimum
	 * size, so that space will be reserved immediately.
	 */
	if (TransactionAbortContext == NULL)
		TransactionAbortContext =
			AllocSetContextCreate(TopMemoryContext,
								  "TransactionAbortContext",
								  32 * 1024,
								  32 * 1024,
								  32 * 1024);

	/*
	 * We shouldn't have a transaction context already.
	 */
	Assert(TopTransactionContext == NULL);

	/*
	 * Create a toplevel context for the transaction.
	 */
	TopTransactionContext =
		AllocSetContextCreate(TopMemoryContext,
							  "TopTransactionContext",
							  ALLOCSET_DEFAULT_MINSIZE,
							  ALLOCSET_DEFAULT_INITSIZE,
							  ALLOCSET_DEFAULT_MAXSIZE);

	/*
	 * In a top-level transaction, CurTransactionContext is the same as
	 * TopTransactionContext.
	 */
	CurTransactionContext = TopTransactionContext;
	s->curTransactionContext = CurTransactionContext;

	/* Make the CurTransactionContext active. */
	MemoryContextSwitchTo(CurTransactionContext);
}

/*
 *	AtStart_ResourceOwner
 */
static void
AtStart_ResourceOwner(void)
{
	TransactionState s = CurrentTransactionState;

	/*
	 * We shouldn't have a transaction resource owner already.
	 */
	Assert(TopTransactionResourceOwner == NULL);

	/*
	 * Create a toplevel resource owner for the transaction.
	 */
	s->curTransactionOwner = ResourceOwnerCreate(NULL, "TopTransaction");

	TopTransactionResourceOwner = s->curTransactionOwner;
	CurTransactionResourceOwner = s->curTransactionOwner;
	CurrentResourceOwner = s->curTransactionOwner;
}

/* ----------------------------------------------------------------
 *						StartSubTransaction stuff
 * ----------------------------------------------------------------
 */

/*
 * AtSubStart_Memory
 */
static void
AtSubStart_Memory(void)
{
	TransactionState s = CurrentTransactionState;

	Assert(CurTransactionContext != NULL);

	/*
	 * Create a CurTransactionContext, which will be used to hold data that
	 * survives subtransaction commit but disappears on subtransaction abort.
	 * We make it a child of the immediate parent's CurTransactionContext.
	 */
	CurTransactionContext = AllocSetContextCreate(CurTransactionContext,
												  "CurTransactionContext",
												  ALLOCSET_DEFAULT_MINSIZE,
												  ALLOCSET_DEFAULT_INITSIZE,
												  ALLOCSET_DEFAULT_MAXSIZE);
	s->curTransactionContext = CurTransactionContext;

	/* Make the CurTransactionContext active. */
	MemoryContextSwitchTo(CurTransactionContext);
}

/*
 * AtSubStart_ResourceOwner
 */
static void
AtSubStart_ResourceOwner(void)
{
	TransactionState s = CurrentTransactionState;

	Assert(s->parent != NULL);

	/*
	 * Create a resource owner for the subtransaction.	We make it a child of
	 * the immediate parent's resource owner.
	 */
	s->curTransactionOwner =
		ResourceOwnerCreate(s->parent->curTransactionOwner,
							"SubTransaction");

	CurTransactionResourceOwner = s->curTransactionOwner;
	CurrentResourceOwner = s->curTransactionOwner;
}

/* ----------------------------------------------------------------
 *						CommitTransaction stuff
 * ----------------------------------------------------------------
 */

/*
 *	RecordTransactionCommit
 *
 * Returns latest XID among xact and its children, or InvalidTransactionId
 * if the xact has no XID.	(We compute that here just because it's easier.)
 *
 * This is exported only to support an ugly hack in VACUUM FULL.
 */
TransactionId
RecordTransactionCommit(void)
{
	TransactionId xid = GetTopTransactionIdIfAny();
	bool		markXidCommitted = TransactionIdIsValid(xid);
	TransactionId latestXid = InvalidTransactionId;
	int			nrels;
	RelFileNode *rels;
	bool		haveNonTemp;
	int			nchildren;
	TransactionId *children;

	/* Get data needed for commit record */
	nrels = smgrGetPendingDeletes(true, &rels, &haveNonTemp);
	nchildren = xactGetCommittedChildren(&children);

	/*
	 * If we haven't been assigned an XID yet, we neither can, nor do we want
	 * to write a COMMIT record.
	 */
	if (!markXidCommitted)
	{
		/*
		 * We expect that every smgrscheduleunlink is followed by a catalog
		 * update, and hence XID assignment, so we shouldn't get here with any
		 * pending deletes.  Use a real test not just an Assert to check this,
		 * since it's a bit fragile.
		 */
		if (nrels != 0)
			elog(ERROR, "cannot commit a transaction that deleted files but has no xid");

		/* Can't have child XIDs either; AssignTransactionId enforces this */
		Assert(nchildren == 0);

		/*
		 * If we didn't create XLOG entries, we're done here; otherwise we
		 * should flush those entries the same as a commit record.	(An
		 * example of a possible record that wouldn't cause an XID to be
		 * assigned is a sequence advance record due to nextval() --- we want
		 * to flush that to disk before reporting commit.)
		 */
		if (XactLastRecEnd.xrecoff == 0)
			goto cleanup;
	}
	else
	{
		/*
		 * Begin commit critical section and insert the commit XLOG record.
		 */
		XLogRecData rdata[3];
		int			lastrdata = 0;
		xl_xact_commit xlrec;

		/* Tell bufmgr and smgr to prepare for commit */
		BufmgrCommit();

		/*
		 * Mark ourselves as within our "commit critical section".	This
		 * forces any concurrent checkpoint to wait until we've updated
		 * pg_clog.  Without this, it is possible for the checkpoint to set
		 * REDO after the XLOG record but fail to flush the pg_clog update to
		 * disk, leading to loss of the transaction commit if the system
		 * crashes a little later.
		 *
		 * Note: we could, but don't bother to, set this flag in
		 * RecordTransactionAbort.	That's because loss of a transaction abort
		 * is noncritical; the presumption would be that it aborted, anyway.
		 *
		 * It's safe to change the inCommit flag of our own backend without
		 * holding the ProcArrayLock, since we're the only one modifying it.
		 * This makes checkpoint's determination of which xacts are inCommit a
		 * bit fuzzy, but it doesn't matter.
		 */
		START_CRIT_SECTION();
		MyProc->inCommit = true;

		SetCurrentTransactionStopTimestamp();
		xlrec.xact_time = xactStopTimestamp;
		xlrec.nrels = nrels;
		xlrec.nsubxacts = nchildren;
		rdata[0].data = (char *) (&xlrec);
		rdata[0].len = MinSizeOfXactCommit;
		rdata[0].buffer = InvalidBuffer;
		/* dump rels to delete */
		if (nrels > 0)
		{
			rdata[0].next = &(rdata[1]);
			rdata[1].data = (char *) rels;
			rdata[1].len = nrels * sizeof(RelFileNode);
			rdata[1].buffer = InvalidBuffer;
			lastrdata = 1;
		}
		/* dump committed child Xids */
		if (nchildren > 0)
		{
			rdata[lastrdata].next = &(rdata[2]);
			rdata[2].data = (char *) children;
			rdata[2].len = nchildren * sizeof(TransactionId);
			rdata[2].buffer = InvalidBuffer;
			lastrdata = 2;
		}
		rdata[lastrdata].next = NULL;

		(void) XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT, rdata);
	}

	/*
	 * Check if we want to commit asynchronously.  If the user has set
	 * synchronous_commit = off, and we're not doing cleanup of any non-temp
	 * rels nor committing any command that wanted to force sync commit, then
	 * we can defer flushing XLOG.	(We must not allow asynchronous commit if
	 * there are any non-temp tables to be deleted, because we might delete
	 * the files before the COMMIT record is flushed to disk.  We do allow
	 * asynchronous commit if all to-be-deleted tables are temporary though,
	 * since they are lost anyway if we crash.)
	 */
	if (XactSyncCommit || forceSyncCommit || haveNonTemp)
	{
		/*
		 * Synchronous commit case.
		 *
		 * Sleep before flush! So we can flush more than one commit records
		 * per single fsync.  (The idea is some other backend may do the
		 * XLogFlush while we're sleeping.  This needs work still, because on
		 * most Unixen, the minimum select() delay is 10msec or more, which is
		 * way too long.)