[freenet-cvs] r19857 - branches/saltedhashstore/freenet/src/…

Author: j16sdiz
Date: 2008-05-09 02:09:10 +0000 (Fri, 09 May 2008)
New Revision: 19857

Modified:
branches/saltedhashstore/freenet/src/freenet/store/SaltedHashFreenetStore.java
Log:
Initial datastore resize code


Modified: branches/saltedhashstore/freenet/src/freenet/store/SaltedHashFreenetStore.java
===================================================================
--- branches/saltedhashstore/freenet/src/freenet/store/SaltedHashFreenetStore.java    2008-05-09 02:08:50 UTC (rev 19856)
+++ branches/saltedhashstore/freenet/src/freenet/store/SaltedHashFreenetStore.java    2008-05-09 02:09:10 UTC (rev 19857)
@@ -28,6 +28,9 @@
import freenet.support.io.FileUtil;
import freenet.support.math.RunningAverage;
import freenet.support.math.SimpleRunningAverage;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;

/**
* Index-less data store based on salted hash
@@ -36,7 +39,7 @@
*/
public class SaltedHashFreenetStore implements FreenetStore {
    private static final boolean OPTION_SAVE_PLAINKEY = true;
-    
+
    private static boolean logMINOR;
    private static boolean logDEBUG;

@@ -116,8 +119,12 @@
    }

    private Entry probeEntry(byte[] routingKey) throws IOException {
-        // TODO probe store resize
-        return probeEntry0(routingKey, storeSize);
+        Entry entry = probeEntry0(routingKey, storeSize);
+
+        if (entry == null && prevStoreSize != 0)
+            entry = probeEntry0(routingKey, prevStoreSize);
+
+        return entry;
    }

    private Entry probeEntry0(byte[] routingKey, long probeStoreSize) throws IOException {
@@ -132,7 +139,6 @@
            entry = readEntry(offset, routingKey);
        } catch (EOFException e) {
            // may occur on resize, silent it a bit
-            //TODO store resize
            Logger.error(this, "EOFException on probeEntry", e);
            return null;
        } finally {
@@ -252,7 +258,7 @@
            System.arraycopy(header, 0, this.header, 0, headerBlockLength);
            this.data = new byte[dataBlockLength];
            System.arraycopy(data, 0, this.data, 0, dataBlockLength);
-            
+
            if (OPTION_SAVE_PLAINKEY) {
                flag |= ENTRY_FLAG_PLAINKEY;
            }
@@ -260,6 +266,21 @@
            isEncrypted = false;
        }

+        /**
+         * @return the storeSize
+         */
+        protected long getStoreSize() {
+            return storeSize;
+        }
+
+        /**
+         * @param storeSize
+         * the storeSize to set
+         */
+        protected void setStoreSize(long storeSize) {
+            this.storeSize = storeSize;
+        }
+
        public Entry(ByteBuffer in) {
            assert in.remaining() == entryTotalLength;

@@ -276,7 +297,7 @@
                plainRoutingKey = new byte[0x20];
                in.get(plainRoutingKey);
            }
-            
+
            // reserved bytes
            in.position((int) ENTRY_HEADER_LENGTH);

@@ -300,10 +321,10 @@
            out.putLong(flag);
            out.putLong(storeSize);

-            if (OPTION_SAVE_PLAINKEY) {
+            if (OPTION_SAVE_PLAINKEY && plainRoutingKey != null) {
                out.put(plainRoutingKey);
            }
-            
+
            // reserved bytes
            out.position((int) ENTRY_HEADER_LENGTH);

@@ -356,7 +377,7 @@
                else
                    return false;
            }
-            
+
            if (plainRoutingKey != null) {
                // we knew the key
                if (!Arrays.equals(this.plainRoutingKey, routingKey)) {
@@ -416,6 +437,10 @@
                throw new RuntimeException(e);
            }
        }
