Implemented CORE-3994: Improve the limbo transactions scan at the end of the sweep.

src/jrd/tpc.cpp

@@ -100,6 +100,63 @@ int TPC_cache_state(thread_db* tdbb, SLONG number)
 }
 
 
+SLONG TPC_find_limbo(Jrd::thread_db* tdbb, SLONG min_number, SLONG max_number)
+{
+/**************************************
+ *
+ *	T P C _ f i n d _ l i m b o
+ *
+ **************************************
+ *
+ * Functional description
+ *	Return the oldest limbo transaction in the given boundaries.
+ *  If not found, return zero.
+ *
+ **************************************/
+	SET_TDBB(tdbb);
+	Database* const dbb = tdbb->getDatabase();
+	CHECK_DBB(dbb);
+
+	fb_assert((ULONG) min_number <= (ULONG) max_number);
+
+	const SLONG trans_per_tip = dbb->dbb_page_manager.transPerTIP;
+
+	// Ensure that the TIP cache is extended to fit the requested transactions
+
+	TPC_initialize_tpc(tdbb, max_number);
+	const TxPageCache* tip_cache = dbb->dbb_tip_cache;
+
+	// All transactions older than the oldest in our TIP cache
+	// are known to be committed, so there's no point looking at them
+
+	if ((ULONG) max_number < (ULONG) tip_cache->tpc_base)
+		return 0;
+
+	if ((ULONG) min_number < (ULONG) tip_cache->tpc_base)
+		min_number = tip_cache->tpc_base;
+
+	// Scan the TIP cache and return the first (i.e. oldest) limbo transaction
+
+	for (ULONG number = min_number;
+		tip_cache && number <= (ULONG) max_number;
+		tip_cache = tip_cache->tpc_next)
+	{
+		if (number >= (ULONG) tip_cache->tpc_base)
+		{
+			for (; number < (ULONG) (tip_cache->tpc_base + trans_per_tip) &&
+				number <= (ULONG) max_number;
+				number++)
+			{
+				if (TRA_state(tip_cache->tpc_transactions, tip_cache->tpc_base, number) == tra_limbo)
+					return number;
+			}
+		}
+	}
+
+	return 0;
+}
+
+
 void TPC_initialize_tpc(thread_db* tdbb, SLONG number)
 {
 /**************************************

src/jrd/tpc_proto.h

@@ -25,6 +25,7 @@
 #define JRD_TPC_PROTO_H
 
 int		TPC_cache_state(Jrd::thread_db*, SLONG);
+SLONG	TPC_find_limbo(Jrd::thread_db*, SLONG, SLONG);
 void	TPC_initialize_tpc(Jrd::thread_db*, SLONG);
 void	TPC_set_state(Jrd::thread_db*, SLONG, SSHORT);
 int		TPC_snapshot_state(Jrd::thread_db*, SLONG);

src/jrd/tra.cpp

@@ -1802,25 +1802,16 @@ void TRA_sweep(thread_db* tdbb)
 
 	if (VIO_sweep(tdbb, transaction, &traceSweep))
 	{
-		const ULONG base = transaction->tra_oldest & ~TRA_MASK;
+		// At this point, we know that no record versions belonging to dead
-		ULONG active = transaction->tra_oldest;
+		// transactions remain anymore. However, there may still be limbo
-		for (; active < (ULONG) transaction->tra_top; active++)
+		// transactions, so we need to find the oldest one between tra_oldest and tra_top.
-		{
+		// As our transaction is read-committed (see sweep_tpb), we have to scan
-			if (transaction->tra_flags & TRA_read_committed)
+		// the global TIP cache.
-			{
+
-				if (TPC_cache_state(tdbb, active) == tra_limbo)
+		const SLONG oldest_limbo =
-					break;
+			TPC_find_limbo(tdbb, transaction->tra_oldest, transaction->tra_top - 1);
-			}
+
-			else
+		const SLONG active = oldest_limbo ? oldest_limbo : transaction->tra_top;
-			{
-				const ULONG byte = TRANS_OFFSET(active - base);
-				const USHORT shift = TRANS_SHIFT(active);
-				if (((transaction->tra_transactions[byte] >> shift) & TRA_MASK) == tra_limbo)
-				{
-					break;
-				}
-			}
-		}
 
 		// Flush page buffers to insure that no dangling records from
 		// dead transactions are left on-disk. This must be done before
@@ -1838,7 +1829,7 @@ void TRA_sweep(thread_db* tdbb)
 		if (header->hdr_oldest_transaction < --transaction_oldest_active)
 		{
 			CCH_MARK_MUST_WRITE(tdbb, &window);
-			header->hdr_oldest_transaction = MIN(active, (ULONG) transaction_oldest_active);
+			header->hdr_oldest_transaction = MIN((ULONG) active, (ULONG) transaction_oldest_active);
 		}
 
 		traceSweep.update(header);