Merge lp:~nfernandez/akiban-persistit/persistit-warmup-capability into lp:akiban-persistit

[Nohemi Fernandez]

Optimization: Creates a file of sample page data to insert into the next 'cold' start of Persistit.

[Nohemi Fernandez (nfernandez)]

Some optimizations to consider for this branch:

1. Add checks for corrupt files
2. Store temporary back-up file of data in case of a system crash
3. Define names for multiple buffer pools-- this will be implemented if we switch from the current file I/O implementation to one using Exchange objects.

[lp:~nfernandez/akiban-persistit/persistit-warmup-capability]

356. By Nohemi Fernandez on 2012-08-02

revert changes to irrelevant files

357. By Nohemi Fernandez on 2012-08-02

merge from trunk

358. By Nohemi Fernandez on 2012-08-02

clean-up diff

[Peter Beaman (pbeaman)]

I've been thinking some more about the general problem and have some suggestions.

This version represents good progress. It passes all tests and makes the warmup process optional.

Here are some additional issues I'd like to address, perhaps over time:

Writing the buffer pool inventory should be optional. I suggest using the same configuration property, isBufferWarmupEnabled(), to control the polling action. I also think we might be able to put the inventory-writing activity on the CHECKPOINT_MANAGER thread since it does work only about once every two minutes. The Javadoc for setBufferWarmupEnabled(boolean) says the property is settable from the Management class. We would want to actually make it so.

When we load a large buffer pool, say 1M pages of 16Kb each (which we already have on largish machines) it will be important to optimize the order in which these are read to reduce physical I/O delays. About the best we can do is to read pages in file offset order which is approximately linear with page address order - "approximately" because some pages may be read from the journal instead of the volume file. We probably want to sort the list of pages being read by physical file address order before reading anything.

We may benefit from repopulating the BufferPool in "clock" order. This will generally not be the same order in which we load the pages. The idea would be to read the pages in physical file order but put them in the _buffers array in an order that approximates their distance from the "clock" pointer at the time the inventory was taken. To do this we'll need a method other than BufferPool#get(...) to read and install the pages. I'm giving this some thought and will probably propose something later. We can do without this for now.

I wonder if there is any benefit in reloading only pages that have been in the BufferPool for awhile. For example, if we inventory the BufferPool during a data load operating we'll record lots of pages in the warmup file that when reloaded will never be used again. One possibility is to snapshot the inventory somewhat more frequently, say once per minute, and only include pages that have been in place for two or three cycles.

Small detail - should have brought this up earlier. Each Volume in the BufferPool has a small integer handle available through Volume#getHandle(). This is a good alternative to recording the volume name.

Finally, we can't put this in trunk without resolving where the warmup file goes; clearly we can't keep it in a fixed location in the /tmp directory. I think if we can resolve the file location, make the writing of the file optional through configuration, fix the Javadoc for setBufferWarmupEnabled not to refer to a method that currently does not exist in Management, we could probably merge this and at least try it in some of our bigger tests.

[lp:~nfernandez/akiban-persistit/persistit-warmup-capability]

359. By Nohemi Fernandez on 2012-08-07

change configuration property for buffer pools

360. By Nohemi Fernandez on 2012-08-07

add configuration setting for warm-up polling time

[Nohemi Fernandez (nfernandez)]

I added configuration properties for the buffer pool warm-up as follows:
1. "bufferinventory" --> path name where buffer pool information will be stored, if null then pool does not get warmed up
2. "bufferpollinginterval" -> number of seconds for polling

review: Needs Resubmitting

[Peter Beaman (pbeaman)]

Does the PAGE_CACHER thread try to call populateWarmuFile using a path of null when? I didn't see logic to disable the thread when the bufferinventoryfile property is missing.

review: Needs Information

[Nohemi Fernandez (nfernandez)]

316 + if (_configuration.getBufferInventoryPathName() != null) {
317 + warmupBufferPools();
318 + }

The page cacher thread is not started unless the above method is called, so the path in BufferPool is unnecessary (null) if there is no path name in the configuration.

