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}