Merge lp:~pbeaman/akiban-persistit/fix-wait-for-durability into lp:akiban-persistit

[Peter Beaman]

Modifies the JournalManager#waitForDurability loop to improve performance on 4-hour TPCC test and adds several smaller changes to fix problems seen while testing 4-hour TPCC extensively on perf02.

* Modify BufferPool preload warmup logging to show elapsed time

* CleanupManager no longer calls pruneObsoleteTransactions and other periodic maintenance actions on every polling cycle (wastes CPU)

* JournalManager: added UrgentFileCountThreshold value settable from MXBean to control journal file count. Revised the urgency() method and friends to work more smoothly with the settable threshold.

* JournalManager: Move the calls to force Volume updates back into the copier thread because waiting until Checkpoints causes massive I/O

* JournalManager: Remove ReentrantReadWriteLock from waitForDurability. Turned out that wasted a great deal of CPU with little benefit. Modified to a purely polling protocol and ensured every path through the loop sleeps for at least one ms. Removed detection of loops without waits and the corresponding tests since by inspection there is a sleep on every conditional branch.

* JournalManager: Corrected time interval calculations in waitForDurability

* JournalManager: Changed calculation of expected IO time to use average rather than max. Actual behavior is very erratic, and our best effort to delay commit() until completion of a force cycle is very inaccurate; we may want to remove the behavior and certainly should modify or remove the documentation.

* JournalManager: Relaxed the leadTime and flusher poll times from 100ms to 500ms. (This change may no longer be necessary; will test and revise as appropriate.)

* Persistit: Corrected a problem in copyBackPages() in which the relaxed pruneObsoleteTransactions() cycle emerged as a race that broke unit tests.

[Peter Beaman (pbeaman)]

Pushed a couple of additional changes inspired by observing tests:

- cleanupPageList is now O(N) rather O(N*N)
- non-transactions store/remove operations now throttle
- the throttle() method itself is now faster - the throttle interval is computed every few seconds by the JOURNAL_COPIER thread.
- a couple of minor code cleanups

[Nathan Williams (nwilliams)]

This look pretty good.

Just a few questions:
IOMeter#DEFAULT_QUIESCENT_IO_THRESHOLD was increased by 1000. The usage of it in JournalManager was also scaled by 1024. I'm not entirely what that is used for, can you elaborate?

JournalManager#selectForCopy() went from asserting the pageNodes were valid to checking it. Can invalid nodes be in the list now?

Minor:
Probably just my preference, but constants with units in the name make one less thing to check (new intervals in CleanupManager).

review: Needs Information

[Peter Beaman (pbeaman)]

Thanks, good questions.

DEFAULT_QUIESCENT_IO_THRESHOLD is used to try to detect that little
I/O is currently being done and when that happens the copier interval
goes down. I'm not wedded to that heuristic, but it has been there
since the beginning. The change was to normalize the units in IOMeter.
 The IORate field attempts to show KBytes per second, whereas the
threshold used to be in Bytes per second. Now the threshold is
expressed in KBytes per second too. Clearly there's a neglected
documentation task - I will fix that.

selectNodesForCopy used to call cleanupPageList before looping, and
the assert made sense. However, because the cleanup operation can be
expensive (even with the new algorithm) I reduced the number of number
of times it is called, and therefore the selectNodesForCopy method can
now see and skip invalid PageNodes.

I agree about units in the constant names and will change.

On Mon, Sep 10, 2012 at 10:29 AM, Nathan Williams <email address hidden> wrote:
> Review: Needs Information
>
> This look pretty good.
>
> Just a few questions:
> IOMeter#DEFAULT_QUIESCENT_IO_THRESHOLD was increased by 1000. The usage of it in JournalManager was also scaled by 1024. I'm not entirely what that is used for, can you elaborate?
>
> JournalManager#selectForCopy() went from asserting the pageNodes were valid to checking it. Can invalid nodes be in the list now?
>
> Minor:
> Probably just my preference, but constants with units in the name make one less thing to check (new intervals in CleanupManager).
> --
> https://code.launchpad.net/~pbeaman/akiban-persistit/fix-wait-for-durability/+merge/123174
> You are the owner of lp:~pbeaman/akiban-persistit/fix-wait-for-durability.

[Peter Beaman (pbeaman)]

Revised as requested. The CleanupManager time constants are name named XXX_MS as are a number of other constants.

From original proposal reverted the changes in JOURNAL_FLUSHER poll time and SOFT commit lead times to their former values. I believe these changes didn't help and are not needed for good performance and want to benchmark again with the original values.

[Nathan Williams (nwilliams)]

Thanks for the tweaks. Looks good to me.

review: Approve

[Akiban Build User (build-akiban)]

There was one failure during build/test:

* unknown exception (check log)

review: Needs Fixing

[Peter Beaman (pbeaman)]

Jenkins glitch. Reapproving.

=== modified file 'src/main/java/com/persistit/BufferPool.java'
--- src/main/java/com/persistit/BufferPool.java	2012-08-24 14:00:17 +0000
+++ src/main/java/com/persistit/BufferPool.java	2012-09-12 22:04:20 +0000
@@ -97,10 +97,11 @@
      */
     private final static int INVENTORY_VERSIONS = 3;
 
+    private final static long NS_PER_SEC = 1000000000;
     /**
-     * Preload log multiple
+     * Preload log message interval, in seconds
      */
