postgresql源码学习（六）—— 回滚（中止）与清理事务_nodejs pgsql回滚

一、 AbortTransaction

1. gdb测试

会话1

会话2

2. 具体代码与跟踪

调用栈如下

static void
AbortTransaction(void)
{
	TransactionState s = CurrentTransactionState;
	TransactionId latestXid;
	bool		is_parallel_worker;
 
	/* Prevent cancel/die interrupt while cleaning up，清理期间避免被中断 */
	HOLD_INTERRUPTS();
 
	/* Make sure we have a valid memory context and resource owner，确认我们有有效的内存上下文和资源管理器 */
	AtAbort_Memory();
	AtAbort_ResourceOwner();
 
	/*
	 * 释放所有轻量锁，而常规锁则在完成abort后才释放
	 */
	LWLockReleaseAll();
 
	/* Clear wait information and command progress indicator，清理等待信息和命令进度指示器 */
	pgstat_report_wait_end();
	pgstat_progress_end_command();
 
	/* Clean up buffer I/O and buffer context locks, too，清理buffer I/O和buffer上下文锁  */
	AbortBufferIO();
	UnlockBuffers();
 
	/* Reset WAL record construction state，重置WAL记录结构状态 */
	XLogResetInsertion();
 
	/* Cancel condition variable sleep，取消条件变量sleep */
	ConditionVariableCancelSleep();
 
	/*
	 * Also clean up any open wait for lock, since the lock manager will choke
	 * if we try to wait for another lock before doing this.清理所有在等待的锁，如果在这步前有在等待的锁，锁管理器将会抑制它
	 */
	LockErrorCleanup();
 
	/*
	 * If any timeout events are still active, make sure the timeout interrupt
	 * is scheduled.  This covers possible loss of a timeout interrupt due to
	 * longjmp'ing out of the SIGINT handler (see notes in handle_sig_alarm).
	 * We delay this till after LockErrorCleanup so that we don't uselessly
	 * reschedule lock or deadlock check timeouts. 处理超时事件
	 */
	reschedule_timeouts();
 
	/*
	 * Re-enable signals, in case we got here by longjmp'ing out of a signal
	 * handler.  We do this fairly early in the sequence so that the timeout
	 * infrastructure will be functional if needed while aborting.
	 */
	PG_SETMASK(&UnBlockSig);

/*
	 * check the current transaction state，检查当前事务状态
	 */
	is_parallel_worker = (s->blockState == TBLOCK_PARALLEL_INPROGRESS);
	if (s->state != TRANS_INPROGRESS && s->state != TRANS_PREPARE)
		elog(WARNING, "AbortTransaction while in %s state",
			 TransStateAsString(s->state));
	Assert(s->parent == NULL);
 
	/*
	 * set the current transaction state information appropriately during the
	 * abort processing，状态设置
	 */
	s->state = TRANS_ABORT;

/*
	 * Reset user ID which might have been changed transiently.  We need this
	 * to clean up in case control escaped out of a SECURITY DEFINER function
	 * or other local change of CurrentUserId; therefore, the prior value of
	 * SecurityRestrictionContext also needs to be restored.
	 *
	 * (Note: it is not necessary to restore session authorization or role
	 * settings here because those can only be changed via GUC, and GUC will
	 * take care of rolling them back if need be.)
     * 设置用户id和安全上下文
	 */
	SetUserIdAndSecContext(s->prevUser, s->prevSecContext);
 
	/* Forget about any active REINDEX.，取消所有活跃的reindex操作 */
	ResetReindexState(s->nestingLevel);
 
	/* Reset logical streaming state.，重置逻辑stream状态 */
	ResetLogicalStreamingState();

   

/* If in parallel mode, clean up workers and exit parallel mode.，如果是并行模式，清理并退出并行模式 */
	if (IsInParallelMode())
	{
		AtEOXact_Parallel(false);
		s->parallelModeLevel = 0;
	}
 
	/*
	 * do abort processing，
	 */
	AfterTriggerEndXact(false); /* 'false' means it's abort */
	AtAbort_Portals();
	smgrDoPendingSyncs(false, is_parallel_worker);
	AtEOXact_LargeObject(false);
	AtAbort_Notify();
	AtEOXact_RelationMap(false, is_parallel_worker);
	AtAbort_Twophase();

/*
	 * Advertise the fact that we aborted in pg_xact (assuming that we got as
	 * far as assigning an XID to advertise).  But if we're inside a parallel
	 * worker, skip this; the user backend must be the one to write the abort record.
	 *	非并行模式，要将abort操作记录到XLOG日志 
*/
	if (!is_parallel_worker)
		latestXid = RecordTransactionAbort(false);
	else
	{
		latestXid = InvalidTransactionId;
 
		/*
		 * Since the parallel leader won't get our value of XactLastRecEnd in
		 * this case, we nudge WAL-writer ourselves in this case.  See related
		 * comments in RecordTransactionAbort for why this matters.
		 */
		XLogSetAsyncXactLSN(XactLastRecEnd);
	}
 
	TRACE_POSTGRESQL_TRANSACTION_ABORT(MyProc->lxid);
 
	/*
	 * Let others know about no transaction in progress by me. Note that this
	 * must be done _before_ releasing locks we hold and _after_
	 * RecordTransactionAbort.
	 */
	ProcArrayEndTransaction(MyProc, latestXid);

   

/*
	 * Post-abort cleanup.  See notes in CommitTransaction() concerning
	 * ordering.  We can skip all of it if the transaction failed before
	 * creating a resource owner.
	 */
	if (TopTransactionResourceOwner != NULL)
	{
		if (is_parallel_worker)
			CallXactCallbacks(XACT_EVENT_PARALLEL_ABORT);
		else
			CallXactCallbacks(XACT_EVENT_ABORT);
 
		ResourceOwnerRelease(TopTransactionResourceOwner,
							 RESOURCE_RELEASE_BEFORE_LOCKS,
							 false, true);
		AtEOXact_Buffers(false);
		AtEOXact_RelationCache(false);
		AtEOXact_Inval(false);
		AtEOXact_MultiXact();
		ResourceOwnerRelease(TopTransactionResourceOwner,
							 RESOURCE_RELEASE_LOCKS,
							 false, true);
		ResourceOwnerRelease(TopTransactionResourceOwner,
							 RESOURCE_RELEASE_AFTER_LOCKS,
							 false, true);
		smgrDoPendingDeletes(false);
 
		AtEOXact_GUC(false, 1);
		AtEOXact_SPI(false);
		AtEOXact_Enum();
		AtEOXact_on_commit_actions(false);
		AtEOXact_Namespace(false, is_parallel_worker);
		AtEOXact_SMgr();
		AtEOXact_Files(false);
		AtEOXact_ComboCid();
		AtEOXact_HashTables(false);
		AtEOXact_PgStat(false, is_parallel_worker);
		AtEOXact_ApplyLauncher(false);
		pgstat_report_xact_timestamp(0);
	}
	/*
	 * State remains TRANS_ABORT until CleanupTransaction().
	 */
	RESUME_INTERRUPTS();
}