[Peter Beaman (pbeaman)]

Suggestions:

- Merge from trunk
- Range-check the buffer poll interval - buffer inventory poll interval. Decide and document in javadoc whether it is specified in ms, seconds or minutes.
- Move property from PersistitUnitTestCase to UnitTestProperties

[lp:~nfernandez/akiban-persistit/persistit-warmup-capability]

361. By Nohemi Fernandez on 2012-08-07

range check buffer inventory poll interval

362. By Nohemi Fernandez on 2012-08-07

merge from trunk

363. By Nohemi Fernandez on 2012-08-07

update WarmupTest header

[Nohemi Fernandez (nfernandez)]

Suggestions followed.

review: Needs Resubmitting

[Peter Beaman (pbeaman)]

Looks okay to me. Thanks for the changes.

review: Approve

=== modified file 'src/main/java/com/persistit/BufferPool.java'
--- src/main/java/com/persistit/BufferPool.java	2012-08-04 19:19:44 +0000
+++ src/main/java/com/persistit/BufferPool.java	2012-08-07 21:14:18 +0000
@@ -16,7 +16,6 @@
 package com.persistit;
 
 import java.io.DataOutputStream;
-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -37,6 +36,7 @@
 import com.persistit.exception.VolumeClosedException;
 import com.persistit.util.Debug;
 import com.persistit.util.Util;
+import java.io.*;
 
 /**
  * A pool of {@link Buffer} objects, maintained on various lists that permit
@@ -52,7 +52,7 @@
     private final static long DEFAULT_WRITER_POLL_INTERVAL = 5000;
 
     private final static int PAGE_WRITER_TRANCHE_SIZE = 5000;
-
+    
     /**
      * Sleep time when buffers are exhausted
      */
@@ -196,11 +196,24 @@
     private volatile long _writerPollInterval = DEFAULT_WRITER_POLL_INTERVAL;
 
     private volatile int _pageWriterTrancheSize = PAGE_WRITER_TRANCHE_SIZE;
+    
+    /**
+     * Polling interval for PageCacher
+     */
+    private volatile long _cacherPollInterval;
+    
     /**
      * The PAGE_WRITER IOTaskRunnable
      */
     private PageWriter _writer;
-
+    
+    /**
+     * The PAGE_CACHER IOTaskRunnable
+     */
+    private PageCacher _cacher;
+        
+    private String _defaultLogPath;
+    
     /**
      * Construct a BufferPool with the specified count of <code>Buffer</code>s
      * of the specified size.
@@ -237,7 +250,7 @@
         _hashTable = new Buffer[_bufferCount * HASH_MULTIPLE];
         _hashLocks = new ReentrantLock[HASH_LOCKS];
         _maxKeys = (_bufferSize - Buffer.HEADER_SIZE) / Buffer.MAX_KEY_RATIO;
-
+        
         for (int index = 0; index < HASH_LOCKS; index++) {
             _hashLocks[index] = new ReentrantLock();
         }
@@ -271,25 +284,59 @@
             throw e;
         }
         _writer = new PageWriter();
-
-    }
-
-    void startThreads() {
+        _cacher = new PageCacher();
+    }
+    
+    void warmupBufferPool(String pathName, String fname) throws PersistitException {
+        File file = new File(pathName, fname + ".log");
+        _defaultLogPath = file.getAbsolutePath();
+        
+        try {
+            if (!file.exists()) {
+               file.createNewFile();
+            }
+            
+            BufferedReader reader = new BufferedReader(new FileReader(file));
+            String currLine;
+            while ((currLine = reader.readLine()) != null) {
+                String[] info = currLine.split(" ");
+                if (info.length == 2) {
+                    Volume vol = _persistit.getVolume(info[1]);
+                    if (vol != null) {
+                        long page = Long.parseLong(info[0]);
+                    	Buffer buff = get(vol, page, false, true);
+                    	buff.release();
+                    }
+                }
+            }
+            reader.close();
+            _cacherPollInterval = _persistit.getConfiguration().getBufferInventoryPollingInterval();
+            _cacher.start();
+        }
+        catch (IOException e) {
+            throw new PersistitException(e);
+        }
+    }
+
+    void startThreads() throws PersistitException {
         _writer.start();
     }
 
     void close() {
         _closed.set(true);
         _persistit.waitForIOTaskStop(_writer);
+        _persistit.waitForIOTaskStop(_cacher);
         _writer = null;
+        _cacher = null;
     }
 
     /**
-     * Abruptly stop (using {@link Thread#stop()}) the writer and collector
+     * Abruptly stop (using {@link Thread#stop()}) the writer, cacher, and collector
      * threads. This method should be used only by tests.
      */
     void crash() {
         IOTaskRunnable.crash(_writer);
+        IOTaskRunnable.crash(_cacher);
     }
 
     void flush(final long timestamp) throws PersistitInterruptedException {
@@ -380,6 +427,35 @@
             buffer.populateInfo(array[index]);
         }
     }
