| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.activemq.store.kahadb; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.Date; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import org.apache.activemq.broker.LockableServiceSupport; |
| import org.apache.activemq.broker.Locker; |
| import org.apache.activemq.store.SharedFileLocker; |
| import org.apache.activemq.store.kahadb.data.KahaEntryType; |
| import org.apache.activemq.store.kahadb.data.KahaTraceCommand; |
| import org.apache.activemq.store.kahadb.disk.journal.Journal; |
| import org.apache.activemq.store.kahadb.disk.journal.Location; |
| import org.apache.activemq.store.kahadb.disk.page.PageFile; |
| import org.apache.activemq.store.kahadb.disk.page.Transaction; |
| import org.apache.activemq.util.ByteSequence; |
| import org.apache.activemq.util.DataByteArrayInputStream; |
| import org.apache.activemq.util.DataByteArrayOutputStream; |
| import org.apache.activemq.util.IOHelper; |
| import org.apache.activemq.util.ServiceStopper; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public abstract class AbstractKahaDBStore extends LockableServiceSupport { |
| |
| static final Logger LOG = LoggerFactory.getLogger(AbstractKahaDBStore.class); |
| |
| public static final String PROPERTY_LOG_SLOW_ACCESS_TIME = "org.apache.activemq.store.kahadb.LOG_SLOW_ACCESS_TIME"; |
| public static final int LOG_SLOW_ACCESS_TIME = Integer.getInteger(PROPERTY_LOG_SLOW_ACCESS_TIME, 0); |
| |
| protected File directory; |
| protected PageFile pageFile; |
| protected Journal journal; |
| protected AtomicLong journalSize = new AtomicLong(0); |
| protected boolean failIfDatabaseIsLocked; |
| protected long checkpointInterval = 5*1000; |
| protected long cleanupInterval = 30*1000; |
| private boolean cleanupOnStop = true; |
| protected boolean checkForCorruptJournalFiles = false; |
| protected boolean checksumJournalFiles = true; |
| protected boolean forceRecoverIndex = false; |
| protected int journalMaxFileLength = Journal.DEFAULT_MAX_FILE_LENGTH; |
| protected int journalMaxWriteBatchSize = Journal.DEFAULT_MAX_WRITE_BATCH_SIZE; |
| protected boolean archiveCorruptedIndex = false; |
| protected boolean enableIndexWriteAsync = false; |
| protected boolean enableJournalDiskSyncs = false; |
| protected boolean deleteAllJobs = false; |
| protected int indexWriteBatchSize = PageFile.DEFAULT_WRITE_BATCH_SIZE; |
| protected boolean useIndexLFRUEviction = false; |
| protected float indexLFUEvictionFactor = 0.2f; |
| protected boolean ignoreMissingJournalfiles = false; |
| protected int indexCacheSize = 1000; |
| protected boolean enableIndexDiskSyncs = true; |
| protected boolean enableIndexRecoveryFile = true; |
| protected boolean enableIndexPageCaching = true; |
| protected boolean archiveDataLogs; |
| protected boolean purgeStoreOnStartup; |
| protected File directoryArchive; |
| |
| protected AtomicBoolean opened = new AtomicBoolean(); |
| protected Thread checkpointThread; |
| protected final Object checkpointThreadLock = new Object(); |
| protected ReentrantReadWriteLock checkpointLock = new ReentrantReadWriteLock(); |
| protected ReentrantReadWriteLock indexLock = new ReentrantReadWriteLock(); |
| |
| /** |
| * @return the name to give this store's PageFile instance. |
| */ |
| protected abstract String getPageFileName(); |
| |
| /** |
| * @return the location of the data directory if no set by configuration. |
| */ |
| protected abstract File getDefaultDataDirectory(); |
| |
| /** |
| * Loads the store from disk. |
| * |
| * Based on configuration this method can either load an existing store or it can purge |
| * an existing store and start in a clean state. |
| * |
| * @throws IOException if an error occurs during the load. |
| */ |
| public abstract void load() throws IOException; |
| |
| /** |
| * Unload the state of the Store to disk and shuts down all resources assigned to this |
| * KahaDB store implementation. |
| * |
| * @throws IOException if an error occurs during the store unload. |
| */ |
| public abstract void unload() throws IOException; |
| |
| @Override |
| protected void doStart() throws Exception { |
| this.indexLock.writeLock().lock(); |
| if (getDirectory() == null) { |
| setDirectory(getDefaultDataDirectory()); |
| } |
| IOHelper.mkdirs(getDirectory()); |
| try { |
| if (isPurgeStoreOnStartup()) { |
| getJournal().start(); |
| getJournal().delete(); |
| getJournal().close(); |
| journal = null; |
| getPageFile().delete(); |
| LOG.info("{} Persistence store purged.", this); |
| setPurgeStoreOnStartup(false); |
| } |
| |
| load(); |
| store(new KahaTraceCommand().setMessage("LOADED " + new Date())); |
| } finally { |
| this.indexLock.writeLock().unlock(); |
| } |
| } |
| |
| @Override |
| protected void doStop(ServiceStopper stopper) throws Exception { |
| unload(); |
| } |
| |
| public PageFile getPageFile() { |
| if (pageFile == null) { |
| pageFile = createPageFile(); |
| } |
| return pageFile; |
| } |
| |
| public Journal getJournal() throws IOException { |
| if (journal == null) { |
| journal = createJournal(); |
| } |
| return journal; |
| } |
| |
| public File getDirectory() { |
| return directory; |
| } |
| |
| public void setDirectory(File directory) { |
| this.directory = directory; |
| } |
| |
| public boolean isArchiveCorruptedIndex() { |
| return archiveCorruptedIndex; |
| } |
| |
| public void setArchiveCorruptedIndex(boolean archiveCorruptedIndex) { |
| this.archiveCorruptedIndex = archiveCorruptedIndex; |
| } |
| |
| public boolean isFailIfDatabaseIsLocked() { |
| return failIfDatabaseIsLocked; |
| } |
| |
| public void setFailIfDatabaseIsLocked(boolean failIfDatabaseIsLocked) { |
| this.failIfDatabaseIsLocked = failIfDatabaseIsLocked; |
| } |
| |
| public boolean isCheckForCorruptJournalFiles() { |
| return checkForCorruptJournalFiles; |
| } |
| |
| public void setCheckForCorruptJournalFiles(boolean checkForCorruptJournalFiles) { |
| this.checkForCorruptJournalFiles = checkForCorruptJournalFiles; |
| } |
| |
| public long getCheckpointInterval() { |
| return checkpointInterval; |
| } |
| |
| public void setCheckpointInterval(long checkpointInterval) { |
| this.checkpointInterval = checkpointInterval; |
| } |
| |
| public long getCleanupInterval() { |
| return cleanupInterval; |
| } |
| |
| public void setCleanupInterval(long cleanupInterval) { |
| this.cleanupInterval = cleanupInterval; |
| } |
| |
| public void setCleanupOnStop(boolean cleanupOnStop) { |
| this.cleanupOnStop = cleanupOnStop; |
| } |
| |
| public boolean getCleanupOnStop() { |
| return this.cleanupOnStop; |
| } |
| |
| public boolean isChecksumJournalFiles() { |
| return checksumJournalFiles; |
| } |
| |
| public void setChecksumJournalFiles(boolean checksumJournalFiles) { |
| this.checksumJournalFiles = checksumJournalFiles; |
| } |
| |
| public boolean isForceRecoverIndex() { |
| return forceRecoverIndex; |
| } |
| |
| public void setForceRecoverIndex(boolean forceRecoverIndex) { |
| this.forceRecoverIndex = forceRecoverIndex; |
| } |
| |
| public int getJournalMaxFileLength() { |
| return journalMaxFileLength; |
| } |
| |
| public void setJournalMaxFileLength(int journalMaxFileLength) { |
| this.journalMaxFileLength = journalMaxFileLength; |
| } |
| |
| public int getJournalMaxWriteBatchSize() { |
| return journalMaxWriteBatchSize; |
| } |
| |
| public void setJournalMaxWriteBatchSize(int journalMaxWriteBatchSize) { |
| this.journalMaxWriteBatchSize = journalMaxWriteBatchSize; |
| } |
| |
| public boolean isEnableIndexWriteAsync() { |
| return enableIndexWriteAsync; |
| } |
| |
| public void setEnableIndexWriteAsync(boolean enableIndexWriteAsync) { |
| this.enableIndexWriteAsync = enableIndexWriteAsync; |
| } |
| |
| public boolean isEnableJournalDiskSyncs() { |
| return enableJournalDiskSyncs; |
| } |
| |
| public void setEnableJournalDiskSyncs(boolean syncWrites) { |
| this.enableJournalDiskSyncs = syncWrites; |
| } |
| |
| public boolean isDeleteAllJobs() { |
| return deleteAllJobs; |
| } |
| |
| public void setDeleteAllJobs(boolean deleteAllJobs) { |
| this.deleteAllJobs = deleteAllJobs; |
| } |
| |
| /** |
| * @return the archiveDataLogs |
| */ |
| public boolean isArchiveDataLogs() { |
| return this.archiveDataLogs; |
| } |
| |
| /** |
| * @param archiveDataLogs the archiveDataLogs to set |
| */ |
| public void setArchiveDataLogs(boolean archiveDataLogs) { |
| this.archiveDataLogs = archiveDataLogs; |
| } |
| |
| /** |
| * @return the directoryArchive |
| */ |
| public File getDirectoryArchive() { |
| return this.directoryArchive; |
| } |
| |
| /** |
| * @param directoryArchive the directoryArchive to set |
| */ |
| public void setDirectoryArchive(File directoryArchive) { |
| this.directoryArchive = directoryArchive; |
| } |
| |
| public int getIndexCacheSize() { |
| return indexCacheSize; |
| } |
| |
| public void setIndexCacheSize(int indexCacheSize) { |
| this.indexCacheSize = indexCacheSize; |
| } |
| |
| public int getIndexWriteBatchSize() { |
| return indexWriteBatchSize; |
| } |
| |
| public void setIndexWriteBatchSize(int indexWriteBatchSize) { |
| this.indexWriteBatchSize = indexWriteBatchSize; |
| } |
| |
| public boolean isUseIndexLFRUEviction() { |
| return useIndexLFRUEviction; |
| } |
| |
| public void setUseIndexLFRUEviction(boolean useIndexLFRUEviction) { |
| this.useIndexLFRUEviction = useIndexLFRUEviction; |
| } |
| |
| public float getIndexLFUEvictionFactor() { |
| return indexLFUEvictionFactor; |
| } |
| |
| public void setIndexLFUEvictionFactor(float indexLFUEvictionFactor) { |
| this.indexLFUEvictionFactor = indexLFUEvictionFactor; |
| } |
| |
| public boolean isEnableIndexDiskSyncs() { |
| return enableIndexDiskSyncs; |
| } |
| |
| public void setEnableIndexDiskSyncs(boolean enableIndexDiskSyncs) { |
| this.enableIndexDiskSyncs = enableIndexDiskSyncs; |
| } |
| |
| public boolean isEnableIndexRecoveryFile() { |
| return enableIndexRecoveryFile; |
| } |
| |
| public void setEnableIndexRecoveryFile(boolean enableIndexRecoveryFile) { |
| this.enableIndexRecoveryFile = enableIndexRecoveryFile; |
| } |
| |
| public boolean isEnableIndexPageCaching() { |
| return enableIndexPageCaching; |
| } |
| |
| public void setEnableIndexPageCaching(boolean enableIndexPageCaching) { |
| this.enableIndexPageCaching = enableIndexPageCaching; |
| } |
| |
| public boolean isPurgeStoreOnStartup() { |
| return this.purgeStoreOnStartup; |
| } |
| |
| public void setPurgeStoreOnStartup(boolean purge) { |
| this.purgeStoreOnStartup = purge; |
| } |
| |
| public boolean isIgnoreMissingJournalfiles() { |
| return ignoreMissingJournalfiles; |
| } |
| |
| public void setIgnoreMissingJournalfiles(boolean ignoreMissingJournalfiles) { |
| this.ignoreMissingJournalfiles = ignoreMissingJournalfiles; |
| } |
| |
| public long size() { |
| if (!isStarted()) { |
| return 0; |
| } |
| try { |
| return journalSize.get() + pageFile.getDiskSize(); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| @Override |
| public Locker createDefaultLocker() throws IOException { |
| SharedFileLocker locker = new SharedFileLocker(); |
| locker.setDirectory(this.getDirectory()); |
| return locker; |
| } |
| |
| @Override |
| public void init() throws Exception { |
| } |
| |
| /** |
| * Store a command in the Journal and process to update the Store index. |
| * |
| * @param command |
| * The specific JournalCommand to store and process. |
| * |
| * @returns the Location where the data was written in the Journal. |
| * |
| * @throws IOException if an error occurs storing or processing the command. |
| */ |
| public Location store(JournalCommand<?> command) throws IOException { |
| return store(command, isEnableIndexDiskSyncs(), null, null, null); |
| } |
| |
| /** |
| * Store a command in the Journal and process to update the Store index. |
| * |
| * @param command |
| * The specific JournalCommand to store and process. |
| * @param sync |
| * Should the store operation be done synchronously. (ignored if completion passed). |
| * |
| * @returns the Location where the data was written in the Journal. |
| * |
| * @throws IOException if an error occurs storing or processing the command. |
| */ |
| public Location store(JournalCommand<?> command, boolean sync) throws IOException { |
| return store(command, sync, null, null, null); |
| } |
| |
| /** |
| * Store a command in the Journal and process to update the Store index. |
| * |
| * @param command |
| * The specific JournalCommand to store and process. |
| * @param onJournalStoreComplete |
| * The Runnable to call when the Journal write operation completes. |
| * |
| * @returns the Location where the data was written in the Journal. |
| * |
| * @throws IOException if an error occurs storing or processing the command. |
| */ |
| public Location store(JournalCommand<?> command, Runnable onJournalStoreComplete) throws IOException { |
| return store(command, isEnableIndexDiskSyncs(), null, null, onJournalStoreComplete); |
| } |
| |
| /** |
| * Store a command in the Journal and process to update the Store index. |
| * |
| * @param command |
| * The specific JournalCommand to store and process. |
| * @param sync |
| * Should the store operation be done synchronously. (ignored if completion passed). |
| * @param before |
| * The Runnable instance to execute before performing the store and process operation. |
| * @param after |
| * The Runnable instance to execute after performing the store and process operation. |
| * |
| * @returns the Location where the data was written in the Journal. |
| * |
| * @throws IOException if an error occurs storing or processing the command. |
| */ |
| public Location store(JournalCommand<?> command, boolean sync, Runnable before, Runnable after) throws IOException { |
| return store(command, sync, before, after, null); |
| } |
| |
| /** |
| * All updated are are funneled through this method. The updates are converted to a |
| * JournalMessage which is logged to the journal and then the data from the JournalMessage |
| * is used to update the index just like it would be done during a recovery process. |
| * |
| * @param command |
| * The specific JournalCommand to store and process. |
| * @param sync |
| * Should the store operation be done synchronously. (ignored if completion passed). |
| * @param before |
| * The Runnable instance to execute before performing the store and process operation. |
| * @param after |
| * The Runnable instance to execute after performing the store and process operation. |
| * @param onJournalStoreComplete |
| * Callback to be run when the journal write operation is complete. |
| * |
| * @returns the Location where the data was written in the Journal. |
| * |
| * @throws IOException if an error occurs storing or processing the command. |
| */ |
| public Location store(JournalCommand<?> command, boolean sync, Runnable before, Runnable after, Runnable onJournalStoreComplete) throws IOException { |
| try { |
| |
| if (before != null) { |
| before.run(); |
| } |
| |
| ByteSequence sequence = toByteSequence(command); |
| Location location; |
| checkpointLock.readLock().lock(); |
| try { |
| |
| long start = System.currentTimeMillis(); |
| location = onJournalStoreComplete == null ? journal.write(sequence, sync) : |
| journal.write(sequence, onJournalStoreComplete); |
| long start2 = System.currentTimeMillis(); |
| |
| process(command, location); |
| |
| long end = System.currentTimeMillis(); |
| if (LOG_SLOW_ACCESS_TIME > 0 && end - start > LOG_SLOW_ACCESS_TIME) { |
| LOG.info("Slow KahaDB access: Journal append took: {} ms, Index update took {} ms", |
| (start2-start), (end-start2)); |
| } |
| } finally { |
| checkpointLock.readLock().unlock(); |
| } |
| |
| if (after != null) { |
| after.run(); |
| } |
| |
| if (checkpointThread != null && !checkpointThread.isAlive()) { |
| startCheckpoint(); |
| } |
| return location; |
| } catch (IOException ioe) { |
| LOG.error("KahaDB failed to store to Journal", ioe); |
| if (brokerService != null) { |
| brokerService.handleIOException(ioe); |
| } |
| throw ioe; |
| } |
| } |
| |
| /** |
| * Loads a previously stored JournalMessage |
| * |
| * @param location |
| * The location of the journal command to read. |
| * |
| * @return a new un-marshaled JournalCommand instance. |
| * |
| * @throws IOException if an error occurs reading the stored command. |
| */ |
| protected JournalCommand<?> load(Location location) throws IOException { |
| ByteSequence data = journal.read(location); |
| DataByteArrayInputStream is = new DataByteArrayInputStream(data); |
| byte readByte = is.readByte(); |
| KahaEntryType type = KahaEntryType.valueOf(readByte); |
| if (type == null) { |
| try { |
| is.close(); |
| } catch (IOException e) { |
| } |
| throw new IOException("Could not load journal record. Invalid location: " + location); |
| } |
| JournalCommand<?> message = (JournalCommand<?>)type.createMessage(); |
| message.mergeFramed(is); |
| return message; |
| } |
| |
| /** |
| * Process a stored or recovered JournalCommand instance and update the DB Index with the |
| * state changes that this command produces. This can be called either as a new DB operation |
| * or as a replay during recovery operations. |
| * |
| * @param command |
| * The JournalCommand to process. |
| * @param location |
| * The location in the Journal where the command was written or read from. |
| */ |
| protected abstract void process(JournalCommand<?> command, Location location) throws IOException; |
| |
| /** |
| * Perform a checkpoint operation with optional cleanup. |
| * |
| * Called by the checkpoint background thread periodically to initiate a checkpoint operation |
| * and if the cleanup flag is set a cleanup sweep should be done to allow for release of no |
| * longer needed journal log files etc. |
| * |
| * @param cleanup |
| * Should the method do a simple checkpoint or also perform a journal cleanup. |
| * |
| * @throws IOException if an error occurs during the checkpoint operation. |
| */ |
| protected void checkpointUpdate(final boolean cleanup) throws IOException { |
| checkpointLock.writeLock().lock(); |
| try { |
| this.indexLock.writeLock().lock(); |
| try { |
| pageFile.tx().execute(new Transaction.Closure<IOException>() { |
| @Override |
| public void execute(Transaction tx) throws IOException { |
| checkpointUpdate(tx, cleanup); |
| } |
| }); |
| } finally { |
| this.indexLock.writeLock().unlock(); |
| } |
| |
| } finally { |
| checkpointLock.writeLock().unlock(); |
| } |
| } |
| |
| /** |
| * Perform the checkpoint update operation. If the cleanup flag is true then the |
| * operation should also purge any unused Journal log files. |
| * |
| * This method must always be called with the checkpoint and index write locks held. |
| * |
| * @param tx |
| * The TX under which to perform the checkpoint update. |
| * @param cleanup |
| * Should the checkpoint also do unused Journal file cleanup. |
| * |
| * @throws IOException if an error occurs while performing the checkpoint. |
| */ |
| protected abstract void checkpointUpdate(Transaction tx, boolean cleanup) throws IOException; |
| |
| /** |
| * Creates a new ByteSequence that represents the marshaled form of the given Journal Command. |
| * |
| * @param data |
| * The Journal Command that should be marshaled to bytes for writing. |
| * |
| * @return the byte representation of the given journal command. |
| * |
| * @throws IOException if an error occurs while serializing the command. |
| */ |
| protected ByteSequence toByteSequence(JournalCommand<?> data) throws IOException { |
| int size = data.serializedSizeFramed(); |
| DataByteArrayOutputStream os = new DataByteArrayOutputStream(size + 1); |
| os.writeByte(data.type().getNumber()); |
| data.writeFramed(os); |
| return os.toByteSequence(); |
| } |
| |
| /** |
| * Create the PageFile instance and configure it using the configuration options |
| * currently set. |
| * |
| * @return the newly created and configured PageFile instance. |
| */ |
| protected PageFile createPageFile() { |
| PageFile index = new PageFile(getDirectory(), getPageFileName()); |
| index.setEnableWriteThread(isEnableIndexWriteAsync()); |
| index.setWriteBatchSize(getIndexWriteBatchSize()); |
| index.setPageCacheSize(getIndexCacheSize()); |
| index.setUseLFRUEviction(isUseIndexLFRUEviction()); |
| index.setLFUEvictionFactor(getIndexLFUEvictionFactor()); |
| index.setEnableDiskSyncs(isEnableIndexDiskSyncs()); |
| index.setEnableRecoveryFile(isEnableIndexRecoveryFile()); |
| index.setEnablePageCaching(isEnableIndexPageCaching()); |
| return index; |
| } |
| |
| /** |
| * Create a new Journal instance and configure it using the currently set configuration |
| * options. If an archive directory is configured than this method will attempt to create |
| * that directory if it does not already exist. |
| * |
| * @return the newly created an configured Journal instance. |
| * |
| * @throws IOException if an error occurs while creating the Journal object. |
| */ |
| protected Journal createJournal() throws IOException { |
| Journal manager = new Journal(); |
| manager.setDirectory(getDirectory()); |
| manager.setMaxFileLength(getJournalMaxFileLength()); |
| manager.setCheckForCorruptionOnStartup(isCheckForCorruptJournalFiles()); |
| manager.setChecksum(isChecksumJournalFiles() || isCheckForCorruptJournalFiles()); |
| manager.setWriteBatchSize(getJournalMaxWriteBatchSize()); |
| manager.setArchiveDataLogs(isArchiveDataLogs()); |
| manager.setSizeAccumulator(journalSize); |
| manager.setEnableAsyncDiskSync(isEnableJournalDiskSyncs()); |
| if (getDirectoryArchive() != null) { |
| IOHelper.mkdirs(getDirectoryArchive()); |
| manager.setDirectoryArchive(getDirectoryArchive()); |
| } |
| return manager; |
| } |
| |
| /** |
| * Starts the checkpoint Thread instance if not already running and not disabled |
| * by configuration. |
| */ |
| protected void startCheckpoint() { |
| if (checkpointInterval == 0 && cleanupInterval == 0) { |
| LOG.info("periodic checkpoint/cleanup disabled, will occur on clean " + (getCleanupOnStop() ? "shutdown/" : "") + "restart"); |
| return; |
| } |
| synchronized (checkpointThreadLock) { |
| boolean start = false; |
| if (checkpointThread == null) { |
| start = true; |
| } else if (!checkpointThread.isAlive()) { |
| start = true; |
| LOG.info("KahaDB: Recovering checkpoint thread after death"); |
| } |
| if (start) { |
| checkpointThread = new Thread("ActiveMQ Journal Checkpoint Worker") { |
| @Override |
| public void run() { |
| try { |
| long lastCleanup = System.currentTimeMillis(); |
| long lastCheckpoint = System.currentTimeMillis(); |
| // Sleep for a short time so we can periodically check |
| // to see if we need to exit this thread. |
| long sleepTime = Math.min(checkpointInterval > 0 ? checkpointInterval : cleanupInterval, 500); |
| while (opened.get()) { |
| Thread.sleep(sleepTime); |
| long now = System.currentTimeMillis(); |
| if( cleanupInterval > 0 && (now - lastCleanup >= cleanupInterval) ) { |
| checkpointCleanup(true); |
| lastCleanup = now; |
| lastCheckpoint = now; |
| } else if( checkpointInterval > 0 && (now - lastCheckpoint >= checkpointInterval )) { |
| checkpointCleanup(false); |
| lastCheckpoint = now; |
| } |
| } |
| } catch (InterruptedException e) { |
| // Looks like someone really wants us to exit this thread... |
| } catch (IOException ioe) { |
| LOG.error("Checkpoint failed", ioe); |
| brokerService.handleIOException(ioe); |
| } |
| } |
| }; |
| |
| checkpointThread.setDaemon(true); |
| checkpointThread.start(); |
| } |
| } |
| } |
| |
| /** |
| * Called from the worker thread to start a checkpoint. |
| * |
| * This method ensure that the store is in an opened state and optionaly logs information |
| * related to slow store access times. |
| * |
| * @param cleanup |
| * Should a cleanup of the journal occur during the checkpoint operation. |
| * |
| * @throws IOException if an error occurs during the checkpoint operation. |
| */ |
| protected void checkpointCleanup(final boolean cleanup) throws IOException { |
| long start; |
| this.indexLock.writeLock().lock(); |
| try { |
| start = System.currentTimeMillis(); |
| if (!opened.get()) { |
| return; |
| } |
| } finally { |
| this.indexLock.writeLock().unlock(); |
| } |
| checkpointUpdate(cleanup); |
| long end = System.currentTimeMillis(); |
| if (LOG_SLOW_ACCESS_TIME > 0 && end - start > LOG_SLOW_ACCESS_TIME) { |
| LOG.info("Slow KahaDB access: cleanup took {}", (end - start)); |
| } |
| } |
| } |
