001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.store.kahadb.disk.page;
018
019import java.io.ByteArrayInputStream;
020import java.io.ByteArrayOutputStream;
021import java.io.DataInputStream;
022import java.io.DataOutputStream;
023import java.io.File;
024import java.io.FileInputStream;
025import java.io.FileOutputStream;
026import java.io.IOException;
027import java.io.InterruptedIOException;
028import java.io.RandomAccessFile;
029import java.util.ArrayList;
030import java.util.Arrays;
031import java.util.Collection;
032import java.util.Collections;
033import java.util.HashMap;
034import java.util.Iterator;
035import java.util.LinkedHashMap;
036import java.util.Map;
037import java.util.Map.Entry;
038import java.util.Properties;
039import java.util.TreeMap;
040import java.util.concurrent.CountDownLatch;
041import java.util.concurrent.atomic.AtomicBoolean;
042import java.util.concurrent.atomic.AtomicLong;
043import java.util.concurrent.atomic.AtomicReference;
044import java.util.zip.Adler32;
045import java.util.zip.Checksum;
046
047import org.apache.activemq.store.kahadb.disk.util.Sequence;
048import org.apache.activemq.store.kahadb.disk.util.SequenceSet;
049import org.apache.activemq.util.DataByteArrayOutputStream;
050import org.apache.activemq.util.IOExceptionSupport;
051import org.apache.activemq.util.IOHelper;
052import org.apache.activemq.util.IntrospectionSupport;
053import org.apache.activemq.util.LFUCache;
054import org.apache.activemq.util.LRUCache;
055import org.apache.activemq.util.RecoverableRandomAccessFile;
056import org.slf4j.Logger;
057import org.slf4j.LoggerFactory;
058
059/**
060 * A PageFile provides you random access to fixed sized disk pages. This object is not thread safe and therefore access to it should
061 * be externally synchronized.
062 * <p/>
063 * The file has 3 parts:
064 * Metadata Space: 4k : Reserved metadata area. Used to store persistent config about the file.
065 * Recovery Buffer Space: Page Size * 1000 : This is a redo log used to prevent partial page writes from making the file inconsistent
066 * Page Space: The pages in the page file.
067 */
068public class PageFile {
069
070    private static final String PAGEFILE_SUFFIX = ".data";
071    private static final String RECOVERY_FILE_SUFFIX = ".redo";
072    private static final String FREE_FILE_SUFFIX = ".free";
073
074    // 4k Default page size.
075    public static final int DEFAULT_PAGE_SIZE = Integer.getInteger("defaultPageSize", 1024*4);
076    public static final int DEFAULT_WRITE_BATCH_SIZE = Integer.getInteger("defaultWriteBatchSize", 1000);
077    public static final int DEFAULT_PAGE_CACHE_SIZE = Integer.getInteger("defaultPageCacheSize", 100);;
078
079    private static final int RECOVERY_FILE_HEADER_SIZE = 1024 * 4;
080    private static final int PAGE_FILE_HEADER_SIZE = 1024 * 4;
081
082    // Recovery header is (long offset)
083    private static final Logger LOG = LoggerFactory.getLogger(PageFile.class);
084
085    // A PageFile will use a couple of files in this directory
086    private final File directory;
087    // And the file names in that directory will be based on this name.
088    private final String name;
089
090    // File handle used for reading pages..
091    private RecoverableRandomAccessFile readFile;
092    // File handle used for writing pages..
093    private RecoverableRandomAccessFile writeFile;
094    // File handle used for writing pages..
095    private RecoverableRandomAccessFile recoveryFile;
096
097    // The size of pages
098    private int pageSize = DEFAULT_PAGE_SIZE;
099
100    // The minimum number of space allocated to the recovery file in number of pages.
101    private int recoveryFileMinPageCount = 1000;
102    // The max size that we let the recovery file grow to.. ma exceed the max, but the file will get resize
103    // to this max size as soon as  possible.
104    private int recoveryFileMaxPageCount = 10000;
105    // The number of pages in the current recovery buffer
106    private int recoveryPageCount;
107
108    private final AtomicBoolean loaded = new AtomicBoolean();
109    // The number of pages we are aiming to write every time we
110    // write to disk.
111    int writeBatchSize = DEFAULT_WRITE_BATCH_SIZE;
112
113    // We keep a cache of pages recently used?
114    private Map<Long, Page> pageCache;
115    // The cache of recently used pages.
116    private boolean enablePageCaching = true;
117    // How many pages will we keep in the cache?
118    private int pageCacheSize = DEFAULT_PAGE_CACHE_SIZE;
119
120    // Should first log the page write to the recovery buffer? Avoids partial
121    // page write failures..
122    private boolean enableRecoveryFile = true;
123    // Will we sync writes to disk. Ensures that data will not be lost after a checkpoint()
124    private boolean enableDiskSyncs = true;
125    // Will writes be done in an async thread?
126    private boolean enabledWriteThread = false;
127
128    // These are used if enableAsyncWrites==true
129    private final AtomicBoolean stopWriter = new AtomicBoolean();
130    private Thread writerThread;
131    private CountDownLatch checkpointLatch;
132
133    // Keeps track of writes that are being written to disk.
134    private final TreeMap<Long, PageWrite> writes = new TreeMap<Long, PageWrite>();
135
136    // Keeps track of free pages.
137    private final AtomicLong nextFreePageId = new AtomicLong();
138    private SequenceSet freeList = new SequenceSet();
139
140    private AtomicReference<SequenceSet> recoveredFreeList = new AtomicReference<SequenceSet>();
141    private AtomicReference<SequenceSet> trackingFreeDuringRecovery = new AtomicReference<SequenceSet>();
142
143    private final AtomicLong nextTxid = new AtomicLong();
144
145    // Persistent settings stored in the page file.
146    private MetaData metaData;
147
148    private final HashMap<File, RandomAccessFile> tmpFilesForRemoval = new HashMap<>();
149
150    private boolean useLFRUEviction = false;
151    private float LFUEvictionFactor = 0.2f;
152
153    /**
154     * Use to keep track of updated pages which have not yet been committed.
155     */
156    static class PageWrite {
157        Page page;
158        byte[] current;
159        byte[] diskBound;
160        long currentLocation = -1;
161        long diskBoundLocation = -1;
162        File tmpFile;
163        int length;
164
165        public PageWrite(Page page, byte[] data) {
166            this.page = page;
167            current = data;
168        }
169
170        public PageWrite(Page page, long currentLocation, int length, File tmpFile) {
171            this.page = page;
172            this.currentLocation = currentLocation;
173            this.tmpFile = tmpFile;
174            this.length = length;
175        }
176
177        public void setCurrent(Page page, byte[] data) {
178            this.page = page;
179            current = data;
180            currentLocation = -1;
181        }
182
183        public void setCurrentLocation(Page page, long location, int length) {
184            this.page = page;
185            this.currentLocation = location;
186            this.length = length;
187            this.current = null;
188        }
189
190        @Override
191        public String toString() {
192            return "[PageWrite:" + page.getPageId() + "-" + page.getType() + "]";
193        }
194
195        @SuppressWarnings("unchecked")
196        public Page getPage() {
197            return page;
198        }
199
200        public byte[] getDiskBound(HashMap<File, RandomAccessFile> tmpFiles) throws IOException {
201            if (diskBound == null && diskBoundLocation != -1) {
202                diskBound = new byte[length];
203                if (tmpFiles.containsKey(tmpFile) && tmpFiles.get(tmpFile).getChannel().isOpen()) {
204                    RandomAccessFile file = tmpFiles.get(tmpFile);
205                    file.seek(diskBoundLocation);
206                    file.read(diskBound);
207                } else {
208                    try (RandomAccessFile file = new RandomAccessFile(tmpFile, "r")) {
209                        file.seek(diskBoundLocation);
210                        file.read(diskBound);
211                    }
212                }
213                diskBoundLocation = -1;
214            }
215            return diskBound;
216        }
217
218        void begin() {
219            if (currentLocation != -1) {
220                diskBoundLocation = currentLocation;
221            } else {
222                diskBound = current;
223            }
224            current = null;
225            currentLocation = -1;
226        }
227
228        /**
229         * @return true if there is no pending writes to do.
230         */
231        boolean done() {
232            diskBoundLocation = -1;
233            diskBound = null;
234            return current == null || currentLocation == -1;
235        }
236
237        boolean isDone() {
238            return diskBound == null && diskBoundLocation == -1 && current == null && currentLocation == -1;
239        }
240    }
241
242    /**
243     * The MetaData object hold the persistent data associated with a PageFile object.
244     */
245    public static class MetaData {
246
247        String fileType;
248        String fileTypeVersion;
249
250        long metaDataTxId = -1;
251        int pageSize;
252        boolean cleanShutdown;
253        long lastTxId;
254        long freePages;
255
256        public String getFileType() {
257            return fileType;
258        }
259
260        public void setFileType(String fileType) {
261            this.fileType = fileType;
262        }
263
264        public String getFileTypeVersion() {
265            return fileTypeVersion;
266        }
267
268        public void setFileTypeVersion(String version) {
269            this.fileTypeVersion = version;
270        }
271
272        public long getMetaDataTxId() {
273            return metaDataTxId;
274        }
275
276        public void setMetaDataTxId(long metaDataTxId) {
277            this.metaDataTxId = metaDataTxId;
278        }
279
280        public int getPageSize() {
281            return pageSize;
282        }
283
284        public void setPageSize(int pageSize) {
285            this.pageSize = pageSize;
286        }
287
288        public boolean isCleanShutdown() {
289            return cleanShutdown;
290        }
291
292        public void setCleanShutdown(boolean cleanShutdown) {
293            this.cleanShutdown = cleanShutdown;
294        }
295
296        public long getLastTxId() {
297            return lastTxId;
298        }
299
300        public void setLastTxId(long lastTxId) {
301            this.lastTxId = lastTxId;
302        }
303
304        public long getFreePages() {
305            return freePages;
306        }
307
308        public void setFreePages(long value) {
309            this.freePages = value;
310        }
311    }
312
313    public Transaction tx() {
314        assertLoaded();
315        return new Transaction(this);
316    }
317
318    /**
319     * Creates a PageFile in the specified directory who's data files are named by name.
320     */
321    public PageFile(File directory, String name) {
322        this.directory = directory;
323        this.name = name;
324    }
325
326    /**
327     * Deletes the files used by the PageFile object.  This method can only be used when this object is not loaded.
328     *
329     * @throws IOException           if the files cannot be deleted.
330     * @throws IllegalStateException if this PageFile is loaded
331     */
332    public void delete() throws IOException {
333        if (loaded.get()) {
334            throw new IllegalStateException("Cannot delete page file data when the page file is loaded");
335        }
336        delete(getMainPageFile());
337        delete(getFreeFile());
338        delete(getRecoveryFile());
339    }
340
341    public void archive() throws IOException {
342        if (loaded.get()) {
343            throw new IllegalStateException("Cannot delete page file data when the page file is loaded");
344        }
345        long timestamp = System.currentTimeMillis();
346        archive(getMainPageFile(), String.valueOf(timestamp));
347        archive(getFreeFile(), String.valueOf(timestamp));
348        archive(getRecoveryFile(), String.valueOf(timestamp));
349    }
350
351    /**
352     * @param file
353     * @throws IOException
354     */
355    private void delete(File file) throws IOException {
356        if (file.exists() && !file.delete()) {
357            throw new IOException("Could not delete: " + file.getPath());
358        }
359    }
360
361    private void archive(File file, String suffix) throws IOException {
362        if (file.exists()) {
363            File archive = new File(file.getPath() + "-" + suffix);
364            if (!file.renameTo(archive)) {
365                throw new IOException("Could not archive: " + file.getPath() + " to " + file.getPath());
366            }
367        }
368    }
369
370    /**
371     * Loads the page file so that it can be accessed for read/write purposes.  This allocates OS resources.  If this is the
372     * first time the page file is loaded, then this creates the page file in the file system.
373     *
374     * @throws IOException           If the page file cannot be loaded. This could be cause the existing page file is corrupt is a bad version or if
375     *                               there was a disk error.
376     * @throws IllegalStateException If the page file was already loaded.
377     */
378    public void load() throws IOException, IllegalStateException {
379        if (loaded.compareAndSet(false, true)) {
380
381            if (enablePageCaching) {
382                if (isUseLFRUEviction()) {
383                    pageCache = Collections.synchronizedMap(new LFUCache<Long, Page>(pageCacheSize, getLFUEvictionFactor()));
384                } else {
385                    pageCache = Collections.synchronizedMap(new LRUCache<Long, Page>(pageCacheSize, pageCacheSize, 0.75f, true));
386                }
387            }
388
389            File file = getMainPageFile();
390            IOHelper.mkdirs(file.getParentFile());
391            writeFile = new RecoverableRandomAccessFile(file, "rw", false);
392            readFile = new RecoverableRandomAccessFile(file, "r");
393
394            if (readFile.length() > 0) {
395                // Load the page size setting cause that can't change once the file is created.
396                loadMetaData();
397                pageSize = metaData.getPageSize();
398            } else {
399                // Store the page size setting cause that can't change once the file is created.
400                metaData = new MetaData();
401                metaData.setFileType(PageFile.class.getName());
402                metaData.setFileTypeVersion("1");
403                metaData.setPageSize(getPageSize());
404                metaData.setCleanShutdown(true);
405                metaData.setFreePages(-1);
406                metaData.setLastTxId(0);
407                storeMetaData();
408            }
409
410            if (enableRecoveryFile) {
411                recoveryFile = new RecoverableRandomAccessFile(getRecoveryFile(), "rw");
412            }
413
414            if (metaData.isCleanShutdown()) {
415                nextTxid.set(metaData.getLastTxId() + 1);
416                if (metaData.getFreePages() > 0) {
417                    loadFreeList();
418                }
419            } else {
420                LOG.debug(toString() + ", Recovering page file...");
421                nextTxid.set(redoRecoveryUpdates());
422                trackingFreeDuringRecovery.set(new SequenceSet());
423            }
424
425            if (writeFile.length() < PAGE_FILE_HEADER_SIZE) {
426                writeFile.setLength(PAGE_FILE_HEADER_SIZE);
427            }
428            nextFreePageId.set((writeFile.length() - PAGE_FILE_HEADER_SIZE) / pageSize);
429
430            metaData.setCleanShutdown(false);
431            storeMetaData();
432            getFreeFile().delete();
433            startWriter();
434            if (trackingFreeDuringRecovery.get() != null) {
435                asyncFreePageRecovery(nextFreePageId.get());
436            }
437        } else {
438            throw new IllegalStateException("Cannot load the page file when it is already loaded.");
439        }
440    }
441
442    private void asyncFreePageRecovery(final long lastRecoveryPage) {
443        Thread thread = new Thread("KahaDB Index Free Page Recovery") {
444            @Override
445            public void run() {
446                try {
447                    recoverFreePages(lastRecoveryPage);
448                } catch (Throwable e) {
449                    if (loaded.get()) {
450                        LOG.warn("Error recovering index free page list", e);
451                    }
452                }
453            }
454        };
455        thread.setPriority(Thread.NORM_PRIORITY);
456        thread.setDaemon(true);
457        thread.start();
458    }
459
460    private void recoverFreePages(final long lastRecoveryPage) throws Exception {
461        LOG.info(toString() + ". Recovering pageFile free list due to prior unclean shutdown..");
462        SequenceSet newFreePages = new SequenceSet();
463        // need new pageFile instance to get unshared readFile
464        PageFile recoveryPageFile = new PageFile(directory, name);
465        recoveryPageFile.loadForRecovery(nextFreePageId.get());
466        try {
467            for (Iterator<Page> i = new Transaction(recoveryPageFile).iterator(true); i.hasNext(); ) {
468                Page page = i.next();
469
470                if (page.getPageId() >= lastRecoveryPage) {
471                    break;
472                }
473
474                if (page.getType() == Page.PAGE_FREE_TYPE) {
475                    newFreePages.add(page.getPageId());
476                }
477            }
478        } finally {
479            recoveryPageFile.readFile.close();
480        }
481
482        LOG.info(toString() + ". Recovered pageFile free list of size: " + newFreePages.rangeSize());
483        if (!newFreePages.isEmpty()) {
484
485            // allow flush (with index lock held) to merge eventually
486            recoveredFreeList.lazySet(newFreePages);
487        } else {
488            // If there is no free pages, set trackingFreeDuringRecovery to allow the broker to have a clean shutdown
489            trackingFreeDuringRecovery.set(null);
490        }
491    }
492
493    private void loadForRecovery(long nextFreePageIdSnap) throws Exception {
494        loaded.set(true);
495        enablePageCaching = false;
496        File file = getMainPageFile();
497        readFile = new RecoverableRandomAccessFile(file, "r");
498        loadMetaData();
499        pageSize = metaData.getPageSize();
500        enableRecoveryFile = false;
501        nextFreePageId.set(nextFreePageIdSnap);
502    }
503
504
505    /**
506     * Unloads a previously loaded PageFile.  This deallocates OS related resources like file handles.
507     * once unloaded, you can no longer use the page file to read or write Pages.
508     *
509     * @throws IOException           if there was a disk error occurred while closing the down the page file.
510     * @throws IllegalStateException if the PageFile is not loaded
511     */
512    public void unload() throws IOException {
513        if (loaded.compareAndSet(true, false)) {
514            flush();
515            try {
516                stopWriter();
517            } catch (InterruptedException e) {
518                throw new InterruptedIOException();
519            }
520
521            if (freeList.isEmpty()) {
522                metaData.setFreePages(0);
523            } else {
524                storeFreeList();
525                metaData.setFreePages(freeList.size());
526            }
527
528            metaData.setLastTxId(nextTxid.get() - 1);
529            if (trackingFreeDuringRecovery.get() != null) {
530                // async recovery incomplete, will have to try again
531                metaData.setCleanShutdown(false);
532            } else {
533                metaData.setCleanShutdown(true);
534            }
535            storeMetaData();
536
537            if (readFile != null) {
538                readFile.close();
539                readFile = null;
540                writeFile.close();
541                writeFile = null;
542                if (enableRecoveryFile) {
543                    recoveryFile.close();
544                    recoveryFile = null;
545                }
546                freeList.clear();
547                if (pageCache != null) {
548                    pageCache = null;
549                }
550                synchronized (writes) {
551                    writes.clear();
552                }
553            }
554        } else {
555            throw new IllegalStateException("Cannot unload the page file when it is not loaded");
556        }
557    }
558
559    public boolean isLoaded() {
560        return loaded.get();
561    }
562
563    public boolean isCleanShutdown() {
564        return metaData != null && metaData.isCleanShutdown();
565    }
566
567    public void allowIOResumption() {
568        loaded.set(true);
569    }
570
571    /**
572     * Flush and sync all write buffers to disk.
573     *
574     * @throws IOException If an disk error occurred.
575     */
576    public void flush() throws IOException {
577
578        if (enabledWriteThread && stopWriter.get()) {
579            throw new IOException("Page file already stopped: checkpointing is not allowed");
580        }
581
582        SequenceSet recovered = recoveredFreeList.get();
583        if (recovered != null) {
584            recoveredFreeList.lazySet(null);
585            SequenceSet inUse = trackingFreeDuringRecovery.get();
586            recovered.remove(inUse);
587            freeList.merge(recovered);
588
589            // all set for clean shutdown
590            trackingFreeDuringRecovery.set(null);
591            inUse.clear();
592        }
593
594        // Setup a latch that gets notified when all buffered writes hits the disk.
595        CountDownLatch checkpointLatch;
596        synchronized (writes) {
597            if (writes.isEmpty()) {
598                return;
599            }
600            if (enabledWriteThread) {
601                if (this.checkpointLatch == null) {
602                    this.checkpointLatch = new CountDownLatch(1);
603                }
604                checkpointLatch = this.checkpointLatch;
605                writes.notify();
606            } else {
607                writeBatch();
608                return;
609            }
610        }
611        try {
612            checkpointLatch.await();
613        } catch (InterruptedException e) {
614            InterruptedIOException ioe = new InterruptedIOException();
615            ioe.initCause(e);
616            throw ioe;
617        }
618    }
619
620
621    @Override
622    public String toString() {
623        return "Page File: " + getMainPageFile();
624    }
625
626    ///////////////////////////////////////////////////////////////////
627    // Private Implementation Methods
628    ///////////////////////////////////////////////////////////////////
629    private File getMainPageFile() {
630        return new File(directory, IOHelper.toFileSystemSafeName(name) + PAGEFILE_SUFFIX);
631    }
632
633    public File getFreeFile() {
634        return new File(directory, IOHelper.toFileSystemSafeName(name) + FREE_FILE_SUFFIX);
635    }
636
637    public File getRecoveryFile() {
638        return new File(directory, IOHelper.toFileSystemSafeName(name) + RECOVERY_FILE_SUFFIX);
639    }
640
641    public long toOffset(long pageId) {
642        return PAGE_FILE_HEADER_SIZE + (pageId * pageSize);
643    }
644
645    private void loadMetaData() throws IOException {
646
647        ByteArrayInputStream is;
648        MetaData v1 = new MetaData();
649        MetaData v2 = new MetaData();
650        try {
651            Properties p = new Properties();
652            byte[] d = new byte[PAGE_FILE_HEADER_SIZE / 2];
653            readFile.seek(0);
654            readFile.readFully(d);
655            is = new ByteArrayInputStream(d);
656            p.load(is);
657            IntrospectionSupport.setProperties(v1, p);
658        } catch (IOException e) {
659            v1 = null;
660        }
661
662        try {
663            Properties p = new Properties();
664            byte[] d = new byte[PAGE_FILE_HEADER_SIZE / 2];
665            readFile.seek(PAGE_FILE_HEADER_SIZE / 2);
666            readFile.readFully(d);
667            is = new ByteArrayInputStream(d);
668            p.load(is);
669            IntrospectionSupport.setProperties(v2, p);
670        } catch (IOException e) {
671            v2 = null;
672        }
673
674        if (v1 == null && v2 == null) {
675            throw new IOException("Could not load page file meta data");
676        }
677
678        if (v1 == null || v1.metaDataTxId < 0) {
679            metaData = v2;
680        } else if (v2 == null || v1.metaDataTxId < 0) {
681            metaData = v1;
682        } else if (v1.metaDataTxId == v2.metaDataTxId) {
683            metaData = v1; // use the first since the 2nd could be a partial..
684        } else {
685            metaData = v2; // use the second cause the first is probably a partial.
686        }
687    }
688
689    private void storeMetaData() throws IOException {
690        // Convert the metadata into a property format
691        metaData.metaDataTxId++;
692        Properties p = new Properties();
693        IntrospectionSupport.getProperties(metaData, p, null);
694
695        ByteArrayOutputStream os = new ByteArrayOutputStream(PAGE_FILE_HEADER_SIZE);
696        p.store(os, "");
697        if (os.size() > PAGE_FILE_HEADER_SIZE / 2) {
698            throw new IOException("Configuation is larger than: " + PAGE_FILE_HEADER_SIZE / 2);
699        }
700        // Fill the rest with space...
701        byte[] filler = new byte[(PAGE_FILE_HEADER_SIZE / 2) - os.size()];
702        Arrays.fill(filler, (byte) ' ');
703        os.write(filler);
704        os.flush();
705
706        byte[] d = os.toByteArray();
707
708        // So we don't loose it.. write it 2 times...
709        writeFile.seek(0);
710        writeFile.write(d);
711        writeFile.sync();
712        writeFile.seek(PAGE_FILE_HEADER_SIZE / 2);
713        writeFile.write(d);
714        writeFile.sync();
715    }
716
717    private void storeFreeList() throws IOException {
718        FileOutputStream os = new FileOutputStream(getFreeFile());
719        DataOutputStream dos = new DataOutputStream(os);
720        SequenceSet.Marshaller.INSTANCE.writePayload(freeList, dos);
721        dos.close();
722    }
723
724    private void loadFreeList() throws IOException {
725        freeList.clear();
726        FileInputStream is = new FileInputStream(getFreeFile());
727        DataInputStream dis = new DataInputStream(is);
728        freeList = SequenceSet.Marshaller.INSTANCE.readPayload(dis);
729        dis.close();
730    }
731
732    ///////////////////////////////////////////////////////////////////
733    // Property Accessors
734    ///////////////////////////////////////////////////////////////////
735
736    /**
737     * Is the recovery buffer used to double buffer page writes.  Enabled by default.
738     *
739     * @return is the recovery buffer enabled.
740     */
741    public boolean isEnableRecoveryFile() {
742        return enableRecoveryFile;
743    }
744
745    /**
746     * Sets if the recovery buffer uses to double buffer page writes.  Enabled by default.  Disabling this
747     * may potentially cause partial page writes which can lead to page file corruption.
748     */
749    public void setEnableRecoveryFile(boolean doubleBuffer) {
750        assertNotLoaded();
751        this.enableRecoveryFile = doubleBuffer;
752    }
753
754    /**
755     * @return Are page writes synced to disk?
756     */
757    public boolean isEnableDiskSyncs() {
758        return enableDiskSyncs;
759    }
760
761    /**
762     * Allows you enable syncing writes to disk.
763     */
764    public void setEnableDiskSyncs(boolean syncWrites) {
765        assertNotLoaded();
766        this.enableDiskSyncs = syncWrites;
767    }
768
769    /**
770     * @return the page size
771     */
772    public int getPageSize() {
773        return this.pageSize;
774    }
775
776    /**
777     * @return the amount of content data that a page can hold.
778     */
779    public int getPageContentSize() {
780        return this.pageSize - Page.PAGE_HEADER_SIZE;
781    }
782
783    /**
784     * Configures the page size used by the page file.  By default it is 4k.  Once a page file is created on disk,
785     * subsequent loads of that file will use the original pageSize.  Once the PageFile is loaded, this setting
786     * can no longer be changed.
787     *
788     * @param pageSize the pageSize to set
789     * @throws IllegalStateException once the page file is loaded.
790     */
791    public void setPageSize(int pageSize) throws IllegalStateException {
792        assertNotLoaded();
793        this.pageSize = pageSize;
794    }
795
796    /**
797     * @return true if read page caching is enabled
798     */
799    public boolean isEnablePageCaching() {
800        return this.enablePageCaching;
801    }
802
803    /**
804     * @param enablePageCaching allows you to enable read page caching
805     */
806    public void setEnablePageCaching(boolean enablePageCaching) {
807        assertNotLoaded();
808        this.enablePageCaching = enablePageCaching;
809    }
810
811    /**
812     * @return the maximum number of pages that will get stored in the read page cache.
813     */
814    public int getPageCacheSize() {
815        return this.pageCacheSize;
816    }
817
818    /**
819     * @param pageCacheSize Sets the maximum number of pages that will get stored in the read page cache.
820     */
821    public void setPageCacheSize(int pageCacheSize) {
822        assertNotLoaded();
823        this.pageCacheSize = pageCacheSize;
824    }
825
826    public boolean isEnabledWriteThread() {
827        return enabledWriteThread;
828    }
829
830    public void setEnableWriteThread(boolean enableAsyncWrites) {
831        assertNotLoaded();
832        this.enabledWriteThread = enableAsyncWrites;
833    }
834
835    public long getDiskSize() throws IOException {
836        return toOffset(nextFreePageId.get());
837    }
838
839    public boolean isFreePage(long pageId) {
840        return freeList.contains(pageId);
841    }
842    /**
843     * @return the number of pages allocated in the PageFile
844     */
845    public long getPageCount() {
846        return nextFreePageId.get();
847    }
848
849    public int getRecoveryFileMinPageCount() {
850        return recoveryFileMinPageCount;
851    }
852
853    public long getFreePageCount() {
854        assertLoaded();
855        return freeList.rangeSize();
856    }
857
858    public void setRecoveryFileMinPageCount(int recoveryFileMinPageCount) {
859        assertNotLoaded();
860        this.recoveryFileMinPageCount = recoveryFileMinPageCount;
861    }
862
863    public int getRecoveryFileMaxPageCount() {
864        return recoveryFileMaxPageCount;
865    }
866
867    public void setRecoveryFileMaxPageCount(int recoveryFileMaxPageCount) {
868        assertNotLoaded();
869        this.recoveryFileMaxPageCount = recoveryFileMaxPageCount;
870    }
871
872    public int getWriteBatchSize() {
873        return writeBatchSize;
874    }
875
876    public void setWriteBatchSize(int writeBatchSize) {
877        this.writeBatchSize = writeBatchSize;
878    }
879
880    public float getLFUEvictionFactor() {
881        return LFUEvictionFactor;
882    }
883
884    public void setLFUEvictionFactor(float LFUEvictionFactor) {
885        this.LFUEvictionFactor = LFUEvictionFactor;
886    }
887
888    public boolean isUseLFRUEviction() {
889        return useLFRUEviction;
890    }
891
892    public void setUseLFRUEviction(boolean useLFRUEviction) {
893        this.useLFRUEviction = useLFRUEviction;
894    }
895
896    ///////////////////////////////////////////////////////////////////
897    // Package Protected Methods exposed to Transaction
898    ///////////////////////////////////////////////////////////////////
899
900    /**
901     * @throws IllegalStateException if the page file is not loaded.
902     */
903    void assertLoaded() throws IllegalStateException {
904        if (!loaded.get()) {
905            throw new IllegalStateException("PageFile is not loaded");
906        }
907    }
908
909    void assertNotLoaded() throws IllegalStateException {
910        if (loaded.get()) {
911            throw new IllegalStateException("PageFile is loaded");
912        }
913    }
914
915    /**
916     * Allocates a block of free pages that you can write data to.
917     *
918     * @param count the number of sequential pages to allocate
919     * @return the first page of the sequential set.
920     * @throws IOException           If an disk error occurred.
921     * @throws IllegalStateException if the PageFile is not loaded
922     */
923    <T> Page<T> allocate(int count) throws IOException {
924        assertLoaded();
925        if (count <= 0) {
926            throw new IllegalArgumentException("The allocation count must be larger than zero");
927        }
928
929        Sequence seq = freeList.removeFirstSequence(count);
930
931        // We may need to create new free pages...
932        if (seq == null) {
933
934            Page<T> first = null;
935            int c = count;
936
937            // Perform the id's only once....
938            long pageId = nextFreePageId.getAndAdd(count);
939            long writeTxnId = nextTxid.getAndAdd(count);
940
941            while (c-- > 0) {
942                Page<T> page = new Page<T>(pageId++);
943                page.makeFree(writeTxnId++);
944
945                if (first == null) {
946                    first = page;
947                }
948
949                addToCache(page);
950                DataByteArrayOutputStream out = new DataByteArrayOutputStream(pageSize);
951                page.write(out);
952                write(page, out.getData());
953
954                // LOG.debug("allocate writing: "+page.getPageId());
955            }
956
957            return first;
958        }
959
960        Page<T> page = new Page<T>(seq.getFirst());
961        page.makeFree(0);
962        // LOG.debug("allocated: "+page.getPageId());
963        return page;
964    }
965
966    long getNextWriteTransactionId() {
967        return nextTxid.incrementAndGet();
968    }
969
970    synchronized void readPage(long pageId, byte[] data) throws IOException {
971        readFile.seek(toOffset(pageId));
972        readFile.readFully(data);
973    }
974
975    public void freePage(long pageId) {
976        freeList.add(pageId);
977        removeFromCache(pageId);
978
979        SequenceSet trackFreeDuringRecovery = trackingFreeDuringRecovery.get();
980        if (trackFreeDuringRecovery != null) {
981            trackFreeDuringRecovery.add(pageId);
982        }
983    }
984
985    @SuppressWarnings("unchecked")
986    private <T> void write(Page<T> page, byte[] data) throws IOException {
987        final PageWrite write = new PageWrite(page, data);
988        Entry<Long, PageWrite> entry = new Entry<Long, PageWrite>() {
989            @Override
990            public Long getKey() {
991                return write.getPage().getPageId();
992            }
993
994            @Override
995            public PageWrite getValue() {
996                return write;
997            }
998
999            @Override
1000            public PageWrite setValue(PageWrite value) {
1001                return null;
1002            }
1003        };
1004        Entry<Long, PageWrite>[] entries = new Map.Entry[]{entry};
1005        write(Arrays.asList(entries));
1006    }
1007
1008    void write(Collection<Map.Entry<Long, PageWrite>> updates) throws IOException {
1009        synchronized (writes) {
1010            if (enabledWriteThread) {
1011                while (writes.size() >= writeBatchSize && !stopWriter.get()) {
1012                    try {
1013                        writes.wait();
1014                    } catch (InterruptedException e) {
1015                        Thread.currentThread().interrupt();
1016                        throw new InterruptedIOException();
1017                    }
1018                }
1019            }
1020
1021            boolean longTx = false;
1022
1023            for (Map.Entry<Long, PageWrite> entry : updates) {
1024                Long key = entry.getKey();
1025                PageWrite value = entry.getValue();
1026                PageWrite write = writes.get(key);
1027                if (write == null) {
1028                    writes.put(key, value);
1029                } else {
1030                    if (value.currentLocation != -1) {
1031                        write.setCurrentLocation(value.page, value.currentLocation, value.length);
1032                        write.tmpFile = value.tmpFile;
1033                        longTx = true;
1034                    } else {
1035                        write.setCurrent(value.page, value.current);
1036                    }
1037                }
1038            }
1039
1040            // Once we start approaching capacity, notify the writer to start writing
1041            // sync immediately for long txs
1042            if (longTx || canStartWriteBatch()) {
1043
1044                if (enabledWriteThread) {
1045                    writes.notify();
1046                } else {
1047                    writeBatch();
1048                }
1049            }
1050        }
1051    }
1052
1053    private boolean canStartWriteBatch() {
1054        int capacityUsed = ((writes.size() * 100) / writeBatchSize);
1055        if (enabledWriteThread) {
1056            // The constant 10 here controls how soon write batches start going to disk..
1057            // would be nice to figure out how to auto tune that value.  Make to small and
1058            // we reduce through put because we are locking the write mutex too often doing writes
1059            return capacityUsed >= 10 || checkpointLatch != null;
1060        } else {
1061            return capacityUsed >= 80 || checkpointLatch != null;
1062        }
1063    }
1064
1065    ///////////////////////////////////////////////////////////////////
1066    // Cache Related operations
1067    ///////////////////////////////////////////////////////////////////
1068    @SuppressWarnings("unchecked")
1069    <T> Page<T> getFromCache(long pageId) {
1070        synchronized (writes) {
1071            PageWrite pageWrite = writes.get(pageId);
1072            if (pageWrite != null) {
1073                return pageWrite.page;
1074            }
1075        }
1076
1077        Page<T> result = null;
1078        if (enablePageCaching) {
1079            result = pageCache.get(pageId);
1080        }
1081        return result;
1082    }
1083
1084    void addToCache(Page page) {
1085        if (enablePageCaching) {
1086            pageCache.put(page.getPageId(), page);
1087        }
1088    }
1089
1090    void removeFromCache(long pageId) {
1091        if (enablePageCaching) {
1092            pageCache.remove(pageId);
1093        }
1094    }
1095
1096    ///////////////////////////////////////////////////////////////////
1097    // Internal Double write implementation follows...
1098    ///////////////////////////////////////////////////////////////////
1099
1100    private void pollWrites() {
1101        try {
1102            while (!stopWriter.get()) {
1103                // Wait for a notification...
1104                synchronized (writes) {
1105                    writes.notifyAll();
1106
1107                    // If there is not enough to write, wait for a notification...
1108                    while (writes.isEmpty() && checkpointLatch == null && !stopWriter.get()) {
1109                        writes.wait(100);
1110                    }
1111
1112                    if (writes.isEmpty()) {
1113                        releaseCheckpointWaiter();
1114                    }
1115                }
1116                writeBatch();
1117            }
1118        } catch (Throwable e) {
1119            LOG.info("An exception was raised while performing poll writes", e);
1120        } finally {
1121            releaseCheckpointWaiter();
1122        }
1123    }
1124
1125    private void writeBatch() throws IOException {
1126
1127        CountDownLatch checkpointLatch;
1128        ArrayList<PageWrite> batch;
1129        synchronized (writes) {
1130            // If there is not enough to write, wait for a notification...
1131
1132            batch = new ArrayList<PageWrite>(writes.size());
1133            // build a write batch from the current write cache.
1134            for (PageWrite write : writes.values()) {
1135                batch.add(write);
1136                // Move the current write to the diskBound write, this lets folks update the
1137                // page again without blocking for this write.
1138                write.begin();
1139                if (write.diskBound == null && write.diskBoundLocation == -1) {
1140                    batch.remove(write);
1141                }
1142            }
1143
1144            // Grab on to the existing checkpoint latch cause once we do this write we can
1145            // release the folks that were waiting for those writes to hit disk.
1146            checkpointLatch = this.checkpointLatch;
1147            this.checkpointLatch = null;
1148        }
1149
1150        try {
1151
1152            // First land the writes in the recovery file
1153            if (enableRecoveryFile) {
1154                Checksum checksum = new Adler32();
1155
1156                recoveryFile.seek(RECOVERY_FILE_HEADER_SIZE);
1157
1158                for (PageWrite w : batch) {
1159                    try {
1160                        checksum.update(w.getDiskBound(tmpFilesForRemoval), 0, pageSize);
1161                    } catch (Throwable t) {
1162                        throw IOExceptionSupport.create("Cannot create recovery file. Reason: " + t, t);
1163                    }
1164                    recoveryFile.writeLong(w.page.getPageId());
1165                    recoveryFile.write(w.getDiskBound(tmpFilesForRemoval), 0, pageSize);
1166                }
1167
1168                // Can we shrink the recovery buffer??
1169                if (recoveryPageCount > recoveryFileMaxPageCount) {
1170                    int t = Math.max(recoveryFileMinPageCount, batch.size());
1171                    recoveryFile.setLength(recoveryFileSizeForPages(t));
1172                }
1173
1174                // Record the page writes in the recovery buffer.
1175                recoveryFile.seek(0);
1176                // Store the next tx id...
1177                recoveryFile.writeLong(nextTxid.get());
1178                // Store the checksum for thw write batch so that on recovery we
1179                // know if we have a consistent
1180                // write batch on disk.
1181                recoveryFile.writeLong(checksum.getValue());
1182                // Write the # of pages that will follow
1183                recoveryFile.writeInt(batch.size());
1184
1185                if (enableDiskSyncs) {
1186                    recoveryFile.sync();
1187                }
1188            }
1189
1190            for (PageWrite w : batch) {
1191                writeFile.seek(toOffset(w.page.getPageId()));
1192                writeFile.write(w.getDiskBound(tmpFilesForRemoval), 0, pageSize);
1193                w.done();
1194            }
1195
1196            if (enableDiskSyncs) {
1197                writeFile.sync();
1198            }
1199
1200        } catch (IOException ioError) {
1201            LOG.info("Unexpected io error on pagefile write of " + batch.size() + " pages.", ioError);
1202            // any subsequent write needs to be prefaced with a considered call to redoRecoveryUpdates
1203            // to ensure disk image is self consistent
1204            loaded.set(false);
1205            throw  ioError;
1206        } finally {
1207            synchronized (writes) {
1208                for (PageWrite w : batch) {
1209                    // If there are no more pending writes, then remove it from
1210                    // the write cache.
1211                    if (w.isDone()) {
1212                        writes.remove(w.page.getPageId());
1213                        if (w.tmpFile != null && tmpFilesForRemoval.containsKey(w.tmpFile)) {
1214                            tmpFilesForRemoval.get(w.tmpFile).close();
1215                            if (!w.tmpFile.delete()) {
1216                                throw new IOException("Can't delete temporary KahaDB transaction file:" + w.tmpFile);
1217                            }
1218                            tmpFilesForRemoval.remove(w.tmpFile);
1219                        }
1220                    }
1221                }
1222            }
1223
1224            if (checkpointLatch != null) {
1225                checkpointLatch.countDown();
1226            }
1227        }
1228    }
1229
1230    public void removeTmpFile(File file, RandomAccessFile randomAccessFile) throws IOException {
1231        if (!tmpFilesForRemoval.containsKey(file)) {
1232            tmpFilesForRemoval.put(file, randomAccessFile);
1233        } else {
1234            randomAccessFile.close();
1235        }
1236    }
1237
1238    private long recoveryFileSizeForPages(int pageCount) {
1239        return RECOVERY_FILE_HEADER_SIZE + ((pageSize + 8L) * pageCount);
1240    }
1241
1242    private void releaseCheckpointWaiter() {
1243        if (checkpointLatch != null) {
1244            checkpointLatch.countDown();
1245            checkpointLatch = null;
1246        }
1247    }
1248
1249    /**
1250     * Inspects the recovery buffer and re-applies any
1251     * partially applied page writes.
1252     *
1253     * @return the next transaction id that can be used.
1254     */
1255    private long redoRecoveryUpdates() throws IOException {
1256        if (!enableRecoveryFile) {
1257            return 0;
1258        }
1259        recoveryPageCount = 0;
1260
1261        // Are we initializing the recovery file?
1262        if (recoveryFile.length() == 0) {
1263            // Write an empty header..
1264            recoveryFile.write(new byte[RECOVERY_FILE_HEADER_SIZE]);
1265            // Preallocate the minium size for better performance.
1266            recoveryFile.setLength(recoveryFileSizeForPages(recoveryFileMinPageCount));
1267            return 0;
1268        }
1269
1270        // How many recovery pages do we have in the recovery buffer?
1271        recoveryFile.seek(0);
1272        long nextTxId = recoveryFile.readLong();
1273        long expectedChecksum = recoveryFile.readLong();
1274        int pageCounter = recoveryFile.readInt();
1275
1276        recoveryFile.seek(RECOVERY_FILE_HEADER_SIZE);
1277        Checksum checksum = new Adler32();
1278        LinkedHashMap<Long, byte[]> batch = new LinkedHashMap<Long, byte[]>();
1279        try {
1280            for (int i = 0; i < pageCounter; i++) {
1281                long offset = recoveryFile.readLong();
1282                byte[] data = new byte[pageSize];
1283                if (recoveryFile.read(data, 0, pageSize) != pageSize) {
1284                    // Invalid recovery record, Could not fully read the data". Probably due to a partial write to the recovery buffer
1285                    return nextTxId;
1286                }
1287                checksum.update(data, 0, pageSize);
1288                batch.put(offset, data);
1289            }
1290        } catch (Exception e) {
1291            // If an error occurred it was cause the redo buffer was not full written out correctly.. so don't redo it.
1292            // as the pages should still be consistent.
1293            LOG.debug("Redo buffer was not fully intact: ", e);
1294            return nextTxId;
1295        }
1296
1297        recoveryPageCount = pageCounter;
1298
1299        // If the checksum is not valid then the recovery buffer was partially written to disk.
1300        if (checksum.getValue() != expectedChecksum) {
1301            return nextTxId;
1302        }
1303
1304        // Re-apply all the writes in the recovery buffer.
1305        for (Map.Entry<Long, byte[]> e : batch.entrySet()) {
1306            writeFile.seek(toOffset(e.getKey()));
1307            writeFile.write(e.getValue());
1308        }
1309
1310        // And sync it to disk
1311        writeFile.sync();
1312        return nextTxId;
1313    }
1314
1315    private void startWriter() {
1316        synchronized (writes) {
1317            if (enabledWriteThread) {
1318                stopWriter.set(false);
1319                writerThread = new Thread("KahaDB Page Writer") {
1320                    @Override
1321                    public void run() {
1322                        pollWrites();
1323                    }
1324                };
1325                writerThread.setPriority(Thread.MAX_PRIORITY);
1326                writerThread.setDaemon(true);
1327                writerThread.start();
1328            }
1329        }
1330    }
1331
1332    private void stopWriter() throws InterruptedException {
1333        if (enabledWriteThread) {
1334            stopWriter.set(true);
1335            writerThread.join();
1336        }
1337    }
1338
1339    public File getFile() {
1340        return getMainPageFile();
1341    }
1342
1343    public File getDirectory() {
1344        return directory;
1345    }
1346}