+    
+    private void populateWarmupFile() throws PersistitException {
+        File file = new File(_defaultLogPath);
+        
+        try {
+            BufferedWriter writer = new BufferedWriter(new FileWriter(file));
+            for (int i = 0; i < _buffers.length; ++i) {
+                Buffer b = _buffers[i];
+                if (b != null && b.isValid() && !b.isDirty()) {
+                    long page = b.getPageAddress();
+                    Volume volume = b.getVolume();
+                    long page2 = b.getPageAddress();
+                    Volume volume2 = b.getVolume();
+                    
+                    // Check if buffer has changed while reading
+                    if (page == page2 && volume == volume2 && volume != null) {
+                        String addr = Long.toString(page);
+                        String vol = volume.getName(); 
+                        writer.append(addr + " " + vol);
+                        writer.newLine();
+                        writer.flush();
+                    }
+                }
+            }
+            writer.close();
+        } catch (IOException e) {
+            throw new PersistitException(e);
+        }
+    }
 
     private boolean selected(Buffer buffer, int includeMask, int excludeMask) {
         return ((includeMask == 0) || (buffer.getStatus() & includeMask) != 0)
@@ -1315,6 +1391,35 @@
             return isFlushing() ? 0 : _writerPollInterval;
         }
     }
+    
+    /**
+     * Implementation of PAGE_CACHER thread
+     */
+    class PageCacher extends IOTaskRunnable {
+        
+        PageCacher() {
+            super(BufferPool.this._persistit);
+        }
+        
+        void start() {
+            start("PAGE_CACHER:" + _bufferSize, _cacherPollInterval);
+        }
+
+        @Override
+        public void runTask() throws Exception {
+            populateWarmupFile();
+        }
+        
+        @Override
+        protected boolean shouldStop() {
+            return _closed.get() && !isFlushing();
+        }
+        
+        @Override
+        protected long pollInterval() {
+            return isFlushing() ? 0 : _cacherPollInterval;
+        }
+    }
 
     @Override
     public String toString() {
 
=== modified file 'src/main/java/com/persistit/Configuration.java'
--- src/main/java/com/persistit/Configuration.java	2012-08-02 14:19:26 +0000
+++ src/main/java/com/persistit/Configuration.java	2012-08-07 21:14:18 +0000
@@ -264,15 +264,23 @@
      * Property name for the "append only" property.
      */
     public final static String APPEND_ONLY_PROPERTY = "appendonly";
-
+    
     /**
      * Property name for the "ignore missing volumes" property.
      */
     public final static String IGNORE_MISSING_VOLUMES_PROPERTY = "ignoremissingvolumes";
+    
     /**
      * Property name to specify the default {@link SplitPolicy}.
      */
     public final static String SPLIT_POLICY_PROPERTY_NAME = "splitpolicy";
+    
+    /**
+     * Property name to specify the"buffer inventory" property name.
+     */
+    public final static String BUFFER_INVENTORY_PROPERTY_NAME = "bufferinventory";
+    
+    public final static String BUFFER_POLLING_INTERVAL_PROPERTY = "bufferpollinginterval";
 
     /**
      * Property name to specify the default {@link JoinPolicy}.
@@ -624,6 +632,8 @@
     private int rmiServerPort;
     private boolean jmx = true;
     private boolean appendOnly;
+    private String bufferInventoryPathName;
+    private long bufferInventoryPollInterval = 3000000; // default five minute polling
     private boolean ignoreMissingVolumes;
     private String tmpVolDir;
     private int tmpVolPageSize;
@@ -702,6 +712,8 @@
     }
 
     void loadProperties() throws InvalidVolumeSpecificationException {
+        setBufferInventoryPathName(getProperty(BUFFER_INVENTORY_PROPERTY_NAME));
+        setBufferInventoryPollingInterval(getLongProperty(BUFFER_POLLING_INTERVAL_PROPERTY, bufferInventoryPollInterval));
         setAppendOnly(getBooleanProperty(APPEND_ONLY_PROPERTY, false));
         setCommitPolicy(getProperty(COMMIT_POLICY_PROPERTY_NAME));
         setConstructorOverride(getBooleanProperty(CONSTRUCTOR_OVERRIDE_PROPERTY_NAME, false));
@@ -1802,6 +1814,59 @@
     }
 
     /**
+     * Return the path name defined by {@link #getBufferInventoryPathName}
+     * @return  the path where file to warm-up Persistit with sample buffer data is stored
+     */
+    public String getBufferInventoryPathName() {
+        return bufferInventoryPathName;
+    }
+
+    /**
+     * <p>
+     * Control where Persistit stores its buffer inventory. In this mode
+     * Persistit restarts with information from the last run. This method initializes
+     * the warm-up file at the specified location, if none is specified the buffer
+     * pool is not warmed up on start-up.
+     * </p>
+     * <p>
+     * Default value is <code>null</code><br />
+     * Property name is {@value #BUFFER_INVENTORY_PROPERTY_NAME}
+     * </p>
+     * 
+     * @param pathName
+     *            the name of the path to the warm-up file
+     */
+    public void setBufferInventoryPathName(String pathName) {
+        bufferInventoryPathName = pathName;
+
+    }
+    
+    /**
+     * Return polling interval defined by {@link #getBufferInventoryPollingInterval}
+     * @return  the number of seconds wait between warm-up polls
+     */
+    public long getBufferInventoryPollingInterval() {
+        return bufferInventoryPollInterval;
+    }
+    
+    /**
+     * <p>
+     * Control the number of seconds between each poll for the 
+     * cache warm-up option in Persistit.
+     * </p>
+     * <p>
+     * Default value is <code>3000</code><br />
+     * Property name is {@value #BUFFER_POLLING_INTERVAL_PROPERTY}
+     * </p>
+     * 
+     * @param seconds
+     *            the number of seconds between polls
+     */
+    public void setBufferInventoryPollingInterval(long seconds) {
+        bufferInventoryPollInterval = Util.rangeCheck(seconds, 60L, Long.MAX_VALUE) * 1000L;
+    }
+    
+    /**
      * Return the value defined by {@link #setIgnoreMissingVolumes(boolean)}
      * 
      * @return the whether to start Persistit in ignore-missing-volumes mode
 
=== modified file 'src/main/java/com/persistit/Persistit.java'
--- src/main/java/com/persistit/Persistit.java	2012-08-02 14:19:26 +0000
+++ src/main/java/com/persistit/Persistit.java	2012-08-07 21:14:18 +0000
@@ -589,6 +589,9 @@
                 initializeVolumes();
                 startJournal();
                 startBufferPools();
+                if (_configuration.getBufferInventoryPathName() != null) {
+                    warmupBufferPools();
+                }
                 finishRecovery();
                 startCheckpointManager();
                 startTransactionIndexPollTask();
@@ -716,6 +719,13 @@
             pool.startThreads();
         }
     }
+    
+    void warmupBufferPools() throws PersistitException {
+        String pathName = _configuration.getBufferInventoryPathName();
+        for (final BufferPool pool : _bufferPoolTable.values()) {
+            pool.warmupBufferPool(pathName, pool.toString());
+        }
+    }
 
     void startJournal() throws PersistitException {
         _journalManager.startJournal();
 
=== added file 'src/test/java/com/persistit/WarmupTest.java'
--- src/test/java/com/persistit/WarmupTest.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/WarmupTest.java	2012-08-07 21:14:18 +0000
@@ -0,0 +1,66 @@
+/**
+ * Copyright © 2012 Akiban Technologies, Inc.  All rights reserved.
+ * 
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Public License v1.0 which
+ * accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * This program may also be available under different license terms.
+ * For more information, see www.akiban.com or contact [e-mail].
+ * 
+ * Contributors:
+ * Akiban Technologies, Inc.
+ */
+
+package com.persistit;
+
+import static org.junit.Assert.*;
+import java.util.Properties;
+
+import org.junit.Test;
+
+import com.persistit.unit.PersistitUnitTestCase;
+
+public class WarmupTest extends PersistitUnitTestCase {
+	
+    @Test
+    public void testWarmup() throws Exception {
+        Exchange ex = _persistit.getExchange("persistit", "WarmupTest", true);
+        for (int i = 1; i <= 1000; i++) {
+            ex.getValue().put(RED_FOX);
+            ex.clear().append(i).store();
+        }
+        
+        // Assumption: only one buffer pool is created
+        int poolCount = 0;
+        String pathName = "";
+        Buffer[] buff = new Buffer[100];
+        for (BufferPool p: _persistit.getBufferPoolHashMap().values()) {
+        	poolCount = p.getBufferCount();
+                pathName = p.toString();
+        	for (int i = 0; i < poolCount; ++i) {
+        		buff[i] = p.getBufferCopy(i);
+        	}
+        }
+        
+        Properties properties = _persistit.getProperties();
+        ex = null;
+        _persistit.close();
+
+        _persistit = new Persistit();
+        _persistit.initialize(properties);
+        
+        int poolCount1 = 0;
+        for (BufferPool p: _persistit.getBufferPoolHashMap().values()) {
+        	poolCount1 = p.getBufferCount();
+        	for (int i = 0; i < poolCount1; ++i) {
+        		Buffer bufferCopy = p.getBufferCopy(i);
+        		assertEquals(bufferCopy.getPageAddress(), buff[i].getPageAddress());
+        		assertEquals(bufferCopy.getPageType(), buff[i].getPageType());
+        		assertEquals(bufferCopy.getBufferSize(), buff[i].getBufferSize());
+        	}
+        }
+        assertEquals(poolCount, poolCount1);         
+    }
+}
 
=== modified file 'src/test/java/com/persistit/unit/PersistitUnitTestCase.java'
--- src/test/java/com/persistit/unit/PersistitUnitTestCase.java	2012-08-02 04:45:28 +0000
+++ src/test/java/com/persistit/unit/PersistitUnitTestCase.java	2012-08-07 21:14:18 +0000
@@ -53,6 +53,7 @@
     @Before
     public void setUp() throws Exception {
         checkNoPersistitThreads();
+        
         _persistit.initialize(getProperties(true));
     }
 
@@ -83,7 +84,7 @@
         } catch (final Throwable t) {
             t.printStackTrace();
         } finally {
-            tearDown();
+           tearDown();
         }
     }
 
 
=== modified file 'src/test/java/com/persistit/unit/UnitTestProperties.java'
--- src/test/java/com/persistit/unit/UnitTestProperties.java	2012-08-02 04:45:28 +0000
+++ src/test/java/com/persistit/unit/UnitTestProperties.java	2012-08-07 21:14:18 +0000
@@ -43,6 +43,7 @@
         p.setProperty("tmpvoldir", "${datapath}");
         p.setProperty("rmiport", System.getProperty("rmiport", "8081"));
         p.setProperty("jmx", "true");
+        p.setProperty("bufferinventory", "/tmp/persistit_test_data");
         return p;
     }