+
+        public boolean isFree() {
+            return (flag & ENTRY_FLAG_OCCUPIED) == 0;
+        }
    }

    /**
@@ -435,13 +460,13 @@
            storeFiles[i] = new File(baseDir, name + ".data-" + fmt.format(i));

            storeRAF[i] = new RandomAccessFile(storeFiles[i], "rw");
-            //TODO store resize
-            if (storeRAF[i].length() == 0) { // New file?
-                storeRAF[i].setLength(entryTotalLength * (storeSize / FILE_SPLIT + 1));
-            }
+
            storeFC[i] = storeRAF[i].getChannel();
            storeFC[i].lock();
        }
+
+        long storeFileSize = Math.max(storeSize, prevStoreSize);
+        setStoreFileSize(storeFileSize);
    }

    /**
@@ -637,6 +662,11 @@
    private Cleaner cleanerThread;

    private class Cleaner extends Thread {
+        /**
+         * How often the clean should run
+         */
+        private static final int CLEANER_PERIOD = 10 * 60 * 1000; // 10 minutes
+
        public Cleaner() {
            setName("Store-" + name + "-Cleaner");
            setPriority(MIN_PRIORITY);
@@ -645,14 +675,15 @@

        public void run() {
            while (!shutdown) {
-                if (prevStoreSize != 0)
-                    moveOldEntries();
-                else
-                    estimateStoreSize();
+                synchronized (cleanerLock) {
+                    if (prevStoreSize != 0)
+                        resizeStore();
+                    else
+                        estimateStoreSize();

-                synchronized (cleanerLock) {
+                    cleanerLock.notifyAll();
                    try {
-                        cleanerLock.wait(10 * 60 * 1000); // 10 minutes
+                        cleanerLock.wait(CLEANER_PERIOD);
                    } catch (InterruptedException e) {
                        Logger.debug(this, "interrupted", e);
                    }
@@ -661,19 +692,227 @@
        }

        /**
-         * Move old entries to new location
+         * Maximum memory to be used on resize
         */
-        private void moveOldEntries() {
-            Logger.minor(this, "move old entries");
-            prevStoreSize = 0;
+        private static final int RESIZE_MEMORY = 2 * 1024 * 1024; // 2MiB
+        /**
+         * Phase 1 Rounds
+         */
+        private static final int RESIZE_PHASE1_ROUND = 12;
+        /**
+         * Maximum resize round
+         */
+        private static final int RESIZE_MAX_ROUND = 16;
+
+        /**
+         * Are we shrinking the store?
+         */
+        private boolean shrinking;
+        private long newEntries;
+        private long oldEntries;
+        private long freeEntries;
+        private long resolvedEntries;
+        private long droppedEntries;
+        private long maxOldItemOffset;
+
+        /**
+         * Numbers of round resize have ran
+         */
+        private int resizeRound;
+
+        /**
+         * Move old entries to new location and resize store
+         */
+        private void resizeStore() {
+            ++resizeRound;
+            Logger.normal(this, "Starting datastore resize (round " + resizeRound + ")");
+
+            if (resizeRound == 1) { // first round
+                if (storeSize < prevStoreSize) {
+                    shrinking = true;
+                } else {
+                    setStoreFileSize(storeSize);
+                }
+                maxOldItemOffset = prevStoreSize - 1;
+            }
+
+            moveOldEntry0(resizeRound > RESIZE_PHASE1_ROUND);
+
+            if (logMINOR)
+                Logger.minor(this, "Finished resize round " + resizeRound + ": newEntries=" + newEntries
+                 + ", oldEntries=" + oldEntries + ", freeEntries=" + freeEntries + ", resolvedEntries="
+                 + resolvedEntries + ", droppedEntries=" + droppedEntries);
+
+            if (shutdown)
+                return;
+
+            // report key count
+            estimatedCount.report(newEntries + oldEntries - droppedEntries);
+
+            // Shrink store file size
+            if (shrinking)
+                setStoreFileSize(Math.max(storeSize, maxOldItemOffset));
+
+            // Check finished
+            if (resizeRound >= RESIZE_MAX_ROUND || oldEntries == 0 || resolvedEntries + droppedEntries >= oldEntries) {
+                // Finished
+                Logger.normal(this, "Datastore resize finished (total " + resizeRound + "rounds)");
+
+                prevStoreSize = 0;
+                resizeRound = 0;
+            }
        }

        /**
-         * Sample to take at a time
+         * Scan all entries and try to move them
         */
-        private static final double SAMPLE_RATE = 0.05; // 5%
+        private void moveOldEntry0(boolean queueItem) {
+            newEntries = 0;
+            oldEntries = 0;
+            freeEntries = 0;
+            resolvedEntries = 0;
+            droppedEntries = 0;

+            List oldItems = null;
+            if (queueItem) {
+                oldItems = new ArrayList();
+            }
+
+            long maxOffset = maxOldItemOffset;
+            maxOldItemOffset = 0;
+            for (long offset = 0; offset <= maxOffset; offset++) {
+                if (logDEBUG && offset % 1024 == 0) {
+                    Logger.debug(this, "Resize progress: newEntries=" + newEntries + ", oldEntries=" + oldEntries
+                     + ", freeEntries=" + freeEntries + ", resolvedEntries=" + resolvedEntries
+                     + ", droppedEntries=" + droppedEntries);
+                }
+
+                if (shutdown)
+                    return;
+
+                if (!lockEntry(offset)) //lock
+                    continue;
+                try {
+                    Entry entry = readEntry(offset, null);
+
+                    if (entry.isFree()) {
+                        // free block
+                        freeEntries++;
+                    } else if (entry.getStoreSize() == storeSize) {
+                        // new store size entries
+                        maxOldItemOffset = offset;
+                        newEntries++;
+                    } else { // if (entry.getStoreSize() == prevStoreSize)
+                        // old store size entries, try to move them
+                        oldEntries++;
+                        maxOldItemOffset = offset;
+
+                        entry.setStoreSize(storeSize);
+                        long newOffset = entry.getOffset();
+
+                        if (newOffset == offset) { // lucky!
+                            writeEntry(entry); // write back entry storeSize
+                            resolvedEntries++;
+                            continue;
+                        }
+
+                        if (!lockEntry(newOffset)) // lock
+                            continue;
+                        try {
+                            // see what's in the new offset
+                            Entry newOffsetEntry = readEntry(newOffset, null);
+
+                            if (newOffsetEntry.isFree()) {
+                                // the new offset is freeeeeeee..
+                                writeEntry(entry);
+                                freeOffset(offset);
+                                resolvedEntries++;
+                            } else if (newOffsetEntry.getStoreSize() == storeSize) {
+                                // new offset already have a new entry, free old entry
+                                freeOffset(offset);
+                                droppedEntries++;
+                            } else if (Arrays.equals(entry.digestedRoutingKey, newOffsetEntry.digestedRoutingKey)) {
+                                // same digested routing key, free the old entry
+                                freeOffset(offset);
+                                resolvedEntries++;
+                            } else if (queueItem) {
+                                // break tie by moveing old item to queue
+                                if (oldItems.size() * entryTotalLength < RESIZE_MEMORY) {
+                                    oldItems.add(newOffsetEntry);
+                                    if (newOffset > offset) {
+                                        oldEntries++; // newOffset wasn't counted count it
+                                    }
+
+                                    writeEntry(entry);
+                                    freeOffset(offset);
+                                    resolvedEntries++;
+                                }
+                            }
+                        } finally {
+                            unlockEntry(newOffset);
+                        }
+                    }
+                } catch (IOException e) {
+                    Logger.debug(this, "IOExcception on moveOldEntries0", e);
+                } finally {
+                    unlockEntry(offset);
+                }
+            }
+
+            if (queueItem) {
+                putBackOldItems(oldItems);
+            }
+        }
+
        /**
+         * Put back oldItems with best effort
+         */
+        private void putBackOldItems(List oldItems) {
+            Iterator it = oldItems.iterator();
+            while (it.hasNext()) {
+                boolean done = false;
+
+                Entry entry = (Entry) it.next();
+                entry.setStoreSize(storeSize);
+
+                long newOffset = entry.getOffset();
+
+                if (!lockEntry(newOffset)) // lock
+                    continue;
+                try {
+                    if (isFree(newOffset)) {
+                        if (logDEBUG)
+                            Logger.debug(this, "Put back old item: " + HexUtil.bytesToHex(entry.digestedRoutingKey));
+                        writeEntry(entry);
+                        done = true;
+                    } else {
+                        if (logDEBUG)
+                            Logger.debug(this, "Drop old item: " + HexUtil.bytesToHex(entry.digestedRoutingKey));
+                    }
+                } catch (IOException e) {
+                    Logger.debug(this, "IOExcception on putBackOldItems", e);
+                } finally {
+                    unlockEntry(newOffset);
+                    it.remove();
+
+                    if (done)
+                        resolvedEntries++;
+                    else
+                        droppedEntries++;
+                }
+            }
+            oldItems.clear();
+        }
+
+        /**
+         * Samples to take on key count estimation
+         */
+        private static final double SAMPLE_RATE = 0.05; // 5%
+        /**
+         * Minimum samples to take on key count estimation
+         */
+        private static final int MIN_SAMPLE = 10000;
+        /**
         * Last sample position
         */
        private long samplePos = 0;
@@ -683,7 +922,7 @@
         */
        private void estimateStoreSize() {
            Logger.minor(this, "start estimating key count");
-            long numSample = Math.min((long) (SAMPLE_RATE * storeSize), 10000);
+            long numSample = Math.min((long) (SAMPLE_RATE * storeSize), MIN_SAMPLE);
            long sampled = 0;
            long occupied = 0;
            while (sampled < numSample) {
@@ -712,9 +951,32 @@
        }
    }

-    public void setMaxKeys(long maxStoreKeys, boolean shrinkNow) throws IOException {
-        // TODO
-        // NO-OP now
+    public void setMaxKeys(long newStoreSize, boolean shrinkNow) throws IOException {
+        Logger.normal(this, "[" + name + "] Resize newStoreSize=" + newStoreSize + ", shinkNow=" + shrinkNow);
+
+        assert newStoreSize > 0;
+
+        synchronized (cleanerLock) {
+            if (newStoreSize == this.storeSize)
+                return;
+
+            if (prevStoreSize != 0) {
+                if (shrinkNow) {
+                    // TODO shrink now
+                } else {
+                    Logger.normal(this, "[" + name + "] resize already in progress, ignore resize request");
+                    return;
+                }
+            }
+
+            prevStoreSize = storeSize;
+            storeSize = newStoreSize;
+            cleanerLock.notifyAll();
+
+            if (shrinkNow) {
+                // TODO shrink now
+            }
+        }
    }

    // ------------- Locking
@@ -941,4 +1203,19 @@
    public long getMaxKeys() {
        return storeSize;
    }
+
+    /**
+     * Change on disk store file size
+     *
+     * @param storeFileSize
+     */
+    private void setStoreFileSize(long storeFileSize) {
+        for (int i = 0; i < FILE_SPLIT; i++) {
+            try {
+                storeRAF[i].setLength(entryTotalLength * (storeFileSize / FILE_SPLIT + 1));
+            } catch (IOException e) {
+                Logger.error(this, "error resizing store file", e);
+            }
+        }
+    }
}