Front ported for CORE-5067 : Blocking new connections as a consequence of the too long sweep security2.fdb

src/jrd/jrd.cpp

@@ -6545,6 +6545,8 @@ bool JRD_shutdown_database(Database* dbb, const unsigned flags)
 #ifdef SUPERSERVER_V2
 	TRA_header_write(tdbb, dbb, 0);	// Update transaction info on header page.
 #endif
+	if (flags & SHUT_DBB_RELEASE_POOLS)
+		TRA_update_counters(tdbb, dbb);
 
 	VIO_fini(tdbb);
 

src/jrd/tpc.cpp

@@ -106,17 +106,24 @@ int TipCache::cacheState(thread_db* tdbb, TraNumber number)
 }
 
 
-TraNumber TipCache::findLimbo(thread_db* tdbb, TraNumber minNumber, TraNumber maxNumber)
+inline bool check_state(int state, ULONG mask)
+{
+	return ((1 << state) & mask) != 0;
+}
+
+TraNumber TipCache::findStates(thread_db* tdbb, TraNumber minNumber, TraNumber maxNumber, 
+	ULONG mask, int& state)
 {
 /**************************************
  *
- *	T P C _ f i n d _ l i m b o
+ *	T P C _ f i n d _ s t a t e s
  *
  **************************************
  *
  * Functional description
- *	Return the oldest limbo transaction in the given boundaries.
- *  If not found, return zero.
+ *	Return the oldest transaction in the given state. Lookup in the 
+ *  [min_number, max_number) bounds.
+ *  If not found, return zero and don't change value of "state".
  *
  **************************************/
 	SET_TDBB(tdbb);
@@ -129,29 +136,50 @@ TraNumber TipCache::findLimbo(thread_db* tdbb, TraNumber minNumber, TraNumber ma
 
 	initializeTpc(tdbb, maxNumber);
 
-	SyncLockGuard sync(&m_sync, SYNC_SHARED, "TipCache::findLimbo");
+	SyncLockGuard sync(&m_sync, SYNC_SHARED, "TipCache::findStates");
 
-	// All transactions older than the oldest in our TIP cache
-	// are known to be committed, so there's no point looking at them
+	const TxPage* tip_cache = m_cache.front();
 
-	TxPage* tip_cache = m_cache.front();
+	// Check for too old transactions (assumed committed)
 
 	if (maxNumber < tip_cache->tpc_base)
 		return 0;
 
-	if (minNumber < tip_cache->tpc_base)
+	if (minNumber < tip_cache->tpc_base || minNumber == 0)
+	{
+		if (check_state(tra_committed, mask))
+		{
+			state = tra_committed;
+			return minNumber;
+		}
+
 		minNumber = tip_cache->tpc_base;
+	}
+
+	bool check_precommitted = false;
+	if (check_state(tra_precommitted, mask))
+	{
+		// If we looking for tra_precommitted only and there is no precommitted
+		// transactions - return immediately
+
+		SyncLockGuard syncPC(&dbb->dbb_pc_sync, SYNC_SHARED, "TipCache::findLimbo");
+		if (dbb->dbb_pc_transactions != NULL)
+			check_precommitted = true;
+		else if (mask == (1 << tra_precommitted))
+			return 0;			
+	}
 
 	const ULONG trans_per_tip = m_dbb->dbb_page_manager.transPerTIP;
 	const TraNumber base = minNumber - minNumber % trans_per_tip;
 
-	// Scan the TIP cache and return the first (i.e. oldest) limbo transaction
+	// Scan the TIP cache and return the first (i.e. oldest) transaction in 
+	// state we are looking for
 
 	FB_SIZE_T pos;
 	if (m_cache.find(base, pos))
 	{
 		for (TraNumber number = minNumber;
-			 pos < m_cache.getCount() && number <= maxNumber;
+			 pos < m_cache.getCount() && number < maxNumber;
 			 pos++)
 		{
 			tip_cache = m_cache[pos];
@@ -159,11 +187,21 @@ TraNumber TipCache::findLimbo(thread_db* tdbb, TraNumber minNumber, TraNumber ma
 			fb_assert(number >= tip_cache->tpc_base);
 			fb_assert(tip_cache->tpc_base < MAX_TRA_NUMBER - trans_per_tip);
 
-			for (; number < (tip_cache->tpc_base + trans_per_tip) && number <= maxNumber;
+			for (; number < (tip_cache->tpc_base + trans_per_tip) && number < maxNumber;
 				 number++)
 			{
-				if (TRA_state(tip_cache->tpc_transactions, tip_cache->tpc_base, number) == tra_limbo)
+				if (check_precommitted && number && TRA_precommited(tdbb, number, number))
+				{
+					state = tra_precommitted;
 					return number;
+				}
+
+				const int tx_state = TRA_state(tip_cache->tpc_transactions, tip_cache->tpc_base, number);
+				if (check_state(tx_state, mask))
+				{
+					state = tx_state;
+					return number;
+				}
 			}
 		}
 	}

src/jrd/tpc_proto.h

@@ -43,7 +43,7 @@ public:
 	~TipCache();
 
 	int cacheState(thread_db*, TraNumber number);
-	TraNumber findLimbo(thread_db* tdbb, TraNumber minNumber, TraNumber maxNumber);
+	TraNumber findStates(thread_db* tdbb, TraNumber minNumber, TraNumber maxNumber, ULONG mask, int& state);
 	void initializeTpc(thread_db*, TraNumber number);
 	void setState(TraNumber number, SSHORT state);
 	int snapshotState(thread_db*, TraNumber number);
@@ -78,9 +78,10 @@ inline int TPC_cache_state(thread_db* tdbb, TraNumber number)
 	 return tdbb->getDatabase()->dbb_tip_cache->cacheState(tdbb, number);
 }
 
-inline TraNumber TPC_find_limbo(thread_db* tdbb, TraNumber minNumber, TraNumber maxNumber)
+inline TraNumber TPC_find_states(thread_db* tdbb, TraNumber minNumber, TraNumber maxNumber,
+	ULONG mask, int& state)
 {
-	 return tdbb->getDatabase()->dbb_tip_cache->findLimbo(tdbb, minNumber, maxNumber);
+	return tdbb->getDatabase()->dbb_tip_cache->findStates(tdbb, minNumber, maxNumber, mask, state);
 }
 
 inline void TPC_initialize_tpc(thread_db* tdbb, TraNumber number)

src/jrd/tra.cpp

@@ -797,6 +797,49 @@ void TRA_unlink_cursor(jrd_tra* transaction, DsqlCursor* cursor)
 }
 
 
+void TRA_update_counters(thread_db* tdbb, Database* dbb)
+{
+/**************************************
+*
+*	T R A _ u p d a t e _ c o u n t e r s
+*
+**************************************
+*
+* Functional description
+*  Update header page using cached values of transactions counters
+*
+**************************************/
+	SET_TDBB(tdbb);
+
+	if (!dbb || dbb->dbb_flags & DBB_read_only || dbb->dbb_flags & DBB_new ||
+		dbb->dbb_oldest_transaction == 0)
+	{
+		return;
+	}
+
+	WIN window(HEADER_PAGE_NUMBER);
+	header_page* header = (header_page*)CCH_FETCH(tdbb, &window, LCK_write, pag_header);
+
+	if (dbb->dbb_oldest_active > header->hdr_oldest_active ||
+		dbb->dbb_oldest_transaction > header->hdr_oldest_transaction ||
+		dbb->dbb_oldest_snapshot > header->hdr_oldest_snapshot)
+	{
+		CCH_MARK_MUST_WRITE(tdbb, &window);
+
+		if (dbb->dbb_oldest_active > header->hdr_oldest_active)
+			header->hdr_oldest_active = dbb->dbb_oldest_active;
+
+		if (dbb->dbb_oldest_transaction > header->hdr_oldest_transaction)
+			header->hdr_oldest_transaction = dbb->dbb_oldest_transaction;
+
+		if (dbb->dbb_oldest_snapshot > header->hdr_oldest_snapshot)
+			header->hdr_oldest_snapshot = dbb->dbb_oldest_snapshot;
+	}
+
+	CCH_RELEASE(tdbb, &window);
+}
+
+
 void TRA_post_resources(thread_db* tdbb, jrd_tra* transaction, ResourceList& resources)
 {
 /**************************************
@@ -1766,8 +1809,10 @@ void TRA_sweep(thread_db* tdbb)
 		// As our transaction is read-committed (see sweep_tpb), we have to scan
 		// the global TIP cache.
 
+		int oldest_state = 0;
 		const TraNumber oldest_limbo =
-			TPC_find_limbo(tdbb, transaction->tra_oldest, transaction->tra_top - 1);
+			TPC_find_states(tdbb, transaction->tra_oldest, transaction->tra_top - 1,
+							1 << tra_limbo, oldest_state);
 
 		const TraNumber active = oldest_limbo ? oldest_limbo : transaction->tra_top;
 
@@ -2526,6 +2571,8 @@ static void start_sweeper(thread_db* tdbb)
 	if (!dbb->allowSweepThread(tdbb))
 		return;
 
+	TRA_update_counters(tdbb, dbb);
+
 	// allocate space for the string and a null at the end
 	const char* pszFilename = tdbb->getAttachment()->att_filename.c_str();
 
@@ -3203,7 +3250,16 @@ static void transaction_start(thread_db* tdbb, jrd_tra* trans)
 	for (; active < number; active++)
 	{
 		if (trans->tra_flags & TRA_read_committed)
-			oldest_state = TPC_cache_state(tdbb, active);
+		{
+			const ULONG mask = (1 << tra_active);
+			active = TPC_find_states(tdbb, active, number, mask, oldest_state);
+			if (!active)
+			{
+				active = number;
+				break;
+			}
+			fb_assert(oldest_state == tra_active);
+		}
 		else
 		{
 			const ULONG byte = TRANS_OFFSET(active - base);
@@ -3293,7 +3349,16 @@ static void transaction_start(thread_db* tdbb, jrd_tra* trans)
 	for (oldest = trans->tra_oldest; oldest < number; oldest++)
 	{
 		if (trans->tra_flags & TRA_read_committed)
-			oldest_state = TPC_cache_state(tdbb, oldest);
+		{
+			const ULONG mask = ~((1 << tra_committed) | (1 << tra_precommitted));
+			oldest = TPC_find_states(tdbb, trans->tra_oldest, number, mask, oldest_state);
+			if (!oldest)
+			{
+				oldest = number;
+				break;
+			}
+			fb_assert(oldest_state != tra_committed && oldest_state != tra_precommitted);
+		}
 		else
 		{
 			const ULONG byte = TRANS_OFFSET(oldest - base);

src/jrd/tra_proto.h

@@ -60,6 +60,7 @@ Jrd::jrd_tra*	TRA_start(Jrd::thread_db* tdbb, ULONG flags, SSHORT lock_timeout,
 Jrd::jrd_tra*	TRA_start(Jrd::thread_db* tdbb, int, const UCHAR*, Jrd::jrd_tra* outer = NULL);
 int		TRA_state(const UCHAR*, TraNumber oldest, TraNumber number);
 void	TRA_sweep(Jrd::thread_db* tdbb);
+void	TRA_update_counters(Jrd::thread_db*, Jrd::Database*);
 int		TRA_wait(Jrd::thread_db* tdbb, Jrd::jrd_tra* trans, TraNumber number, Jrd::jrd_tra::wait_t wait);
 void	TRA_attach_request(Jrd::jrd_tra* transaction, Jrd::jrd_req* request);
 void	TRA_detach_request(Jrd::jrd_req* request);