-    private final static int INVENTORY_PRELOAD_LOG_MESSAGE_MULTIPLE = 10000;
+    private final static long INVENTORY_PRELOAD_LOG_MESSAGE_NS = 60L * NS_PER_SEC;
 
     /**
      * The Persistit instance that references this BufferPool.
@@ -1451,6 +1452,9 @@
     void preloadBufferInventory() {
         int count = 0;
         int total = 0;
+        final long startTime = System.nanoTime();
+        long reportTime = startTime;
+
         try {
             final JournalManager jman = _persistit.getJournalManager();
             final Exchange exchange = getBufferInventoryExchange();
@@ -1496,8 +1500,11 @@
                     final Buffer buff = get(vol, pn.getPageAddress(), false, true);
                     buff.release();
                     count++;
-                    if ((count % INVENTORY_PRELOAD_LOG_MESSAGE_MULTIPLE) == 0) {
-                        _persistit.getLogBase().bufferInventoryProgress.log(count, total);
+                    final long now = System.nanoTime();
+                    if (now - reportTime >= INVENTORY_PRELOAD_LOG_MESSAGE_NS) {
+                        _persistit.getLogBase().bufferInventoryProgress.log(count, total, (now - reportTime)
+                                / NS_PER_SEC);
+                        reportTime = now;
                     }
                     if (count >= _bufferCount) {
                         //
@@ -1513,7 +1520,8 @@
         } catch (final PersistitException e) {
             _persistit.getLogBase().bufferInventoryException.log(e);
         } finally {
-            _persistit.getLogBase().bufferInventoryProgress.log(count, total);
+            final long now = System.nanoTime();
+            _persistit.getLogBase().bufferInventoryProgress.log(count, total, (now - reportTime) / NS_PER_SEC);
         }
     }
 

=== modified file 'src/main/java/com/persistit/CleanupManager.java'
--- src/main/java/com/persistit/CleanupManager.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/CleanupManager.java	2012-09-12 22:04:20 +0000
@@ -36,15 +36,19 @@
         void performAction(Persistit persistit) throws PersistitException;
     }
 
-    final static long DEFAULT_CLEANUP_INTERVAL = 1000;
+    final static long DEFAULT_CLEANUP_INTERVAL_MS = 1000;
 
     final static int DEFAULT_QUEUE_SIZE = 50000;
 
-    final static int WORKLIST_LENGTH = 500;
-
-    private final static long DEFAULT_MINIMUM_PRUNING_DELAY = 1000;
-
-    final Queue<CleanupAction> _cleanupActionQueue = new ArrayBlockingQueue<CleanupAction>(DEFAULT_QUEUE_SIZE);
+    private final static int WORKLIST_LENGTH = 500;
+
+    private final static long MINIMUM_MAINTENANCE_INTERVAL_NS = 1000000000L;
+
+    private final static long MINIMUM_PRUNE_OBSOLETE_TRANSACTIONS_INTERVAL_NS = 50000000000L;
+
+    private final static long DEFAULT_MINIMUM_PRUNING_DELAY_NS = 1000;
+
+    private final Queue<CleanupAction> _cleanupActionQueue = new ArrayBlockingQueue<CleanupAction>(DEFAULT_QUEUE_SIZE);
 
     private final AtomicBoolean _closed = new AtomicBoolean();
 
@@ -56,7 +60,11 @@
 
     private final AtomicLong _errors = new AtomicLong();
 
-    private final AtomicLong _minimumPruningDelay = new AtomicLong(DEFAULT_MINIMUM_PRUNING_DELAY);
+    private final AtomicLong _minimumPruningDelay = new AtomicLong(DEFAULT_MINIMUM_PRUNING_DELAY_NS);
+
+    private long _lastMaintenance;
+
+    private long _lastPruneObsoleteTransactions;
 
     CleanupManager(final Persistit persistit) {
         super(persistit);
@@ -64,7 +72,10 @@
 
     public void start() {
         _closed.set(false);
-        start("CLEANUP_MANAGER", DEFAULT_CLEANUP_INTERVAL);
+        final long now = System.nanoTime();
+        _lastMaintenance = now;
+        _lastPruneObsoleteTransactions = now;
+        start("CLEANUP_MANAGER", DEFAULT_CLEANUP_INTERVAL_MS);
     }
 
     public void close(final boolean flush) throws PersistitException {
@@ -138,9 +149,19 @@
 
     @Override
     public void poll() throws Exception {
-        _persistit.getIOMeter().poll();
-        _persistit.cleanup();
-        _persistit.getJournalManager().pruneObsoleteTransactions();
+
+        final long now = System.nanoTime();
+        if (now - _lastMaintenance > MINIMUM_MAINTENANCE_INTERVAL_NS) {
+            _persistit.getIOMeter().poll();
+            _persistit.cleanup();
+            _lastMaintenance = now;
+        }
+
+        if (now - _lastPruneObsoleteTransactions > MINIMUM_PRUNE_OBSOLETE_TRANSACTIONS_INTERVAL_NS) {
+            _persistit.getJournalManager().pruneObsoleteTransactions();
+            _lastPruneObsoleteTransactions = now;
+        }
+
         final List<CleanupAction> workList = new ArrayList<CleanupAction>(WORKLIST_LENGTH);
         synchronized (this) {
             while (workList.size() < WORKLIST_LENGTH) {

=== modified file 'src/main/java/com/persistit/Exchange.java'
--- src/main/java/com/persistit/Exchange.java	2012-08-28 14:29:50 +0000
+++ src/main/java/com/persistit/Exchange.java	2012-09-12 22:04:20 +0000
@@ -1290,6 +1290,9 @@
         if (!isDirectoryExchange()) {
             _persistit.checkSuspended();
         }
+        if (!_ignoreTransactions && !_transaction.isActive()) {
+            _persistit.getJournalManager().throttle();
+        }
         // TODO: directoryExchange, and lots of tests, don't use transactions.
         // Skip MVCC for now.
         int options = StoreOptions.WAIT;
@@ -2990,13 +2993,9 @@
     }
 
     private boolean removeInternal(final Direction selection, final boolean fetchFirst) throws PersistitException {
-        assertCorrectThread(true);
-        _persistit.checkClosed();
-
         if (selection != EQ && selection != GTEQ && selection != GT) {
             throw new IllegalArgumentException("Invalid mode " + selection);
         }
-
         final int keySize = _key.getEncodedSize();
 
         _key.copyTo(_spareKey3);
@@ -3005,6 +3004,7 @@
         // Special case for empty key
         if (keySize == 0) {
             if (selection == EQ) {
+                assertCorrectThread(true);
                 return false;
             }
             _spareKey3.append(BEFORE);
@@ -3094,6 +3094,13 @@
         assertCorrectThread(true);
         _persistit.checkClosed();
 
+        if (!isDirectoryExchange()) {
+            _persistit.checkSuspended();
+        }
+        if (!_ignoreTransactions && !_transaction.isActive()) {
+            _persistit.getJournalManager().throttle();
+        }
+
         if (_ignoreTransactions || !_transaction.isActive()) {
             return raw_removeKeyRangeInternal(key1, key2, fetchFirst, false);
         }

=== modified file 'src/main/java/com/persistit/IOMeter.java'
--- src/main/java/com/persistit/IOMeter.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/IOMeter.java	2012-09-12 22:04:20 +0000
@@ -32,6 +32,7 @@
 
 import com.persistit.mxbeans.IOMeterMXBean;
 import com.persistit.util.ArgParser;
+import com.persistit.util.Util;
 
 /**
  * 
@@ -57,7 +58,9 @@
     private final static String DUMP_FORMAT = "time=%,12d op=%2s vol=%4s page=%,16d addr=%,16d size=%,8d index=%,7d";
     private final static int DUMP_RECORD_LENGTH = 37;
 
-    private final static int DEFAULT_QUIESCENT_IO_THRESHOLD = 100000;
+    private final static int DEFAULT_QUIESCENT_IO_THRESHOLD_KBYTES_PER_SEC = 100;
+    private final static int MINIMUM_QUIESCENT_IO_THRESHOLD_KBYTES_PER_SEC = 0;
+    private final static int MAXIMUM_QUIESCENT_IO_THRESHOLD_KBYTES_PER_SEC = 1000000;
 
     private final static int READ_PAGE_FROM_VOLUME = 1;
     private final static int READ_PAGE_FROM_JOURNAL = 2;
@@ -72,7 +75,7 @@
 
     private final static int ITEM_COUNT = 11;
 
-    private long _quiescentIOthreshold = DEFAULT_QUIESCENT_IO_THRESHOLD;
+    private long _quiescentIOthreshold = DEFAULT_QUIESCENT_IO_THRESHOLD_KBYTES_PER_SEC;
 
     private final AtomicReference<DataOutputStream> _logStream = new AtomicReference<DataOutputStream>();
 
@@ -149,6 +152,7 @@
 
     /**
      * @return the quiescentIOthreshold
+     * @see #setQuiescentIOthreshold(long)
      */
     @Override
     public synchronized long getQuiescentIOthreshold() {
@@ -156,12 +160,18 @@
     }
 
     /**
+     * Persistit monitors the rate at which new I/O operations are created. When
+     * the IORate falls below the quiescentIOthreshold, expressed in KBytes per
+     * second, the JOURNAL_COPIER accelerates its work to try to clean up older
+     * journal files.
+     * 
      * @param quiescentIOthreshold
      *            the quiescentIOthreshold to set
      */
     @Override
