Try if tombstone is eligable for pruning before locking on it's key (e…

…lastic#28767)

Pruning tombstones is quite expensive since we have to walk though all
deletes in the live version map and acquire a lock on every value even though
it's impossible to prune it. This change does a pre-check if a delete is old enough
and if not it skips acquireing the lock.

server/src/main/java/org/elasticsearch/index/engine/LiveVersionMap.java

@@ -375,25 +375,31 @@ void removeTombstoneUnderLock(BytesRef uid) {
         }
     }
 
+    private boolean canRemoveTombstone(long currentTime, long pruneInterval, DeleteVersionValue versionValue) {
+        // check if the value is old enough to be removed
+        final boolean isTooOld = currentTime - versionValue.time > pruneInterval;
+        // version value can't be removed it's
+        // not yet flushed to lucene ie. it's part of this current maps object
+        final boolean isNotTrackedByCurrentMaps = versionValue.time < maps.getMinDeleteTimestamp();
+        return isTooOld && isNotTrackedByCurrentMaps;
+    }
+
     void pruneTombstones(long currentTime, long pruneInterval) {
         for (Map.Entry<BytesRef, DeleteVersionValue> entry : tombstones.entrySet()) {
-            final BytesRef uid = entry.getKey();
-            try (Releasable lock = keyedLock.tryAcquire(uid)) {
-                // we use tryAcquire here since this is a best effort and we try to be least disruptive
-                // this method is also called under lock in the engine under certain situations such that this can lead to deadlocks
-                // if we do use a blocking acquire. see #28714
-                if (lock != null) { // did we get the lock?
-                    // can we do it without this lock on each value? maybe batch to a set and get the lock once per set?
-                    // Must re-get it here, vs using entry.getValue(), in case the uid was indexed/deleted since we pulled the iterator:
-                    final DeleteVersionValue versionValue = tombstones.get(uid);
-                    if (versionValue != null) {
-                        // check if the value is old enough to be removed
-                        final boolean isTooOld = currentTime - versionValue.time > pruneInterval;
-                        if (isTooOld) {
-                            // version value can't be removed it's
-                            // not yet flushed to lucene ie. it's part of this current maps object
-                            final boolean isNotTrackedByCurrentMaps = versionValue.time < maps.getMinDeleteTimestamp();
-                            if (isNotTrackedByCurrentMaps) {
+            // we do check before we actually lock the key - this way we don't need to acquire the lock for tombstones that are not
+            // prune-able. If the tombstone changes concurrently we will re-read and step out below since if we can't collect it now w
+            // we won't collect the tombstone below since it must be newer than this one.
+            if (canRemoveTombstone(currentTime, pruneInterval, entry.getValue())) {
+                final BytesRef uid = entry.getKey();
+                try (Releasable lock = keyedLock.tryAcquire(uid)) {
+                    // we use tryAcquire here since this is a best effort and we try to be least disruptive
+                    // this method is also called under lock in the engine under certain situations such that this can lead to deadlocks
+                    // if we do use a blocking acquire. see #28714
+                    if (lock != null) { // did we get the lock?
+                        // Must re-get it here, vs using entry.getValue(), in case the uid was indexed/deleted since we pulled the iterator:
+                        final DeleteVersionValue versionValue = tombstones.get(uid);
+                        if (versionValue != null) {
+                            if (canRemoveTombstone(currentTime, pruneInterval, versionValue)) {
                                 removeTombstoneUnderLock(uid);
                             }
                         }