3. 主要流程图

二、 CleanupTransaction

• 主要功能：释放事务所占用的内存资源
• 与AbortTransaction的区别：该函数是终止事务退出时最后调用的函数，做最后的实际清理工作。而AbortTransaction没有实际释放相关资源，只是切换一些资源的状态，使其能够被其他事务获得。

1. gdb测试

会话1

会话2

2. 具体代码与跟踪

调用栈如下

static void
CleanupTransaction(void)
{
	TransactionState s = CurrentTransactionState;
 
	/*
	 * State should still be TRANS_ABORT from AbortTransaction().事务状态应该为TRANS_ABORT，因为本函数在AbortTransaction()之后调用
	 */
	if (s->state != TRANS_ABORT)
		elog(FATAL, "CleanupTransaction: unexpected state %s",
			 TransStateAsString(s->state));
 
	/*
	 * do abort cleanup processing
	 */
	AtCleanup_Portals();		/* now safe to release portal memory，安全地释放portal内存 */
	AtEOXact_Snapshot(false, true); /* and release the transaction's snapshots，释放事务快照 */
 
	CurrentResourceOwner = NULL;	/* and resource owner */
	if (TopTransactionResourceOwner)
		ResourceOwnerDelete(TopTransactionResourceOwner);
	s->curTransactionOwner = NULL;
	CurTransactionResourceOwner = NULL;
	TopTransactionResourceOwner = NULL;
 
	AtCleanup_Memory();			/* and transaction memory */

s->fullTransactionId = InvalidFullTransactionId;
	s->subTransactionId = InvalidSubTransactionId;
	s->nestingLevel = 0;
	s->gucNestLevel = 0;
	s->childXids = NULL;
	s->nChildXids = 0;
	s->maxChildXids = 0;
	s->parallelModeLevel = 0;
 
	XactTopFullTransactionId = InvalidFullTransactionId;
	nParallelCurrentXids = 0;
 
	/*
	 * done with abort processing, set current transaction state back to
	 * default
	 */
	s->state = TRANS_DEFAULT;
}

3. 主要流程图

 

参考：

《PostgreSQL技术内幕：事务处理深度探索》第1章

《PostgreSQL数据库内核分析》第7章