Bug 31834278: Improve recovering from persistence by defering contentious maintenance tasks (merge 14.1.1 -> ce/14.1.1 @81396)

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v14.1.1.0/": change = 81397]

Author: harveyraja

prj/coherence-core/src/main/java/com/oracle/coherence/persistence/PersistentStore.java

@@ -8,6 +8,8 @@
 
 import com.oracle.coherence.common.base.Collector;
 
+import com.tangosol.util.NullImplementation;
+
 /**
  * PersistentStore represents a persistence facility to store and recover
  * key/value pairs. Each key-value pair is namespaced by a numeric "extent"
@@ -91,6 +93,22 @@ public interface PersistentStore<R>
      */
     public long[] extents();
 
+    /**
+     * Suggest to this PersistentStore that the caller requires exclusive access
+     * to this store until {@link AutoCloseable#close() close} is called on the
+     * returned {@link AutoCloseable}.
+     * <p>
+     * Note: the caller <b>must</b> call {@link AutoCloseable#close() close} on
+     *       the returned object
+     *
+     * @return an {@link AutoCloseable} object that <b>requires</b> close to be
+     *         called on it when exclusive access is no longer needed
+     */
+    public default AutoCloseable exclusively()
+        {
+        return NullImplementation.getAutoCloseable();
+        }
+
     // ----- store operations -----------------------------------------------
 
     /**

prj/coherence-core/src/main/java/com/tangosol/persistence/AbstractPersistenceManager.java

@@ -7,8 +7,8 @@
 package com.tangosol.persistence;
 
 import com.oracle.coherence.common.base.Collector;
-
 import com.oracle.coherence.common.base.Blocking;
+
 import com.oracle.coherence.persistence.AsyncPersistenceException;
 import com.oracle.coherence.persistence.ConcurrentAccessException;
 import com.oracle.coherence.persistence.FatalAccessException;
@@ -61,6 +61,7 @@
 
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
 
 /**
  * Abstract implementation of a ReadBuffer-based PersistentManager.
@@ -1122,6 +1123,16 @@ public long[] extents()
             return al;
             }
 
+        @Override
+        public AutoCloseable exclusively()
+            {
+            ensureReady();
+
+            lockWrite();
+
+            return instantiateExclusiveClosable();
+            }
+
         /**
          * {@inheritDoc}
          */
@@ -1371,7 +1382,7 @@ public void abort(Object oToken)
                     {
                     // guard against any unexpected throwable
                     CacheFactory.log("Caught an exception while aborting transaction for token \""
-                            + oToken + "\": " + printStackTrace(e), CacheFactory.LOG_ERR);
+                            + oToken + "\": " + printStackTrace(e), CacheFactory.LOG_QUIET);
                     }
                 }
             }
@@ -1648,6 +1659,28 @@ protected void setState(int nState)
                 }
             }
 
+        /**
+         * Return true if this store is in exclusive mode.
+         *
+         * @return true if this store is in exclusive mode
+         */
+        protected boolean isExclusive()
+            {
+            return ((WriteLock) f_lock.writeLock()).isHeldByCurrentThread();
+            }
+
+        /**
+         * Return an {@link AutoCloseable} that will transition this PersistentStore
+         * out of exclusive mode.
+         *
+         * @return an AutoCloseable that will transition this PersistentStore
+         *         out of exclusive mode
+         */
+        protected AutoCloseable instantiateExclusiveClosable()
+            {
+            return this::unlockWrite;
+            }
+
         // ----- Object methods ---------------------------------------------
 
         /**

prj/coherence-core/src/main/java/com/tangosol/persistence/SafePersistenceWrappers.java

@@ -825,6 +825,23 @@ public long[] extents()
             return super.extents();
             }
 
+        /**
+         * {@inheritDoc}
+         */
+        public AutoCloseable exclusively()
+            {
+            try
+                {
+                return getStore().exclusively();
+                }
+            catch (Throwable t)
+                {
+                onException((T) t);
+                }
+
+            return super.exclusively();
+            }
+
         /**
          * {@inheritDoc}
          */

prj/coherence-core/src/main/java/com/tangosol/persistence/bdb/BerkeleyDBManager.java

