Author: brandonwilliams Date: Wed Mar 16 18:37:19 2011 New Revision: 1082254
URL: http://svn.apache.org/viewvc?rev=1082254&view=rev Log: Forgotten files from 2272 Added: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingCache.java (with props) cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java (with props) cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingRowCache.java (with props) Added: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingCache.java URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingCache.java?rev=1082254&view=auto ============================================================================== --- cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingCache.java (added) +++ cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingCache.java Wed Mar 16 18:37:19 2011 @@ -0,0 +1,263 @@ +/** + * 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.cassandra.cache; + +import java.io.*; +import java.nio.ByteBuffer; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.CompactionManager; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.io.ICompactionInfo; +import org.apache.cassandra.io.util.BufferedRandomAccessFile; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.WrappedRunnable; + +public abstract class AutoSavingCache<K, V> extends JMXInstrumentedCache<K, V> +{ + private static final Logger logger = LoggerFactory.getLogger(AutoSavingCache.class); + + protected final String cfName; + protected final String tableName; + protected volatile ScheduledFuture<?> saveTask; + protected final ColumnFamilyStore.CacheType cacheType; + + public AutoSavingCache(String tableName, String cfName, ColumnFamilyStore.CacheType cacheType, int capacity) + { + super(tableName, cfName + cacheType, capacity); + this.tableName = tableName; + this.cfName = cfName; + this.cacheType = cacheType; + } + + public abstract ByteBuffer translateKey(K key); + public abstract double getConfiguredCacheSize(CFMetaData cfm); + + public int getAdjustedCacheSize(long expectedKeys) + { + CFMetaData cfm = DatabaseDescriptor.getCFMetaData(tableName, cfName); + return (int)Math.min(FBUtilities.absoluteFromFraction(getConfiguredCacheSize(cfm), expectedKeys), Integer.MAX_VALUE); + } + + public File getCachePath() + { + return DatabaseDescriptor.getSerializedCachePath(tableName, cfName, cacheType); + } + + public Writer getWriter() + { + return new Writer(); + } + + public void scheduleSaving(int savePeriodInSeconds) + { + if (saveTask != null) + { + saveTask.cancel(false); // Do not interrupt an in-progress save + saveTask = null; + } + if (savePeriodInSeconds > 0) + { + Runnable runnable = new WrappedRunnable() + { + public void runMayThrow() + { + submitWrite(); + } + }; + saveTask = StorageService.scheduledTasks.scheduleWithFixedDelay(runnable, + savePeriodInSeconds, + savePeriodInSeconds, + TimeUnit.SECONDS); + } + } + + public Future<?> submitWrite() + { + return CompactionManager.instance.submitCacheWrite(getWriter()); + } + + public Set<DecoratedKey> readSaved() + { + File path = getCachePath(); + Set<DecoratedKey> keys = new TreeSet<DecoratedKey>(); + if (path.exists()) + { + DataInputStream in = null; + try + { + long start = System.currentTimeMillis(); + + logger.info(String.format("reading saved cache %s", path)); + in = new DataInputStream(new BufferedInputStream(new FileInputStream(path))); + while (in.available() > 0) + { + int size = in.readInt(); + byte[] bytes = new byte[size]; + in.readFully(bytes); + ByteBuffer buffer = ByteBuffer.wrap(bytes); + DecoratedKey key; + try + { + key = StorageService.getPartitioner().decorateKey(buffer); + } + catch (Exception e) + { + logger.info(String.format("unable to read entry #%s from saved cache %s; skipping remaining entries", + keys.size(), path.getAbsolutePath()), e); + break; + } + keys.add(key); + } + if (logger.isDebugEnabled()) + logger.debug(String.format("completed reading (%d ms; %d keys) saved cache %s", + System.currentTimeMillis() - start, keys.size(), path)); + } + catch (IOException ioe) + { + logger.warn(String.format("error reading saved cache %s", path.getAbsolutePath()), ioe); + } + finally + { + FileUtils.closeQuietly(in); + } + } + return keys; + } + + /** + * Resizes the cache based on a key estimate. + * Caller is in charge of synchronizing this correctly if needed + */ + public void updateCacheSize(long keys) + { + if (!isCapacitySetManually()) + { + int cacheSize = getAdjustedCacheSize(keys); + if (cacheSize != getCapacity()) + { + // update cache size for the new volume + if (logger.isDebugEnabled()) + logger.debug(cacheType + " capacity for " + cfName + " is " + cacheSize); + updateCapacity(cacheSize); + } + } + } + + public void reduceCacheSize() + { + if (getCapacity() > 0) + { + int newCapacity = (int) (DatabaseDescriptor.getReduceCacheCapacityTo() * getSize()); + logger.warn(String.format("Reducing %s %s capacity from %d to %s to reduce memory pressure", + cfName, cacheType, getCapacity(), newCapacity)); + setCapacity(newCapacity); + } + } + + public class Writer implements ICompactionInfo + { + private final Set<K> keys; + private final long estimatedTotalBytes; + private long bytesWritten; + + private Writer() + { + keys = getKeySet(); + + long bytes = 0; + for (K key : keys) + bytes += translateKey(key).remaining(); + + // an approximation -- the keyset can change while saving + estimatedTotalBytes = bytes; + } + + public void saveCache() throws IOException + { + long start = System.currentTimeMillis(); + File path = getCachePath(); + + if (keys.size() == 0 || estimatedTotalBytes == 0) + { + logger.debug("Deleting {} (cache is empty)"); + path.delete(); + return; + } + + logger.debug("Saving {}", path); + File tmpFile = File.createTempFile(path.getName(), null, path.getParentFile()); + + BufferedRandomAccessFile out = new BufferedRandomAccessFile(tmpFile, "rw", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, true); + try + { + for (K key : keys) + { + ByteBuffer bytes = translateKey(key); + ByteBufferUtil.writeWithLength(bytes, out); + bytesWritten += bytes.remaining(); + } + } + finally + { + out.close(); + } + + path.delete(); // ignore error if it didn't exist + if (!tmpFile.renameTo(path)) + throw new IOException("Unable to rename " + tmpFile + " to " + path); + logger.info(String.format("Saved %s (%d items) in %d ms", + path.getName(), keys.size(), (System.currentTimeMillis() - start))); + } + + public long getTotalBytes() + { + // keyset can change in size, thus totalBytes can too + return Math.max(estimatedTotalBytes, getBytesComplete()); + } + + public long getBytesComplete() + { + return bytesWritten; + } + + public String getTaskType() + { + return "Save " + getCachePath().getName(); + } + + public String getColumnFamily() + { + return cfName; + } + } +} Propchange: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingCache.java ------------------------------------------------------------------------------ svn:eol-style = native Added: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java?rev=1082254&view=auto ============================================================================== --- cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java (added) +++ cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java Wed Mar 16 18:37:19 2011 @@ -0,0 +1,29 @@ +package org.apache.cassandra.cache; + +import java.nio.ByteBuffer; + +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.utils.Pair; + +public class AutoSavingKeyCache<K extends Pair<Descriptor, DecoratedKey>, V> extends AutoSavingCache<K, V> +{ + public AutoSavingKeyCache(String tableName, String cfName, int capacity) + { + super(tableName, cfName, ColumnFamilyStore.CacheType.KEY_CACHE_TYPE, capacity); + } + + @Override + public double getConfiguredCacheSize(CFMetaData cfm) + { + return cfm == null ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cfm.getKeyCacheSize(); + } + + @Override + public ByteBuffer translateKey(K key) + { + return key.right.key; + } +} Propchange: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java ------------------------------------------------------------------------------ svn:eol-style = native Added: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingRowCache.java URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingRowCache.java?rev=1082254&view=auto ============================================================================== --- cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingRowCache.java (added) +++ cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingRowCache.java Wed Mar 16 18:37:19 2011 @@ -0,0 +1,27 @@ +package org.apache.cassandra.cache; + +import java.nio.ByteBuffer; + +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DecoratedKey; + +public class AutoSavingRowCache<K extends DecoratedKey, V> extends AutoSavingCache<K, V> +{ + public AutoSavingRowCache(String tableName, String cfName, int capacity) + { + super(tableName, cfName, ColumnFamilyStore.CacheType.ROW_CACHE_TYPE, capacity); + } + + @Override + public double getConfiguredCacheSize(CFMetaData cfm) + { + return cfm == null ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cfm.getRowCacheSize(); + } + + @Override + public ByteBuffer translateKey(K key) + { + return key.key; + } +} Propchange: cassandra/trunk/src/java/org/apache/cassandra/cache/AutoSavingRowCache.java ------------------------------------------------------------------------------ svn:eol-style = native
