diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 0555b02a8d..d15c44b5b7 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -338,9 +338,9 @@ static void DisplayXidCache(void); static void KnownAssignedXidsCompress(bool force); static void KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid, bool exclusive_lock); -static bool KnownAssignedXidsSearch(TransactionId xid, bool remove); +static int KnownAssignedXidsSearch(TransactionId xid, bool remove, int start); static bool KnownAssignedXidExists(TransactionId xid); -static void KnownAssignedXidsRemove(TransactionId xid); +static int KnownAssignedXidsRemove(TransactionId xid, int start); static void KnownAssignedXidsRemoveTree(TransactionId xid, int nsubxids, TransactionId *subxids); static void KnownAssignedXidsRemovePreceding(TransactionId xid); @@ -4771,6 +4771,9 @@ KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid, } } +#define KAX_INVALID -1 +#define KAX_START KAX_INVALID + /* * KnownAssignedXidsSearch * @@ -4780,21 +4783,30 @@ KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid, * Caller must hold ProcArrayLock in shared or exclusive mode. * Exclusive lock must be held for remove = true. */ -static bool -KnownAssignedXidsSearch(TransactionId xid, bool remove) +static int +KnownAssignedXidsSearch(TransactionId xid, bool remove, int start) { ProcArrayStruct *pArray = procArray; int first, last; int head; int tail; - int result_index = -1; + int result_index = KAX_INVALID; if (remove) { /* we hold ProcArrayLock exclusively, so no need for spinlock */ - tail = pArray->tailKnownAssignedXids; head = pArray->headKnownAssignedXids; + + /* + * Start at the tail, or if we are removing a set of sorted + * xids, we can set the tail to the location of previous removal, + * reducing search time when N is large. + */ + if (start == KAX_START) + tail = pArray->tailKnownAssignedXids; + else + tail = start; } else { @@ -4831,10 +4843,10 @@ KnownAssignedXidsSearch(TransactionId xid, bool remove) } if (result_index < 0) - return false; /* not in array */ + return KAX_INVALID; /* not in array */ if (!KnownAssignedXidsValid[result_index]) - return false; /* in array, but invalid */ + return KAX_INVALID; /* in array, but invalid */ if (remove) { @@ -4865,7 +4877,7 @@ KnownAssignedXidsSearch(TransactionId xid, bool remove) } } - return true; + return result_index; } /* @@ -4878,16 +4890,18 @@ KnownAssignedXidExists(TransactionId xid) { Assert(TransactionIdIsValid(xid)); - return KnownAssignedXidsSearch(xid, false); + return KnownAssignedXidsSearch(xid, false, KAX_START) == KAX_INVALID ? false : true; } /* * Remove the specified XID from KnownAssignedXids[]. * * Caller must hold ProcArrayLock in exclusive mode. + * + * Returns the array offset of the removed element, or KAX_INVALID if not present. */ -static void -KnownAssignedXidsRemove(TransactionId xid) +static int +KnownAssignedXidsRemove(TransactionId xid, int start) { Assert(TransactionIdIsValid(xid)); @@ -4903,7 +4917,7 @@ KnownAssignedXidsRemove(TransactionId xid) * actual errors, but it would be complicated and probably not worth it. * So, just ignore the search result. */ - (void) KnownAssignedXidsSearch(xid, true); + return KnownAssignedXidsSearch(xid, true, start); } /* @@ -4911,18 +4925,23 @@ KnownAssignedXidsRemove(TransactionId xid) * Remove xid (if it's not InvalidTransactionId) and all the subxids. * * Caller must hold ProcArrayLock in exclusive mode. + * + * Since the xid and subxids are sorted, we can chain them together so + * we start searching for next xid at the point where we removed the previous + * xid in the tree. First search starts at KAX_START. */ static void KnownAssignedXidsRemoveTree(TransactionId xid, int nsubxids, TransactionId *subxids) { int i; + int start = KAX_START; if (TransactionIdIsValid(xid)) - KnownAssignedXidsRemove(xid); + start = KnownAssignedXidsRemove(xid, start); for (i = 0; i < nsubxids; i++) - KnownAssignedXidsRemove(subxids[i]); + start = KnownAssignedXidsRemove(subxids[i], start); /* Opportunistically compress the array */ KnownAssignedXidsCompress(false);