@@ -819,6 +819,7 @@ public void commitInternal(Object oToken)
                 {
                 throw ensurePersistenceException(e);
                 }
+            maintainEnvironment();
             }
 
         /**
@@ -835,6 +836,7 @@ public void abortInternal(Object oToken)
                 {
                 throw ensurePersistenceException(e);
                 }
+            maintainEnvironment();
             }
 
         // ----- helpers ----------------------------------------------------
@@ -1083,6 +1085,18 @@ protected Transaction ensureTransaction(Object oToken)
             throw new IllegalArgumentException("illegal token: " + oToken);
             }
 
+        @Override
+        protected AutoCloseable instantiateExclusiveClosable()
+            {
+            AutoCloseable parent = super.instantiateExclusiveClosable();
+            return () ->
+                {
+                parent.close();
+
+                maintainEnvironment();
+                };
+            }
+
         /**
          * Perform any necessary maintenance of the underlying BerkeleyDB
          * environment.
@@ -1094,38 +1108,41 @@ protected void maintainEnvironment(ReadBuffer bufKey, ReadBuffer bufValue)
             {
             if (MAINTENANCE_ENABLED)
                 {
-                // calculate the approximate number of bytes written to the
-                // environment by the current operation
-                int cbCurrent = bufKey.length();
-                if (bufValue != null)
-                    {
-                    cbCurrent += bufValue.length();
-                    }
-
-                // update the approximate running count of bytes written
-                long cbLast    = f_cbWritten.getAndAdd(cbCurrent);
-                long cbWritten = cbLast + cbCurrent;
+                // calculate the approximate number of bytes written to the environment
+                // by the current operation and update the running count
+                f_cbWritten.getAndAdd(bufKey.length() + (bufValue == null ? 0 : bufValue.length()));
+                }
+            }
 
+        /**
+         * Perform any necessary maintenance of the underlying BerkeleyDB
+         * environment.
+         */
+        protected void maintainEnvironment()
+            {
+            if (MAINTENANCE_ENABLED)
+                {
                 // determine if maintenance is already in progress
-                if (m_fMaintenanceScheduled)
+                if (isExclusive() || m_fMaintenanceScheduled)
                     {
                     // nothing to do
                     return;
                     }
 
+                long cbWritten = f_cbWritten.get();
+
                 // determine if it's time to checkpoint the environment
                 boolean fCheckpointRequired = m_cbLastCheckpoint >= CHECKPOINT_INTERVAL;
 
                 // determine if it's time to clean the environment
                 boolean fCleanRequired = m_cbLastClean >= CLEAN_INTERVAL;
 
                 // determine if it's time to compress the environment
-                boolean fCompressRequired = getSafeTimeMillis()
-                        >= m_ldtLastCompress + COMPRESS_INTERVAL;
+                boolean fCompressRequired = getSafeTimeMillis() >= m_ldtLastCompress + COMPRESS_INTERVAL;
 
                 // determine if it's time to update statistics
                 boolean fStatsRequired = f_cChecks.incrementAndGet() == STATS_CHECK_COUNT ||
-                        (cbWritten >= STATS_CHECK_BYTES && cbLast < STATS_CHECK_BYTES);
+                                         cbWritten >= STATS_CHECK_BYTES;
 
                 if (fCheckpointRequired || fCleanRequired || fCompressRequired || fStatsRequired)
                     {

prj/coherence-core/src/main/java/com/tangosol/util/NullImplementation.java

@@ -423,13 +423,23 @@ public static <R> PersistentStore<R> getPersistentStore(Class<R> clz)
     /**
     * Factory method: Obtain a null implementation of a {@link ResourceRegistry}.
     *
-    * @return an implementation of ResourceRegistry that does nothing.
+    * @return an implementation of ResourceRegistry that does nothing
     */
     public static ResourceRegistry getResourceRegistry()
         {
         return NullResourceRegistry.INSTANCE;
         }
 
+    /**
+    * Factory method: Obtain a null implementation of an {@link AutoCloseable}.
+    *
+    * @return an implementation of an AutoCloseable that does nothing
+    */
+    public static AutoCloseable getAutoCloseable()
+        {
+        return () -> {};
+        }
+
     // ----- inner classes ----------------------------------------------
 
     /**