-    public synchronized void setQuiescentIOthreshold(final long quiescentIO) {
-        _quiescentIOthreshold = quiescentIO;
+    public synchronized void setQuiescentIOthreshold(final long quiescentIOthreshold) {
+        _quiescentIOthreshold = Util.rangeCheck(quiescentIOthreshold, MINIMUM_QUIESCENT_IO_THRESHOLD_KBYTES_PER_SEC,
+                MAXIMUM_QUIESCENT_IO_THRESHOLD_KBYTES_PER_SEC);
     }
 
     /**

=== modified file 'src/main/java/com/persistit/IntegrityCheck.java'
--- src/main/java/com/persistit/IntegrityCheck.java	2012-09-11 14:14:55 +0000
+++ src/main/java/com/persistit/IntegrityCheck.java	2012-09-12 22:04:20 +0000
@@ -386,6 +386,26 @@
     }
 
     /**
+     * Control output format. When CSV mode is enabled, the output is organized
+     * as comma-separated-variable text that can be imported into a spreadsheet.
+     * 
+     * @param csvMode
+     */
+    public void setCsvMode(final boolean csvMode) {
+        _csv = csvMode;
+    }
+
+    /**
+     * Indicate whether CSV mode is enabled. If so the output is organized as
+     * comma-separated-variable text that can be imported into a spreadsheet.
+     * 
+     * @return <code>true<c/code> if CSV mode is enabled.
+     */
+    public boolean isCsvMode() {
+        return _csv;
+    }
+
+    /**
      * Indicate whether missing index pages should be added when an index "hole"
      * is discovered.
      * 

=== modified file 'src/main/java/com/persistit/JournalManager.java'
--- src/main/java/com/persistit/JournalManager.java	2012-09-11 14:14:55 +0000
+++ src/main/java/com/persistit/JournalManager.java	2012-09-12 22:04:20 +0000
@@ -29,15 +29,15 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -80,9 +80,9 @@
     final static int GENTLE_COMMIT_DELAY_MILLIS = 12;
     private final static long NS_PER_MS = 1000000L;
     private final static int IO_MEASUREMENT_CYCLES = 8;
-
-    private final static int TOO_MANY_WARN_THRESHOLD = 15;
-    private final static int TOO_MANY_ERROR_THRESHOLD = 20;
+    private final static int TOO_MANY_WARN_THRESHOLD = 5;
+    private final static int TOO_MANY_ERROR_THRESHOLD = 10;
+    private final static long KILO = 1024;
 
     /**
      * REGEX expression that recognizes the name of a journal file.
@@ -104,7 +104,6 @@
     private final Map<TreeDescriptor, Integer> _treeToHandleMap = new HashMap<TreeDescriptor, Integer>();
 
     private final Map<Integer, TreeDescriptor> _handleToTreeMap = new HashMap<Integer, TreeDescriptor>();
-
     private final Map<Long, TransactionMapItem> _liveTransactionMap = new HashMap<Long, TransactionMapItem>();
 
     private final Persistit _persistit;
@@ -184,9 +183,9 @@
 
     private final AtomicLong _totalFlushIoTime = new AtomicLong();
 
-    private volatile long _flushInterval = DEFAULT_FLUSH_INTERVAL;
+    private volatile long _flushInterval = DEFAULT_FLUSH_INTERVAL_MS;
 
-    private volatile long _slowIoAlertThreshold = DEFAULT_SLOW_IO_ALERT_THRESHOLD;
+    private volatile long _slowIoAlertThreshold = DEFAULT_SLOW_IO_ALERT_THRESHOLD_MS;
 
     private final TransactionPlayer _player = new TransactionPlayer(new JournalTransactionPlayerSupport());
 
@@ -201,7 +200,7 @@
      * performs I/O. Hopefully we can set good defaults and not expose these as
      * knobs.
      */
-    private volatile long _copierInterval = DEFAULT_COPIER_INTERVAL;
+    private volatile long _copierInterval = DEFAULT_COPIER_INTERVAL_MS;
 
     private volatile int _copiesPerCycle = DEFAULT_COPIES_PER_CYCLE;
 
@@ -213,7 +212,9 @@
 
     private boolean _allowHandlesForTempVolumesAndTrees;
 
-    private final AtomicLong _waitLoopsWithNoDelay = new AtomicLong();
+    private volatile int _urgentFileCountThreshold = DEFAULT_URGENT_FILE_COUNT_THRESHOLD;
+
+    private volatile long _throttleSleepInterval;
 
     /**
      * <p>
@@ -537,10 +538,22 @@
 
     @Override
     public void setSlowIoAlertThreshold(final long slowIoAlertThreshold) {
-        Util.rangeCheck(slowIoAlertThreshold, MINIMUM_SLOW_ALERT_THRESHOLD, MAXIMUM_SLOW_ALERT_THRESHOLD);
+        Util.rangeCheck(slowIoAlertThreshold, MINIMUM_SLOW_ALERT_THRESHOLD_MS, MAXIMUM_SLOW_ALERT_THRESHOLD_MS);
         _slowIoAlertThreshold = slowIoAlertThreshold;
     }
 
+    @Override
+    public int getUrgentFileCountThreshold() {
+        return _urgentFileCountThreshold;
+    }
+
+    @Override
+    public void setUrgentFileCountThreshold(final int threshold) {
+        Util.rangeCheck(threshold, MINIMUM_URGENT_FILE_COUNT_THRESHOLD, MAXIMUM_URGENT_FILE_COUNT_THRESHOLD);
+        _urgentFileCountThreshold = threshold;
+
+    }
+
     /**
      * Compute an "urgency" factor that determines how vigorously the
      * JOURNAL_COPIER thread should perform I/O. This number is computed on a
@@ -554,8 +567,8 @@
         if (_copyFast.get()) {
             return URGENT;
         }
-        final int journalFileCount = getJournalFileCount();
-        return Math.min(URGENT, journalFileCount);
+        final int remainingFiles = _urgentFileCountThreshold - getJournalFileCount();
+        return Math.max(0, Math.min(URGENT - remainingFiles, URGENT));
     }
 
     /**
@@ -567,13 +580,9 @@
      * @throws PersistitInterruptedException
      */
     public void throttle() throws PersistitInterruptedException {
-        final int urgency = urgency();
-        if (!_appendOnly.get()) {
-            if (urgency == URGENT) {
-                Util.sleep(URGENT_COMMIT_DELAY_MILLIS);
-            } else if (urgency >= ALMOST_URGENT) {
-                Util.sleep(GENTLE_COMMIT_DELAY_MILLIS);
-            }
+        final long interval = _throttleSleepInterval;
+        if (interval > 0) {
+            Util.sleep(interval);
         }
     }
 
@@ -1007,14 +1016,6 @@
         //
         force();
         //
-        // Make sure all copied pages have been flushed to disk.
-        //
-        for (final Volume vol : _volumeToHandleMap.keySet()) {
-            if (vol.isOpened()) {
-                vol.getStorage().force();
-            }
-        }
-        //
         // Prepare room for CP.OVERHEAD bytes in the journal. If doing so
         // started a new journal file then there's no need to write another
         // CP record.
@@ -1227,7 +1228,6 @@
     static long fileToGeneration(final File file) {
         final Matcher matcher = PATH_PATTERN.matcher(file.getName());
         if (matcher.matches()) {
-            // TODO - validate range
             return Long.parseLong(matcher.group(2));
         } else {
             return -1;
@@ -1237,7 +1237,6 @@
     static String fileToPath(final File file) {
         final Matcher matcher = PATH_PATTERN.matcher(file.getPath());
         if (matcher.matches()) {
-            // TODO - validate range
             return matcher.group(1);
         } else {
             return null;
@@ -1677,11 +1676,6 @@
             }
         }
         //
-        // Remove the page list entries too.
-        //
-        _droppedPageCount += cleanupPageList();
-
-        //
         // Remove any PageNode from the branchMap having a timestamp less
         // than the checkpoint. Generally all such entries are removed after
         // the first checkpoint that has been established after recovery.
@@ -2168,6 +2162,20 @@
             } finally {
                 _copying.set(false);
             }
+
+            long throttleInterval = 0;
+            if (!_appendOnly.get()) {
+                final int urgency = urgency();
+                if (urgency == URGENT) {
+                    throttleInterval = URGENT_COMMIT_DELAY_MILLIS;
+                } else if (urgency > ALMOST_URGENT) {
+                    throttleInterval = GENTLE_COMMIT_DELAY_MILLIS;
+                }
+            }
+            if (throttleInterval != _throttleSleepInterval) {
+                _throttleSleepInterval = throttleInterval;
+            }
+
         }
 
         @Override
@@ -2175,14 +2183,14 @@
             return _closed.get() || _shouldStop;
         }
 
-        @Override
         /**
-         * Return a nice interval, in milliseconds, to wait between
-         * copierCycle invocations. The interval decreases as interval 
-         * goes up, and becomes zero when the urgency is 10. The interval
-         * is also zero if there has be no recent I/O activity invoked
-         * by other activities.
+         * Return a nice interval, in milliseconds, to wait between copierCycle
+         * invocations. The interval decreases as interval goes up, and becomes
+         * zero when the urgency is greater than or equal to 8. The interval is
+         * also zero if there has be no recent I/O activity invoked by other
+         * activities.
          */
+        @Override
         public long getPollInterval() {
             final IOMeter iom = _persistit.getIOMeter();
             final long pollInterval = super.getPollInterval();
@@ -2198,7 +2206,7 @@
 
             int divisor = 1;
 
-            if (iom.recentCharge() < iom.getQuiescentIOthreshold()) {
+            if (iom.recentCharge() < iom.getQuiescentIOthreshold() * KILO) {
                 divisor = HALF_URGENT;
             } else if (urgency > HALF_URGENT) {
                 divisor = urgency - HALF_URGENT;
@@ -2214,8 +2222,6 @@
 
     private class JournalFlusher extends IOTaskRunnable {
 
-        final ReentrantReadWriteLock _lock = new ReentrantReadWriteLock();
-
         volatile long _lastExceptionTimestamp = 0;
         volatile Exception _lastException = null;
 
@@ -2248,6 +2254,7 @@
              * posted an _endTimestamp larger than flushedTimestamp.
              */
             final long now = System.nanoTime();
+            long remainingStallTime = stallTime;
 
             while (true) {
                 /*
@@ -2269,10 +2276,10 @@
                     endTimestamp = _endTimestamp;
                     startTime = _startTime;
                     endTime = _endTime;
-                    if (flushedTimestamp > startTimestamp && startTimestamp > endTimestamp) {
-                        estimatedRemainingIoNanos = Math.max(startTime + _expectedIoTime - now, 0);
-                    }
                     if (startTimestamp == _startTimestamp && endTimestamp == _endTimestamp) {
+                        if (flushedTimestamp > startTimestamp && startTimestamp > endTimestamp) {
+                            estimatedRemainingIoNanos = Math.max(startTime + _expectedIoTime - now, 0);
+                        }
                         break;
                     }
                     Util.spinSleep();
@@ -2280,22 +2287,23 @@
 
                 if (endTimestamp > flushedTimestamp && startTimestamp > flushedTimestamp) {
                     /*
-                     * Done - commit is fully durable
+                     * Done - commit is durable
                      */
                     break;
                 }
 
                 long remainingSleepNanos;
-                boolean didWait = false;
                 if (estimatedRemainingIoNanos == -1) {
                     remainingSleepNanos = Math.max(0, _flushInterval - (now - endTime));
                 } else {
                     remainingSleepNanos = _flushInterval;
                 }
 
-                long estimatedNanosToFinish = Math.max(estimatedRemainingIoNanos, 0);
+                long estimatedNanosToFinish;
                 if (startTimestamp < flushedTimestamp) {
-                    estimatedNanosToFinish += remainingSleepNanos + _expectedIoTime;
+                    estimatedNanosToFinish = remainingSleepNanos + _expectedIoTime;
+                } else {
+                    estimatedNanosToFinish = estimatedRemainingIoNanos;
                 }
 
                 if (leadTime > 0 && leadTime * NS_PER_MS >= estimatedNanosToFinish) {
@@ -2311,41 +2319,20 @@
                      * possible (determined by stallTime) before kicking the
                      * JOURNAL_FLUSHER to write the caller's transaction.
                      */
-                    final long delay = stallTime * NS_PER_MS - estimatedNanosToFinish;
-                    if (delay > 0) {
-                        Util.sleep(delay / NS_PER_MS);
-                        didWait = true;
-                    }
-                    kick();
-                    if (delay <= 0) {
-                        didWait = true;
-                        try {
-                            if (_lock.readLock().tryLock(NS_PER_MS, TimeUnit.NANOSECONDS)) {
-                                _lock.readLock().unlock();
-                            }
-                        } catch (final InterruptedException e) {
-                            throw new PersistitInterruptedException(e);
-                        }
+                    if (remainingStallTime > 0) {
+                        Util.sleep(remainingStallTime);
+                        remainingStallTime = 0;
+                    } else {
+                        kick();
+                        Util.spinSleep();
                     }
                 } else {
                     /*
-                     * Otherwise, wait until the I/O is about half done and then
-                     * retry.
+                     * Otherwise wait for concurrent I/O operation to finish. Do
+                     * this by polling because our experiments with using locks
+                     * here showed significant excess CPU consumption.
                      */
-                    final long delay = ((estimatedNanosToFinish - leadTime * NS_PER_MS) / 2) + NS_PER_MS;
-                    try {
-                        if (delay > 0) {
-                            didWait = true;
-                            if (_lock.readLock().tryLock(delay, TimeUnit.NANOSECONDS)) {
-                                _lock.readLock().unlock();
-                            }
-                        }
-                    } catch (final InterruptedException e) {
-                        throw new PersistitInterruptedException(e);
-                    }
-                }
-                if (!didWait) {
-                    _waitLoopsWithNoDelay.incrementAndGet();
+                    Util.spinSleep();
                 }
             }
             if (_lastExceptionTimestamp > flushedTimestamp) {
@@ -2370,7 +2357,6 @@
                      * waitForDurability to know when the I/O operation has
                      * finished.
                      */
-                    _lock.writeLock().lock();
                     try {
                         _startTimestamp = _persistit.getTimestampAllocator().updateTimestamp();
                         _startTime = System.nanoTime();
@@ -2382,22 +2368,24 @@
                     } finally {
                         _endTime = System.nanoTime();
                         _endTimestamp = _persistit.getTimestampAllocator().updateTimestamp();
-                        _lock.writeLock().unlock();
                     }
 
                     final long elapsed = _endTime - _startTime;
                     _totalFlushCycles.incrementAndGet();
                     _totalFlushIoTime.addAndGet(elapsed);
                     _ioTimes[_ioCycle] = elapsed;
-                    _ioCycle = (_ioCycle + 1) % _ioTimes.length;
+                    _ioCycle = (_ioCycle + 1) % IO_MEASUREMENT_CYCLES;
 
-                    long max = 0;
-                    for (int index = 0; index < _ioTimes.length; index++) {
-                        max = Math.max(max, _ioTimes[index]);
+                    long avg = 0;
+                    for (int index = 0; index < IO_MEASUREMENT_CYCLES; index++) {
+                        avg += _ioTimes[index];
                     }
-                    _expectedIoTime = max;
+                    avg /= IO_MEASUREMENT_CYCLES;
+
+                    _expectedIoTime = avg;
                     if (elapsed > _slowIoAlertThreshold * NS_PER_MS) {
-                        _persistit.getLogBase().longJournalIO.log(elapsed / NS_PER_MS);
+                        _persistit.getLogBase().longJournalIO.log(elapsed / NS_PER_MS, IO_MEASUREMENT_CYCLES, avg
+                                / NS_PER_MS);
                     }
 
                 } catch (final Exception e) {
@@ -2416,6 +2404,7 @@
             } finally {
                 _flushing.set(false);
             }
+
         }
 
         @Override
@@ -2426,14 +2415,12 @@
 
     synchronized void selectForCopy(final List<PageNode> list) {
         list.clear();
-        _droppedPageCount += cleanupPageList();
         if (!_appendOnly.get()) {
             final long timeStampUpperBound = Math.min(getLastValidCheckpointTimestamp(), _copierTimestampLimit);
             for (final Iterator<PageNode> iterator = _pageList.iterator(); iterator.hasNext();) {
                 final PageNode pageNode = iterator.next();
-                for (PageNode pn = pageNode; pn != null; pn = pn.getPrevious()) {
+                for (PageNode pn = pageNode; pn != null && !pn.isInvalid(); pn = pn.getPrevious()) {
                     if (pn.getTimestamp() < timeStampUpperBound) {
-                        assert !pn.isInvalid();
                         list.add(pn);
                         break;
                     }
@@ -2517,6 +2504,7 @@
         Volume volumeRef = null;
         Volume volume = null;
         int handle = -1;
+        final Set<Volume> volumes = new HashSet<Volume>();
 
         for (final Iterator<PageNode> iterator = list.iterator(); iterator.hasNext();) {
             if (_closed.get() && !_copyFast.get() || _appendOnly.get()) {
@@ -2567,6 +2555,7 @@
 
             try {
                 volume.getStorage().writePage(bb, pageAddress);
+                volumes.add(volume);
             } catch (final PersistitException ioe) {
                 _persistit.getLogBase().copyException.log(ioe, volume, pageNode.getPageAddress(),
                         pageNode.getJournalAddress());
@@ -2578,6 +2567,10 @@
                     pageNode.getJournalAddress(), urgency());
         }
 
+        for (final Volume vol : volumes) {
+            vol.getStorage().force();
+        }
+
     }
 
     private void cleanupForCopy(final List<PageNode> list) throws PersistitException {
@@ -2715,26 +2708,21 @@
      * @return Count of removed PageNode instances.
      */
     int cleanupPageList() {
-        int to = -1;
-        int count = 0;
-        for (int index = _pageList.size(); --index >= 0;) {
-            if (_pageList.get(index).isInvalid()) {
-                if (to == -1) {
-                    to = index;
-                }
-            } else {
-                if (to != -1) {
-                    _pageList.removeRange(index + 1, to + 1);
-                    count += to - index;
-                    to = -1;
-                }
+        final int size = _pageList.size();
+        int from;
+        for (from = 0; from < size && !_pageList.get(from).isInvalid(); from++)
+            ;
+        int to = from;
+        for (from = from + 1; from < size; from++) {
+            final PageNode pn = _pageList.get(from);
+            if (!pn.isInvalid()) {
+                _pageList.set(to++, pn);
             }
         }
-        if (to != -1) {
-            _pageList.removeRange(0, to + 1);
-            count += to + 1;
+        if (size > to) {
+            _pageList.removeRange(to, size);
         }
-        return count;
+        return size - to;
     }
 
     private void reportJournalFileCount() {
@@ -2744,11 +2732,11 @@
          */
         final int journalFileCount = getJournalFileCount();
         if (journalFileCount != _lastReportedJournalFileCount) {
-            if (journalFileCount > TOO_MANY_ERROR_THRESHOLD) {
+            if (journalFileCount > TOO_MANY_ERROR_THRESHOLD + _urgentFileCountThreshold) {
                 _persistit.getAlertMonitor()
                         .post(new Event(AlertLevel.ERROR, _persistit.getLogBase().tooManyJournalFilesError,
                                 journalFileCount), AlertMonitor.MANY_JOURNAL_FILES);
-            } else if (journalFileCount > TOO_MANY_WARN_THRESHOLD) {
+            } else if (journalFileCount > TOO_MANY_WARN_THRESHOLD + _urgentFileCountThreshold) {
                 _persistit.getAlertMonitor()
                         .post(new Event(AlertLevel.WARN, _persistit.getLogBase().tooManyJournalFilesWarning,
                                 journalFileCount), AlertMonitor.MANY_JOURNAL_FILES);
@@ -2923,11 +2911,11 @@
         _liveTransactionMap.clear();
     }
 
-    synchronized long getCurrentJournalSize() {
+    long getCurrentJournalSize() {
         return _currentAddress % _blockSize;
     }
 
-    synchronized int getJournalFileCount() {
+    int getJournalFileCount() {
         return (int) (_currentAddress / _blockSize - _baseAddress / _blockSize) + 1;
     }
 
@@ -2989,8 +2977,4 @@
     public SortedMap<Integer, TreeDescriptor> queryTreeMap() {
         return new TreeMap<Integer, TreeDescriptor>(_handleToTreeMap);
     }
-
-    long getWaitLoopsWithNoDelay() {
-        return _waitLoopsWithNoDelay.get();
-    }
 }

=== modified file 'src/main/java/com/persistit/Persistit.java'
--- src/main/java/com/persistit/Persistit.java	2012-09-11 14:14:55 +0000
+++ src/main/java/com/persistit/Persistit.java	2012-09-12 22:04:20 +0000
@@ -327,11 +327,11 @@
     private final static SplitPolicy DEFAULT_SPLIT_POLICY = SplitPolicy.PACK_BIAS;
     private final static JoinPolicy DEFAULT_JOIN_POLICY = JoinPolicy.EVEN_BIAS;
     private final static CommitPolicy DEFAULT_TRANSACTION_COMMIT_POLICY = CommitPolicy.SOFT;
-    private final static long DEFAULT_COMMIT_LEAD_TIME = 100;
-    private final static long DEFAULT_COMMIT_STALL_TIME = 1;
-    private final static long MAX_COMMIT_LEAD_TIME = 5000;
-    private final static long MAX_COMMIT_STALL_TIME = 5000;
-    private final static long FLUSH_DELAY_INTERVAL = 5000;
+    private final static long DEFAULT_COMMIT_LEAD_TIME_MS = 100;
+    private final static long DEFAULT_COMMIT_STALL_TIME_MS = 1;
+    private final static long MAX_COMMIT_LEAD_TIME_MS = 5000;
+    private final static long MAX_COMMIT_STALL_TIME_MS = 5000;
+    private final static long LOG_FLUSH_DELAY_INTERVAL_MS = 5000;
 
     private final static int MAX_FATAL_ERROR_MESSAGES = 10;
 
@@ -365,7 +365,7 @@
         public void run() {
             while (!_stop) {
                 try {
-                    Util.sleep(FLUSH_DELAY_INTERVAL);
+                    Util.sleep(LOG_FLUSH_DELAY_INTERVAL_MS);
                 } catch (final PersistitInterruptedException ie) {
                     break;
                 }
@@ -458,9 +458,9 @@
 
     private volatile CommitPolicy _defaultCommitPolicy = DEFAULT_TRANSACTION_COMMIT_POLICY;
 
-    private volatile long _commitLeadTime = DEFAULT_COMMIT_LEAD_TIME;
+    private volatile long _commitLeadTime = DEFAULT_COMMIT_LEAD_TIME_MS;
 
-    private volatile long _commitStallTime = DEFAULT_COMMIT_STALL_TIME;
+    private volatile long _commitStallTime = DEFAULT_COMMIT_STALL_TIME_MS;
 
     private final ThreadLocal<SoftReference<int[]>> _intArrayThreadLocal = new ThreadLocal<SoftReference<int[]>>();
 
@@ -1447,6 +1447,8 @@
          */
         for (int i = 0; i < 5; ++i) {
             if (!_closed.get() && _initialized.get()) {
+                _transactionIndex.updateActiveTransactionCache();
+                _journalManager.pruneObsoleteTransactions();
                 _checkpointManager.checkpoint();
                 _journalManager.copyBack();
                 final int fileCount = _journalManager.getJournalFileCount();
@@ -2051,7 +2053,7 @@
     }
 
     void setTransactionCommitleadTime(final long time) {
-        _commitLeadTime = Util.rangeCheck(time, 0, MAX_COMMIT_LEAD_TIME);
+        _commitLeadTime = Util.rangeCheck(time, 0, MAX_COMMIT_LEAD_TIME_MS);
     }
 
     long getTransactionCommitStallTime() {
@@ -2059,7 +2061,7 @@
     }
 
     void setTransactionCommitStallTime(final long time) {
-        _commitStallTime = Util.rangeCheck(time, 0, MAX_COMMIT_STALL_TIME);
+        _commitStallTime = Util.rangeCheck(time, 0, MAX_COMMIT_STALL_TIME_MS);
     }
 
 /**

=== modified file 'src/main/java/com/persistit/TransactionPlayer.java'
--- src/main/java/com/persistit/TransactionPlayer.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/TransactionPlayer.java	2012-09-12 22:04:20 +0000
@@ -300,7 +300,9 @@
         if (VolumeStructure.DIRECTORY_TREE_NAME.equals(td.getTreeName())) {
             return volume.getStructure().directoryExchange();
         } else {
-            return _support.getPersistit().getExchange(volume, td.getTreeName(), true);
+            final Exchange exchange = _support.getPersistit().getExchange(volume, td.getTreeName(), true);
+            exchange.ignoreTransactions();
+            return exchange;
         }
     }
 

=== modified file 'src/main/java/com/persistit/logging/LogBase.java'
--- src/main/java/com/persistit/logging/LogBase.java	2012-08-24 14:00:17 +0000
+++ src/main/java/com/persistit/logging/LogBase.java	2012-09-12 22:04:20 +0000
@@ -232,7 +232,7 @@
     @Message("WARNING|Crash retried %,d times on %s")
     public final LogItem crashRetry = PersistitLogMessage.empty();
 
-    @Message("WARNING|Journal flush operation took %,dms")
+    @Message("WARNING|Journal flush operation took %,dms last %,d cycles average is %,dms")
     public final LogItem longJournalIO = PersistitLogMessage.empty();
 
     @Message("INFO|Normal journal file count %,d")
@@ -247,7 +247,7 @@
     @Message("INFO|Preloading buffer pool inventory recorded at %tc")
     public final LogItem bufferInventoryLoad = PersistitLogMessage.empty();
 
-    @Message("INFO|Preloaded %,d of %,d buffers")
+    @Message("INFO|Preloaded %,d of %,d buffers in %,d seconds")
     public final LogItem bufferInventoryProgress = PersistitLogMessage.empty();
 
     @Message("WARNING|Exception while writing buffer pool inventory %s")

=== modified file 'src/main/java/com/persistit/mxbeans/IOMeterMXBean.java'
--- src/main/java/com/persistit/mxbeans/IOMeterMXBean.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/mxbeans/IOMeterMXBean.java	2012-09-12 22:04:20 +0000
@@ -69,7 +69,7 @@
     /**
      * @return the quiescentIOthreshold
      */
-    @Description("Disk I/O scheduling parameter in bytes per second specifying threshold "
+    @Description("Disk I/O scheduling parameter in KBytes per second specifying threshold "
             + "between \"quiescent\" and \"busy\" states")
     public long getQuiescentIOthreshold();
 
@@ -77,7 +77,7 @@
      * @param quiescentIO
      *            the quiescentIOthreshold to set
      */
-    @Description("Disk I/O scheduling parameter in bytes per second specifying threshold "
+    @Description("Disk I/O scheduling parameter in KBytes per second specifying threshold "
             + "between \"quiescent\" and \"busy\" states")
     public void setQuiescentIOthreshold(long quiescentIO);
 

=== modified file 'src/main/java/com/persistit/mxbeans/JournalManagerMXBean.java'
--- src/main/java/com/persistit/mxbeans/JournalManagerMXBean.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/mxbeans/JournalManagerMXBean.java	2012-09-12 22:04:20 +0000
@@ -76,14 +76,20 @@
      * Default time interval (in milliseconds) between calls to the
      * FileChannel.force() method.
      */
-    final static long DEFAULT_FLUSH_INTERVAL = 100;
+    final static long DEFAULT_FLUSH_INTERVAL_MS = 100;
 
     /**
      * Default time interval (in milliseconds) between calls to the journal
      * copier method.
      */
-    final static long DEFAULT_COPIER_INTERVAL = 10000;
-
+    final static long DEFAULT_COPIER_INTERVAL_MS = 10000;
+    /**
+     * Default journal file count at which transactions are throttled to allow
+     * copier to catch up.
+     */
+    final static int DEFAULT_URGENT_FILE_COUNT_THRESHOLD = 15;
+    final static int MINIMUM_URGENT_FILE_COUNT_THRESHOLD = 5;
+    final static int MAXIMUM_URGENT_FILE_COUNT_THRESHOLD = 100;
     /**
      * Default value for maximum pages to be copied per cycle.
      */
@@ -94,17 +100,17 @@
      * exceptions on attempts to write to the journal. Prevents excessively
      * verbose log on repeated failures.
      */
-    final static long DEFAULT_LOG_REPEAT_INTERVAL = 60000L;
-    final static long MINIMUM_LOG_REPEAT_INTERVAL = 1000L;
-    final static long MAXIMUM_LOG_REPEAT_INTERVAL = Long.MAX_VALUE;
+    final static long DEFAULT_LOG_REPEAT_INTERVAL_MS = 60000L;
+    final static long MINIMUM_LOG_REPEAT_INTERVAL_MS = 1000L;
+    final static long MAXIMUM_LOG_REPEAT_INTERVAL_MS = Long.MAX_VALUE;
     /**
      * Default threshold time in milliseconds for JournalManager flush
      * operations. If a flush operation takes longer than this time, a WARNING
      * message is written to the log.
      */
-    final static long DEFAULT_SLOW_IO_ALERT_THRESHOLD = 2000L;
-    final static long MINIMUM_SLOW_ALERT_THRESHOLD = 100L;
-    final static long MAXIMUM_SLOW_ALERT_THRESHOLD = Long.MAX_VALUE;
+    final static long DEFAULT_SLOW_IO_ALERT_THRESHOLD_MS = 2000L;
+    final static long MINIMUM_SLOW_ALERT_THRESHOLD_MS = 100L;
+    final static long MAXIMUM_SLOW_ALERT_THRESHOLD_MS = Long.MAX_VALUE;
 
     /**
      * File name appended when journal path specifies only a directory
@@ -115,12 +121,6 @@
      */
     final static String PATH_FORMAT = "%s.%012d";
 
-    /**
-     * Default setting for number of pages in the page map before the urgency of
-     * copying starts to increase.
-     */
-    final static int DEFAULT_PAGE_MAP_SIZE_BASE = 250000;
-
     final static int MAXIMUM_CONCURRENT_TRANSACTIONS = 10000;
 
     @Description("Number of transaction map items in the live map")
@@ -243,4 +243,9 @@
     @Description("Threshold in  milliseconds for warnings of long duration flush cycles")
     void setSlowIoAlertThreshold(long slowIoAlertThreshold);
 
+    @Description("Journal file count threshold for throttling transactions")
+    int getUrgentFileCountThreshold();
+
+    @Description("Journal file count threshold for throttling transactions")
+    void setUrgentFileCountThreshold(int threshold);
 }

=== modified file 'src/test/java/com/persistit/JournalManagerTest.java'
--- src/test/java/com/persistit/JournalManagerTest.java	2012-09-04 21:01:23 +0000
+++ src/test/java/com/persistit/JournalManagerTest.java	2012-09-12 22:04:20 +0000
@@ -33,6 +33,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -43,7 +44,6 @@
 
 import com.persistit.CheckpointManager.Checkpoint;
 import com.persistit.JournalManager.PageNode;
-import com.persistit.Transaction.CommitPolicy;
 import com.persistit.TransactionPlayer.TransactionPlayerListener;
 import com.persistit.exception.PersistitException;
 import com.persistit.unit.ConcurrentUtil.ThrowingRunnable;
@@ -440,7 +440,7 @@
          * Randomly invalidated PageNodes
          */
         {
-            final int SIZE = 5000;
+            final int SIZE = 1000000;
             final Random random = new Random(1);
             final List<PageNode> source = testCleanupPageListSource(SIZE);
             int next = -1;
@@ -448,8 +448,8 @@
                 if (index < next) {
                     source.get(index).invalidate();
                 } else {
-                    index += random.nextInt(50);
-                    next = random.nextInt(50) + index;
+                    index += random.nextInt(5);
+                    next = random.nextInt(5) + index;
                 }
             }
             testCleanupPageListHelper(source);
@@ -538,34 +538,16 @@
                 .getIgnoredUpdates() > 0);
     }
 
-    @Test
-    public void waitForDurabilitySoaksCPU() throws Exception {
-        _persistit.setDefaultTransactionCommitPolicy(CommitPolicy.HARD);
-        final JournalManager jman = _persistit.getJournalManager();
-        long waitLoopsWithoutDelay = jman.getWaitLoopsWithNoDelay();
-        final Exchange exchange = _persistit.getExchange("persistit", "JournalManagerTest", true);
-        final Transaction txn = exchange.getTransaction();
-        for (int count = 0; count < 1000; count++) {
-            txn.begin();
-            exchange.getValue().put(RED_FOX + count);
-            exchange.to(count).store();
-            txn.commit();
-            txn.end();
-        }
-        waitLoopsWithoutDelay = jman.getWaitLoopsWithNoDelay() - waitLoopsWithoutDelay;
-        assertEquals("Wait loops without delay", 0, waitLoopsWithoutDelay);
-    }
-
     private List<PageNode> testCleanupPageListSource(final int size) {
         final List<PageNode> source = new ArrayList<PageNode>(size);
-        for (int index = 0; index < 1000000; index++) {
+        for (int index = 0; index < size; index++) {
             source.add(new PageNode(0, index, index * 10, index));
         }
         return source;
     }
 
     private void testCleanupPageListHelper(final List<PageNode> source) throws Exception {
-        final List<PageNode> cleaned = new ArrayList<PageNode>(source);
+        final List<PageNode> cleaned = new LinkedList<PageNode>(source);
         for (final Iterator<PageNode> iterator = cleaned.iterator(); iterator.hasNext();) {
             if (iterator.next().isInvalid()) {
                 iterator.remove();