Merge lp:~pbeaman/akiban-persistit/tree_builder2 into lp:akiban-persistit

[Peter Beaman]

Fairly minor tweaks discovered and tested while running a prototype bulk load on a CAOI schema.

- added missing JavaDoc
- made a TreeBuilder reusable after a merge (for the tbPK case)
- separated _sortKeyCount and _mergedKeyCount and added accessors
- added missing condition on mustWrite in BufferPool#invalidateSmallVolume and added the write logic to BufferPool#invalidateLargeVolume so that the methods match
- added guard code in Volume#close to avoid executing the close logic twice

[Nathan Williams (nwilliams)]

Looks good.

review: Approve

[lp:~pbeaman/akiban-persistit/tree_builder2]

408. By Akiban Build User on 2013-01-03

merge pbeaman: Fairly minor tweaks discovered and tested while running a prototype bulk load on a CAOI schema.

https://code.launchpad.net/~pbeaman/akiban-persistit/tree_builder2/+merge/141794

=== modified file 'src/main/java/com/persistit/BufferPool.java'
--- src/main/java/com/persistit/BufferPool.java	2012-12-31 21:03:59 +0000
+++ src/main/java/com/persistit/BufferPool.java	2013-01-03 17:32:52 +0000
@@ -546,7 +546,7 @@
         if (ratio < SMALL_VOLUME_RATIO) {
             return invalidateSmallVolume(volume, false);
         } else {
-            return invalidateLargeVolume(volume);
+            return invalidateLargeVolume(volume, false);
         }
     }
 
@@ -569,7 +569,7 @@
                             try {
                                 if ((buffer.getVolume() == volume || volume == null) && !buffer.isFixed()
                                         && buffer.isValid()) {
-                                    if (buffer.isDirty()) {
+                                    if (mustWrite && buffer.isDirty()) {
                                         buffer.writePage();
                                     }
                                     invalidate(buffer);
@@ -602,7 +602,7 @@
 
     }
 
-    boolean invalidateLargeVolume(final Volume volume) throws PersistitInterruptedException {
+    boolean invalidateLargeVolume(final Volume volume, final boolean mustWrite) throws PersistitException {
         boolean result = true;
         int markedAvailable = 0;
         for (int index = 0; index < _bufferCount; index++) {
@@ -613,6 +613,9 @@
                     boolean invalidated = false;
                     try {
                         if ((buffer.getVolume() == volume || volume == null) && !buffer.isFixed() && buffer.isValid()) {
+                            if (mustWrite && buffer.isDirty()) {
+                                buffer.writePage();
+                            }
                             invalidate(buffer);
                             invalidated = true;
                         }
@@ -642,17 +645,13 @@
         Debug.$assert0.t(buffer.isValid() && buffer.isOwnedAsWriterByMe());
 
         while (!detach(buffer)) {
-            //
-            // Spin until detach succeeds. Note: this method must not throw an
-            // Exception
-            // because it is called in at at critical time when cleanup must be
-            // done.
-            // It is not possible to lock the hash bucket here due to possible
-            // deadlock.
-            // However, the likelihood of a lengthy live-lock is infinitesimal
-            // so polling
-            // is acceptable.
-            //
+            /*
+             * Spin until detach succeeds. Note: this method must not throw an
+             * Exception because it is called in at at critical time when
+             * cleanup must be done. It is not possible to lock the hash bucket
+             * here due to possible deadlock. However, the likelihood of a
+             * lengthy live-lock is infinitesimal so polling is acceptable.
+             */
             try {
                 Thread.sleep(1);
             } catch (final InterruptedException ie) {

=== modified file 'src/main/java/com/persistit/TreeBuilder.java'
--- src/main/java/com/persistit/TreeBuilder.java	2012-12-31 21:03:59 +0000
+++ src/main/java/com/persistit/TreeBuilder.java	2013-01-03 17:32:52 +0000
@@ -16,6 +16,7 @@
 package com.persistit;
 
 import java.io.File;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
@@ -145,8 +146,11 @@
     private final List<Volume> _sortVolumes = new ArrayList<Volume>();
     private final int _pageSize;
     private final int _pageLimit;
-    private final AtomicLong _keyCount = new AtomicLong();
+    private final AtomicLong _sortedKeyCount = new AtomicLong();
+    private final AtomicLong _mergedKeyCount = new AtomicLong();
     private volatile long _reportKeyCountMultiple = REPORT_REPORT_MULTIPLE;
+    private Volume _currentSortVolume;
+    private int _nextDirectoryIndex;
 
     private final Set<Tree> _allTrees = new HashSet<Tree>();
     private final List<Tree> _sortedTrees = new ArrayList<Tree>();
@@ -158,8 +162,6 @@
         }
     };
 
-    private Volume _currentSortVolume;
-    private int _nextDirectoryIndex;
 
     private final Comparator<Tree> _defaultTreeComparator = new Comparator<Tree>() {
         @Override
@@ -273,22 +275,58 @@
         return pageSize;
     }
 
+    /**
+     * @return Name provide when TreeBuilder was constructed. Default name is
+     *         "TreeBuilder".
+     */
     public final String getName() {
         return _name;
     }
 
+    /**
+     * Set the count of keys inserted or merged per call to
+     * {@link #reportSorted(long)} or {@link #reportMerged(long)}.
+     * 
+     * @param multiple
+     */
     public final void setReportKeyCountMultiple(final long multiple) {
         _reportKeyCountMultiple = Util.rangeCheck(multiple, 1, Long.MAX_VALUE);
     }
 
+    /**
+     * 
+     * @return Count of keys inserted or merged per call to
+     *         {@link #reportSorted(long)} or {@link #reportMerged(long)}
+     */
     public final long getReportKeyCountMultiple() {
         return _reportKeyCountMultiple;
     }
 
+    /**
+     * @return Count of sort volumes that have been created while sorting keys
+     */
     public final synchronized int getSortVolumeCount() {
         return _sortVolumes.size();
     }
 
+    /**
+     * @return Number of keys stored in sort trees
+     */
+    public long getSortedKeyCount() {
+        return _sortedKeyCount.get();
+    }
+
+    /**
+     * @return Number of keys merged into destination trees
+     */
+    public long getMergedKeyCount() {
+        return _mergedKeyCount.get();
+    }
+
+    /**
+     * @return List of destination
+     *         <code>Tree<code> instances. This list is built as keys are stored.
+     */
     public final List<Tree> getTrees() {
         final List<Tree> list = new ArrayList<Tree>();
         list.addAll(_allTrees);
@@ -296,7 +334,36 @@
         return list;
     }
 
-    public final void setSortTreeDirectories(final List<File> directories) throws Exception {
+    /**
+     * <p>
+     * Define a list of directories in which sort volumes will be created. This
+     * method can be used to override the default value provided by
+     * {@link Configuration#getTmpVolDir()} to control more closely where sort
+     * trees will be stored. If the list is empty then the directory defined by
+     * the <code>Configuration</code> will be used. If multiple directories are
+     * declared then volumes will be allocated to them in round-robin fashion.
+     * This technique can distribute large load sets over multiple volumes and
+     * can allow for interleaved disk reads during the merge process.
+     * </p>
+     * <p>
+     * If a <code>File</code> supplied to this method does not exist, an attempt
+     * is made to create it as a directory. This method also attempts to create
+     * and delete a file in each supplied directory to ensure that if there is a
+     * file permission or other problem, it is detected immediately, rather than
+     * much later during the sort process.
+     * </p>
+     * 
+     * @param directories
+     *            List of <code>File</code> instances, each of which must be a
+     *            directory
+     * @throws IllegalArgumentException
+     *             if a supplied file exists and is not a directory or cannot be
+     *             created as a new directory
+     * @throws IOException
+     *             if an attempt to create a file in one of the supplied
+     *             directories fails
+     */
+    public final void setSortTreeDirectories(final List<File> directories) throws IOException {
         if (directories == null || directories.isEmpty()) {
             synchronized (this) {
                 _directories.clear();
@@ -333,14 +400,39 @@
         }
     }
 
+    /**
+     * 
+     * @return List of directories set via the
+     *         {@link #setSortTreeDirectories(List)} method.
+     */
     public final List<File> getSortTreeDirectories() {
         return Collections.unmodifiableList(_directories);
     }
 
+    /**
+     * Store a key-value pair into a sort tree. The {@link Tree}, {@link Key}
+     * and {@link Value} are specified by the supplied {@link Exchange}.
+     * 
+     * @param exchange
+     *            The Exchange
+     * @throws Exception
+     */
     public final void store(final Exchange exchange) throws Exception {
         store(exchange.getTree(), exchange.getKey(), exchange.getValue());
     }
 
+    /**
+     * Store a key-value pair for a specified <code>Tree</code> into a sort
+     * tree.
+     * 
+     * @param tree
+     *            the Tree
+     * @param key
+     *            the Key
+     * @param value
+     *            the Value
+     * @throws Exception
+     */
     public final void store(final Tree tree, final Key key, final Value value) throws Exception {
         final Map<Tree, Exchange> map = _sortExchangeMapThreadLocal.get();
         Exchange ex = map.get(tree);
@@ -365,7 +457,7 @@
             }
         }
         if (stored) {
-            final long count = _keyCount.incrementAndGet();
+            final long count = _sortedKeyCount.incrementAndGet();
             if ((count % _reportKeyCountMultiple) == 0) {
                 reportSorted(count);
             }
@@ -394,10 +486,9 @@
      * @throws Exception
      */
     public synchronized void merge() throws Exception {
-        if ((_keyCount.get() % _reportKeyCountMultiple) != 0) {
-            reportSorted(_keyCount.get());
+        if ((_mergedKeyCount.get() % _reportKeyCountMultiple) != 0) {
+            reportSorted(_mergedKeyCount.get());
         }
-        _keyCount.set(0);
         _sortedTrees.clear();
         _sortedTrees.addAll(_allTrees);
         Tree currentTree = null;
@@ -435,8 +526,8 @@
                 }
                 if (stored) {
                     afterMergeKey(ex);
-                    if ((_keyCount.incrementAndGet() % _reportKeyCountMultiple) == 0) {
-                        reportMerged(_keyCount.get());
+                    if ((_mergedKeyCount.incrementAndGet() % _reportKeyCountMultiple) == 0) {
+                        reportMerged(_mergedKeyCount.get());
                     }
                 }
             }
@@ -449,13 +540,38 @@
                 node = next;
             }
         }
-        if ((_keyCount.get() % _reportKeyCountMultiple) != 0) {
-            reportMerged(_keyCount.get());
-        }
-        _keyCount.set(0);
+        if ((_mergedKeyCount.get() % _reportKeyCountMultiple) != 0) {
+            reportMerged(_mergedKeyCount.get());
+        }
+        reset();
+    }
+
+    private synchronized void reset() throws Exception {
+        Exception exception = null;
+        for (final Volume volume : _sortVolumes) {
+            try {
+                volume.close();
+            } catch (PersistitException e) {
+                if (exception == null) {
+                    exception = e;
+                }
+            }
+        }
+        _sortVolumes.clear();
+        _currentSortVolume = null;
+        _nextDirectoryIndex = 0;
         _sortExchangeMapThreadLocal.get().clear();
         _allTrees.clear();
         _sortedTrees.clear();
+        if (exception != null) {
+            throw exception;
+        }
+    }
+
+    public void clear() throws Exception {
+        _sortedKeyCount.set(0);
+        _mergedKeyCount.set(0);
+        reset();
     }
 
     private synchronized Volume getSortVolume() throws Exception {

=== modified file 'src/main/java/com/persistit/Volume.java'
--- src/main/java/com/persistit/Volume.java	2012-12-31 21:03:59 +0000
+++ src/main/java/com/persistit/Volume.java	2013-01-03 17:32:52 +0000
@@ -51,6 +51,7 @@
     private final String _name;
     private long _id;
     private final AtomicBoolean _closing = new AtomicBoolean();
+    private final AtomicBoolean _closed = new AtomicBoolean();
     private final AtomicInteger _handle = new AtomicInteger();
     private final AtomicReference<Object> _appCache = new AtomicReference<Object>();
 
@@ -198,6 +199,9 @@
             if (!storage.claim(true, timeout)) {
                 throw new InUseException("Unable to acquire claim on " + this);
             }
+            if (_closed.get()) {
+                break;
+            }
             try {
                 //
                 // BufferPool#invalidate may fail and return false if other
@@ -208,6 +212,7 @@
                     getStructure().close();
                     getStorage().close();
                     getStatistics().reset();
+                    _closed.set(true);
                     break;
                 }
             } finally {