You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by li...@apache.org on 2013/06/13 20:18:21 UTC
svn commit: r1492797 [2/3] - in /hbase/branches/0.89-fb/src:
main/java/org/apache/hadoop/hbase/io/hfile/
main/java/org/apache/hadoop/hbase/io/hfile/bucket/
main/java/org/apache/hadoop/hbase/regionserver/
main/java/org/apache/hadoop/hbase/regionserver/m...
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocator.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocator.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocator.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocator.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,450 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.util.ConditionUtil;
+
+/**
+ * This class is used to allocate a block with specified size and free the block
+ * when evicting. It manages an array of buckets, each bucket is associated with
+ * a size and caches elements up to this size. For completely empty bucket, this
+ * size could be re-specified dynamically.
+ *
+ * This class is not thread safe.
+ */
+public final class BucketAllocator {
+ static final Log LOG = LogFactory.getLog(BucketAllocator.class);
+
+ final private static class Bucket {
+ private final int[] bucketSizes;
+ private final long bucketCapacity;
+ private long baseOffset;
+ private int itemAllocationSize, sizeIndex;
+ private int itemCount;
+ private int freeList[];
+ private int freeCount, usedCount;
+
+ public Bucket(int[] bucketSizes, long bucketCapacity, long offset) {
+ this.bucketSizes = bucketSizes;
+ this.bucketCapacity = bucketCapacity;
+ baseOffset = offset;
+ sizeIndex = -1;
+ }
+
+ void reconfigure(int sizeIndex) {
+ this.sizeIndex = Preconditions.checkPositionIndex(sizeIndex,
+ bucketSizes.length);
+ itemAllocationSize = bucketSizes[sizeIndex];
+ itemCount = (int) ((bucketCapacity) / (long) itemAllocationSize);
+ freeCount = itemCount;
+ usedCount = 0;
+ freeList = new int[itemCount];
+ for (int i = 0; i < freeCount; ++i)
+ freeList[i] = i;
+ }
+
+ public boolean isUninstantiated() {
+ return sizeIndex == -1;
+ }
+
+ public int sizeIndex() {
+ return sizeIndex;
+ }
+
+ public int itemAllocationSize() {
+ return itemAllocationSize;
+ }
+
+ public boolean hasFreeSpace() {
+ return freeCount > 0;
+ }
+
+ public boolean isCompletelyFree() {
+ return usedCount == 0;
+ }
+
+ public int freeCount() {
+ return freeCount;
+ }
+
+ public int usedCount() {
+ return usedCount;
+ }
+
+ public int freeBytes() {
+ return freeCount * itemAllocationSize;
+ }
+
+ public int usedBytes() {
+ return usedCount * itemAllocationSize;
+ }
+
+ public long baseOffset() {
+ return baseOffset;
+ }
+
+ /**
+ * Allocate a block in this bucket, return the offset representing the
+ * position in physical space
+ * @return the offset in the IOEngine
+ */
+ public long allocate() {
+ Preconditions.checkState(freeCount > 0, "No space to allocate!");
+ Preconditions.checkState(sizeIndex != -1);
+ ++usedCount;
+ return ConditionUtil
+ .checkPositiveOffset(baseOffset + (freeList[--freeCount] * itemAllocationSize));
+ }
+
+ private void free(long offset) {
+ Preconditions.checkState(usedCount > 0);
+ Preconditions.checkState(freeCount < itemCount,
+ "duplicate free, offset: " + offset);
+ offset = ConditionUtil.checkOffset(offset - baseOffset,
+ itemCount * itemAllocationSize);
+ Preconditions.checkState(offset % itemAllocationSize == 0);
+ int item = (int) (offset / (long) itemAllocationSize);
+ Preconditions.checkState(!freeListContains(item), "Item at " + offset +
+ " already on freelist!");
+
+ --usedCount;
+ freeList[freeCount++] = item;
+ }
+
+ private boolean freeListContains(int blockNo) {
+ for (int i = 0; i < freeCount; ++i) {
+ if (freeList[i] == blockNo) return true;
+ }
+ return false;
+ }
+ }
+
+ public final class BucketSizeInfo {
+ // Free bucket means it has space to allocate a block;
+ // Completely free bucket means it has no block.
+ private List<Bucket> bucketList, freeBuckets, completelyFreeBuckets;
+ private int sizeIndex;
+
+ BucketSizeInfo(int sizeIndex) {
+ bucketList = new ArrayList<Bucket>();
+ freeBuckets = new ArrayList<Bucket>();
+ completelyFreeBuckets = new ArrayList<Bucket>();
+ this.sizeIndex = sizeIndex;
+ }
+
+ public void instantiateBucket(Bucket b) {
+ Preconditions.checkArgument(b.isUninstantiated() || b.isCompletelyFree());
+ b.reconfigure(sizeIndex);
+ bucketList.add(b);
+ freeBuckets.add(b);
+ completelyFreeBuckets.add(b);
+ }
+
+ public int sizeIndex() {
+ return sizeIndex;
+ }
+
+ /**
+ * Find a bucket to allocate a block
+ * @return the offset in the IOEngine
+ */
+ public long allocateBlock() {
+ Bucket b = null;
+ if (freeBuckets.size() > 0) // Use up an existing one first...
+ b = freeBuckets.get(freeBuckets.size() - 1);
+ if (b == null) {
+ b = grabGlobalCompletelyFreeBucket();
+ if (b != null) instantiateBucket(b);
+ }
+ if (b == null) return -1;
+ long result = b.allocate();
+ blockAllocated(b);
+ return result;
+ }
+
+ void blockAllocated(Bucket b) {
+ if (!b.isCompletelyFree()) completelyFreeBuckets.remove(b);
+ if (!b.hasFreeSpace()) freeBuckets.remove(b);
+ }
+
+ public Bucket findAndRemoveCompletelyFreeBucket() {
+ Bucket b = null;
+ Preconditions.checkState(bucketList.size() > 0);
+ if (bucketList.size() == 1) {
+ // So we never get complete starvation of a bucket for a size
+ return null;
+ }
+
+ if (completelyFreeBuckets.size() > 0) {
+ b = completelyFreeBuckets.get(0);
+ removeBucket(b);
+ }
+ return b;
+ }
+
+ private void removeBucket(Bucket b) {
+ Preconditions.checkArgument(b.isCompletelyFree());
+ bucketList.remove(b);
+ freeBuckets.remove(b);
+ completelyFreeBuckets.remove(b);
+ }
+
+ public void freeBlock(Bucket b, long offset) {
+ Preconditions.checkArgument(bucketList.contains(b));
+ // else we shouldn't have anything to free...
+ Preconditions.checkArgument(!completelyFreeBuckets.contains(b),
+ "nothing to free!");
+ b.free(offset);
+ if (!freeBuckets.contains(b)) freeBuckets.add(b);
+ if (b.isCompletelyFree()) completelyFreeBuckets.add(b);
+ }
+
+ public IndexStatistics statistics() {
+ long free = 0, used = 0;
+ for (Bucket b : bucketList) {
+ free += b.freeCount();
+ used += b.usedCount();
+ }
+ return new IndexStatistics(free, used, bucketSizes[sizeIndex]);
+ }
+ }
+
+ private final int bucketSizes[];
+
+ /**
+ * Round up the given block size to bucket size, and get the corresponding
+ * BucketSizeInfo
+ * @param blockSize
+ * @return BucketSizeInfo
+ */
+ public BucketSizeInfo roundUpToBucketSizeInfo(int blockSize) {
+ for (int i = 0; i < bucketSizes.length; ++i)
+ if (blockSize <= bucketSizes[i])
+ return bucketSizeInfos[i];
+ return null;
+ }
+
+
+ static public final int FEWEST_ITEMS_IN_BUCKET = 4;
+ // The capacity size for each bucket
+
+ private final long bucketCapacity;
+
+ private final Bucket[] buckets;
+ private final BucketSizeInfo[] bucketSizeInfos;
+ private final long totalSize;
+
+ private long usedSize = 0;
+
+ BucketAllocator(int[] bucketSizes, long availableSpace) throws BucketAllocatorException {
+ this.bucketSizes = bucketSizes;
+ int bigItemSize = bucketSizes[bucketSizes.length - 1];
+ bucketCapacity = FEWEST_ITEMS_IN_BUCKET * bigItemSize;
+ buckets = new Bucket[(int) (availableSpace / bucketCapacity)];
+ if (buckets.length < bucketSizes.length)
+ throw new BucketAllocatorException(
+ "Bucket allocator size too small - must have room for at least "
+ + bucketSizes.length + " buckets");
+ bucketSizeInfos = new BucketSizeInfo[bucketSizes.length];
+ for (int i = 0; i < bucketSizes.length; ++i) {
+ bucketSizeInfos[i] = new BucketSizeInfo(i);
+ }
+ for (int i = 0; i < buckets.length; ++i) {
+ buckets[i] = new Bucket(bucketSizes, bucketCapacity, bucketCapacity * i);
+ bucketSizeInfos[i < bucketSizes.length ? i : bucketSizes.length - 1]
+ .instantiateBucket(buckets[i]);
+ }
+ this.totalSize = ((long) buckets.length) * bucketCapacity;
+ }
+
+ public String getInfo() {
+ StringBuilder sb = new StringBuilder(1024);
+ for (int i = 0; i < buckets.length; ++i) {
+ Bucket b = buckets[i];
+ sb.append(" Bucket ").append(i).append(": ").append(b.itemAllocationSize());
+ sb.append(" freeCount=").append(b.freeCount()).append(" used=")
+ .append(b.usedCount());
+ sb.append('\n');
+ }
+ return sb.toString();
+ }
+
+ public long getUsedSize() {
+ return this.usedSize;
+ }
+
+ public long getFreeSize() {
+ return this.totalSize - getUsedSize();
+ }
+
+ public long getTotalSize() {
+ return this.totalSize;
+ }
+
+ /**
+ * Allocate a block with specified size. Return the offset
+ * @param blockSize size of block
+ * @throws BucketAllocatorException,CacheFullException
+ * @return the offset in the IOEngine
+ */
+ public synchronized long allocateBlock(int blockSize) throws CacheFullException,
+ BucketAllocatorException {
+ Preconditions.checkArgument(blockSize > 0);
+ BucketSizeInfo bsi = roundUpToBucketSizeInfo(blockSize);
+ if (bsi == null) {
+ throw new BucketAllocatorException("Allocation too big size=" + blockSize);
+ }
+ long offset = bsi.allocateBlock();
+
+ // Ask caller to free up space and try again!
+ if (offset < 0)
+ throw new CacheFullException(blockSize, bsi.sizeIndex());
+ usedSize += bucketSizes[bsi.sizeIndex()];
+ return offset;
+ }
+
+ private Bucket grabGlobalCompletelyFreeBucket() {
+ for (BucketSizeInfo bsi : bucketSizeInfos) {
+ Bucket b = bsi.findAndRemoveCompletelyFreeBucket();
+ if (b != null) return b;
+ }
+ return null;
+ }
+
+ /**
+ * Free a block with the offset
+ * @param offset block's offset
+ * @return size freed
+ */
+ public synchronized int freeBlock(long offset) {
+ int bucketNo = (int) (offset / bucketCapacity);
+ Preconditions.checkPositionIndex(bucketNo, buckets.length);
+ Bucket targetBucket = buckets[bucketNo];
+ bucketSizeInfos[targetBucket.sizeIndex()].freeBlock(targetBucket, offset);
+ usedSize -= targetBucket.itemAllocationSize();
+ return targetBucket.itemAllocationSize();
+ }
+
+ public int sizeOfAllocation(long offset) {
+ int bucketNo = (int) (offset / bucketCapacity);
+ Preconditions.checkPositionIndex(bucketNo, buckets.length);
+ Bucket targetBucket = buckets[bucketNo];
+ return targetBucket.itemAllocationSize();
+ }
+
+ public static class IndexStatistics {
+ private long freeCount, usedCount, itemSize, totalCount;
+
+ public long freeCount() {
+ return freeCount;
+ }
+
+ public long usedCount() {
+ return usedCount;
+ }
+
+ public long totalCount() {
+ return totalCount;
+ }
+
+ public long freeBytes() {
+ return freeCount * itemSize;
+ }
+
+ public long usedBytes() {
+ return usedCount * itemSize;
+ }
+
+ public long totalBytes() {
+ return totalCount * itemSize;
+ }
+
+ public long itemSize() {
+ return itemSize;
+ }
+
+ public IndexStatistics(long free, long used, long itemSize) {
+ setTo(free, used, itemSize);
+ }
+
+ public IndexStatistics() {
+ setTo(-1, -1, 0);
+ }
+
+ public void setTo(long free, long used, long itemSize) {
+ this.itemSize = itemSize;
+ this.freeCount = free;
+ this.usedCount = used;
+ this.totalCount = free + used;
+ }
+ }
+
+ public void dumpToLog() {
+ logStatistics();
+ StringBuilder sb = new StringBuilder();
+ for (Bucket b : buckets) {
+ sb.append("Bucket:").append(b.baseOffset).append('\n');
+ sb.append(" Size index: " + b.sizeIndex() + "; Free:" + b.freeCount
+ + "; used:" + b.usedCount + "; freelist\n");
+ for (int i = 0; i < b.freeCount(); ++i)
+ sb.append(b.freeList[i]).append(',');
+ sb.append('\n');
+ }
+ LOG.info(sb);
+ }
+
+ public void logStatistics() {
+ IndexStatistics total = new IndexStatistics();
+ IndexStatistics[] stats = getIndexStatistics(total);
+ LOG.info("Bucket allocator statistics follow:\n");
+ LOG.info(" Free bytes=" + total.freeBytes() + "+; used bytes="
+ + total.usedBytes() + "; total bytes=" + total.totalBytes());
+ for (IndexStatistics s : stats) {
+ LOG.info(" Object size " + s.itemSize() + " used=" + s.usedCount()
+ + "; free=" + s.freeCount() + "; total=" + s.totalCount());
+ }
+ }
+
+ public IndexStatistics[] getIndexStatistics(IndexStatistics grandTotal) {
+ IndexStatistics[] stats = getIndexStatistics();
+ long totalfree = 0, totalused = 0;
+ for (IndexStatistics stat : stats) {
+ totalfree += stat.freeBytes();
+ totalused += stat.usedBytes();
+ }
+ grandTotal.setTo(totalfree, totalused, 1);
+ return stats;
+ }
+
+ public IndexStatistics[] getIndexStatistics() {
+ IndexStatistics[] stats = new IndexStatistics[bucketSizes.length];
+ for (int i = 0; i < stats.length; ++i)
+ stats[i] = bucketSizeInfos[i].statistics();
+ return stats;
+ }
+}
+
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocatorException.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocatorException.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocatorException.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketAllocatorException.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,34 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import java.io.IOException;
+
+
+/**
+ * Thrown by {@link BucketAllocator}
+ */
+public class BucketAllocatorException extends IOException {
+ private static final long serialVersionUID = 2479119906660788096L;
+
+ BucketAllocatorException(String reason) {
+ super(reason);
+ }
+}
+
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,993 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.io.HeapSize;
+import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.CachedBlock;
+import org.apache.hadoop.hbase.io.hfile.LruBlockCache;
+import org.apache.hadoop.hbase.regionserver.StoreFile;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.HasThread;
+import org.apache.hadoop.hbase.util.IdLock;
+import org.apache.hadoop.util.StringUtils;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A modified version of BucketCache (by TaoBao/FusionIO) imported from
+ * HBASE-7404 patch. Simplified to only handle byte arrays (for use as a
+ * L2 cache).
+ */
+public class BucketCache implements HeapSize {
+
+ private static final Log LOG = LogFactory.getLog(BucketCache.class);
+
+ // TODO: these must be configurable
+ /** Priority buckets */
+ private static final float DEFAULT_SINGLE_FACTOR = 0.25f;
+ private static final float DEFAULT_MULTI_FACTOR = 0.50f;
+ private static final float DEFAULT_MEMORY_FACTOR = 0.25f;
+ private static final float DEFAULT_EXTRA_FREE_FACTOR = 0.10f;
+
+ private static final float DEFAULT_ACCEPT_FACTOR = 0.95f;
+ private static final float DEFAULT_MIN_FACTOR = 0.85f;
+
+ /** Statistics thread */
+ private static final int statThreadPeriod = 3 * 60;
+
+ public final static int DEFAULT_WRITER_THREADS = 3;
+ public final static int DEFAULT_WRITER_QUEUE_ITEMS = 64;
+
+ // Store/read block data
+ private final IOEngine ioEngine;
+
+ // Store the block in this map before writing it to cache
+ private final ConcurrentHashMap<BlockCacheKey, RAMQueueEntry> ramCache;
+ // In this map, store the block's meta data like offset, length
+ private final ConcurrentHashMap<BlockCacheKey, BucketEntry> backingMap;
+
+ /**
+ * Flag if the cache is enabled or not... We shut it off if there are IO
+ * errors for some time, so that Bucket IO exceptions/errors don't bring down
+ * the HBase server.
+ */
+ private volatile boolean cacheEnabled;
+
+ private final ArrayList<BlockingQueue<RAMQueueEntry>> writerQueues =
+ new ArrayList<BlockingQueue<RAMQueueEntry>>();
+
+ private final WriterThread writerThreads[];
+
+ /** Volatile boolean to track if free space is in process or not */
+ private volatile boolean freeInProgress = false;
+ private final Lock freeSpaceLock = new ReentrantLock();
+
+ private final AtomicLong realCacheSize = new AtomicLong(0);
+ private final AtomicLong heapSize = new AtomicLong(0);
+ /** Current number of cached elements */
+ private final AtomicLong blockNumber = new AtomicLong(0);
+ private final AtomicLong failedBlockAdditions = new AtomicLong(0);
+
+ /** Cache access count (sequential ID) */
+ private final AtomicLong accessCount = new AtomicLong(0);
+
+ private final Object[] cacheWaitSignals;
+ private static final int DEFAULT_CACHE_WAIT_TIME = 50;
+
+
+ // Used in test now. If the flag is false and the cache speed is very fast,
+ // bucket cache will skip some blocks when caching. If the flag is true, we
+ // will wait blocks flushed to IOEngine for some time when caching
+ boolean wait_when_cache = false;
+
+ private BucketCacheStats cacheStats = new BucketCacheStats();
+
+ /** Approximate block size */
+ private final long blockSize;
+
+ /** Duration of IO errors tolerated before we disable cache, 1 min as default */
+ private final int ioErrorsTolerationDuration;
+ // 1 min
+ public static final int DEFAULT_ERROR_TOLERATION_DURATION = 60 * 1000;
+ // Start time of first IO error when reading or writing IO Engine, it will be
+ // reset after a successful read/write.
+ private volatile long ioErrorStartTime = -1;
+
+ /** Minimum buffer size for ByteBufferIOEngine */
+ public static final int MIN_BUFFER_SIZE = 4 * 1024 * 1024;
+
+ /**
+ * A "sparse lock" implementation allowing to lock on a particular block
+ * identified by offset. The purpose of this is to avoid freeing the block
+ * which is being read.
+ *
+ * TODO:We could extend the IdLock to IdReadWriteLock for better.
+ */
+ private IdLock offsetLock = new IdLock();
+
+ private final ConcurrentIndex<String, BlockCacheKey> blocksByHFile =
+ new ConcurrentIndex<String, BlockCacheKey>(new Comparator<BlockCacheKey>() {
+ @Override
+ public int compare(BlockCacheKey a, BlockCacheKey b) {
+ if (a.getOffset() == b.getOffset()) {
+ return 0;
+ } else if (a.getOffset() < b.getOffset()) {
+ return -1;
+ }
+ return 1;
+ }
+ });
+
+ /** Statistics thread schedule pool (for heavy debugging, could remove) */
+ private final ScheduledExecutorService scheduleThreadPool =
+ Executors.newScheduledThreadPool(1,
+ new ThreadFactoryBuilder()
+ .setNameFormat("BucketCache Statistics #%d")
+ .setDaemon(true)
+ .build());
+
+ private final int[] bucketSizes;
+ // Allocate or free space for the block
+ private final BucketAllocator bucketAllocator;
+
+ // TODO (avf): perhaps use a Builder or a separate config object?
+ public BucketCache(String ioEngineName, long capacity, int writerThreadNum,
+ int writerQLen, int ioErrorsTolerationDuration, int[] bucketSizes,
+ Configuration conf)
+ throws IOException {
+ this.bucketSizes = bucketSizes;
+ this.ioEngine = getIOEngineFromName(ioEngineName, capacity, conf);
+ this.writerThreads = new WriterThread[writerThreadNum];
+ this.cacheWaitSignals = new Object[writerThreadNum];
+ long blockNumCapacity = capacity / 16384;
+ if (blockNumCapacity >= Integer.MAX_VALUE) {
+ // Enough for about 32TB of cache!
+ throw new IllegalArgumentException("Cache capacity is too large, only support 32TB now");
+ }
+
+ this.blockSize = StoreFile.DEFAULT_BLOCKSIZE_SMALL;
+ this.ioErrorsTolerationDuration = ioErrorsTolerationDuration;
+
+ bucketAllocator = new BucketAllocator(bucketSizes, capacity);
+ for (int i = 0; i < writerThreads.length; ++i) {
+ writerQueues.add(new ArrayBlockingQueue<RAMQueueEntry>(writerQLen));
+ this.cacheWaitSignals[i] = new Object();
+ }
+
+ this.ramCache = new ConcurrentHashMap<BlockCacheKey, RAMQueueEntry>();
+
+ this.backingMap = new ConcurrentHashMap<BlockCacheKey, BucketEntry>((int) blockNumCapacity);
+
+ final String threadName = Thread.currentThread().getName();
+ this.cacheEnabled = true;
+ for (int i = 0; i < writerThreads.length; ++i) {
+ writerThreads[i] = new WriterThread(writerQueues.get(i), i);
+ writerThreads[i].setName(threadName + "-BucketCacheWriter-" + i);
+ writerThreads[i].start();
+ }
+ // Run the statistics thread periodically to print the cache statistics log
+ this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
+ statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
+ LOG.info("Started bucket cache");
+ }
+
+ /**
+ * Get the IOEngine from the IO engine name
+ * @param ioEngineName Name of the io engine
+ * @param capacity Maximum capacity of the io engine
+ * @param conf Optional configuration object for additional parameters
+ * @return Instance of the correct IOEngine
+ * @throws IllegalArgumentException If the name of the io engine is invalid
+ * @throws IOException If there is an error instantiating the io engine
+ */
+ private IOEngine getIOEngineFromName(String ioEngineName, long capacity,
+ Configuration conf)
+ throws IOException {
+ int requestedBufferSize = conf == null ?
+ CacheConfig.DEFAULT_L2_BUCKET_CACHE_BUFFER_SIZE :
+ conf.getInt(CacheConfig.L2_BUCKET_CACHE_BUFFER_SIZE_KEY,
+ CacheConfig.DEFAULT_L2_BUCKET_CACHE_BUFFER_SIZE);
+ int bufferSize = Math.max(requestedBufferSize, bucketSizes[0]);
+ boolean isDirect;
+ if (ioEngineName.startsWith("offheap"))
+ isDirect = true;
+ else if (ioEngineName.startsWith("heap"))
+ isDirect = false;
+ else
+ throw new IllegalArgumentException(
+ "Don't understand io engine name " + ioEngineName +
+ " for cache. Must be heap or offheap");
+ LOG.info("Initiating ByteBufferIOEngine with " + ioEngineName +
+ " allocation...");
+ if (bufferSize != requestedBufferSize) {
+ LOG.warn("Requested per-buffer size " + requestedBufferSize +
+ ", but actual per-buffer size will be: " + bufferSize);
+ } else {
+ LOG.info("Size per-buffer: " + StringUtils.humanReadableInt(bufferSize));
+ }
+ return new ByteBufferIOEngine(capacity, bufferSize, isDirect);
+ }
+
+ /**
+ * Cache the block with the specified name and buffer.
+ * @param cacheKey block's cache key
+ * @param buf block buffer
+ */
+ public void cacheBlock(BlockCacheKey cacheKey, byte[] buf) {
+ cacheBlock(cacheKey, buf, false);
+ }
+
+ /**
+ * Cache the block with the specified name and buffer.
+ * @param cacheKey block's cache key
+ * @param cachedItem block buffer
+ * @param inMemory if block is in-memory
+ */
+ public void cacheBlock(BlockCacheKey cacheKey, byte[] cachedItem, boolean inMemory) {
+ cacheBlockWithWait(cacheKey, cachedItem, inMemory, wait_when_cache);
+ }
+
+ /**
+ * Cache the block to ramCache
+ * @param cacheKey block's cache key
+ * @param cachedItem block buffer
+ * @param inMemory if block is in-memory
+ * @param wait if true, blocking wait when queue is full
+ */
+ public void cacheBlockWithWait(BlockCacheKey cacheKey, byte[] cachedItem,
+ boolean inMemory, boolean wait) {
+ if (!cacheEnabled)
+ return;
+
+ if (backingMap.containsKey(cacheKey) || ramCache.containsKey(cacheKey))
+ return;
+
+ /*
+ * Stuff the entry into the RAM cache so it can get drained to the
+ * persistent store
+ */
+ RAMQueueEntry re = new RAMQueueEntry(cacheKey, cachedItem,
+ accessCount.incrementAndGet(), inMemory);
+ ramCache.put(cacheKey, re);
+ int queueNum = (cacheKey.hashCode() & 0x7FFFFFFF) % writerQueues.size();
+ BlockingQueue<RAMQueueEntry> bq = writerQueues.get(queueNum);
+ boolean successfulAddition = bq.offer(re);
+ if (!successfulAddition && wait) {
+ synchronized (cacheWaitSignals[queueNum]) {
+ try {
+ cacheWaitSignals[queueNum].wait(DEFAULT_CACHE_WAIT_TIME);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ successfulAddition = bq.offer(re);
+ }
+ if (!successfulAddition) {
+ ramCache.remove(cacheKey);
+ failedBlockAdditions.incrementAndGet();
+ } else {
+ this.blockNumber.incrementAndGet();
+ this.heapSize.addAndGet(cachedItem.length);
+ blocksByHFile.put(cacheKey.getHfileName(), cacheKey);
+ }
+ }
+
+
+ /**
+ * Get the buffer of the block with the specified key.
+ * @param key block's cache key
+ * @param caching true if the caller caches blocks on cache misses
+ */
+ public byte[] getBlock(BlockCacheKey key, boolean caching) {
+ return getBlock(key, caching, false);
+ }
+
+ /**
+ * Get the buffer of the block with the specified key.
+ * @param key block's cache key
+ * @param caching true if the caller caches blocks on cache misses
+ * @param repeat Whether this is a repeat lookup for the same block
+ * @return buffer of specified cache key, or null if not in cache
+ */
+ public byte[] getBlock(BlockCacheKey key, boolean caching, boolean repeat) {
+ if (!cacheEnabled)
+ return null;
+ RAMQueueEntry re = ramCache.get(key);
+ if (re != null) {
+ cacheStats.hit(caching);
+ re.access(accessCount.incrementAndGet());
+ return re.getData();
+ }
+ BucketEntry bucketEntry = backingMap.get(key);
+ if(bucketEntry!=null) {
+ long start = System.nanoTime();
+ IdLock.Entry lockEntry = null;
+ try {
+ lockEntry = offsetLock.getLockEntry(bucketEntry.offset());
+ if (bucketEntry.equals(backingMap.get(key))) {
+ int len = bucketEntry.getLength();
+ byte[] bytes = new byte[len];
+ ioEngine.read(bytes, bucketEntry.offset());
+ long timeTaken = System.nanoTime() - start;
+ cacheStats.hit(caching);
+ cacheStats.ioHit(timeTaken);
+ bucketEntry.access(accessCount.incrementAndGet());
+ if (this.ioErrorStartTime > 0) {
+ ioErrorStartTime = -1;
+ }
+ return bytes;
+ }
+ } catch (IOException ioex) {
+ LOG.error("Failed reading block " + key + " from bucket cache", ioex);
+ checkIOErrorIsTolerated();
+ } finally {
+ if (lockEntry != null) {
+ offsetLock.releaseLockEntry(lockEntry);
+ }
+ }
+ }
+ if(!repeat)cacheStats.miss(caching);
+ return null;
+ }
+
+ public void clearCache() {
+ for (BlockCacheKey key: this.backingMap.keySet()) {
+ evictBlock(key);
+ }
+ }
+
+ public boolean evictBlock(BlockCacheKey cacheKey) {
+ if (!cacheEnabled) return false;
+ RAMQueueEntry removedBlock = ramCache.remove(cacheKey);
+ if (removedBlock != null) {
+ this.blockNumber.decrementAndGet();
+ this.heapSize.addAndGet(-1 * removedBlock.getData().length);
+ }
+ BucketEntry bucketEntry = backingMap.get(cacheKey);
+ if (bucketEntry == null) { return false; }
+ IdLock.Entry lockEntry = null;
+ try {
+ lockEntry = offsetLock.getLockEntry(bucketEntry.offset());
+ if (bucketEntry.equals(backingMap.remove(cacheKey))) {
+ bucketAllocator.freeBlock(bucketEntry.offset());
+ realCacheSize.addAndGet(-1 * bucketEntry.getLength());
+ blocksByHFile.remove(cacheKey.getHfileName(), cacheKey);
+ if (removedBlock == null) {
+ this.blockNumber.decrementAndGet();
+ }
+ } else {
+ return false;
+ }
+ } catch (IOException ie) {
+ LOG.warn("Failed evicting block " + cacheKey);
+ return false;
+ } finally {
+ if (lockEntry != null) {
+ offsetLock.releaseLockEntry(lockEntry);
+ }
+ }
+ cacheStats.evicted(bucketEntry.getPriority());
+ return true;
+ }
+
+ /*
+ * Statistics thread. Periodically prints the cache statistics to the log.
+ */
+ private static class StatisticsThread extends Thread {
+ BucketCache bucketCache;
+
+ public StatisticsThread(BucketCache bucketCache) {
+ super("BucketCache.StatisticsThread");
+ setDaemon(true);
+ this.bucketCache = bucketCache;
+ }
+ @Override
+ public void run() {
+ bucketCache.logStats();
+ }
+ }
+
+ public void logStats() {
+ if (!LOG.isDebugEnabled()) return;
+ // Log size
+ long totalSize = bucketAllocator.getTotalSize();
+ long usedSize = bucketAllocator.getUsedSize();
+ long freeSize = totalSize - usedSize;
+ long cacheSize = this.realCacheSize.get();
+ LOG.debug("BucketCache Stats: " +
+ "failedBlockAdditions=" + this.failedBlockAdditions.get() + ", " +
+ "total=" + StringUtils.byteDesc(totalSize) + ", " +
+ "free=" + StringUtils.byteDesc(freeSize) + ", " +
+ "usedSize=" + StringUtils.byteDesc(usedSize) +", " +
+ "cacheSize=" + StringUtils.byteDesc(cacheSize) +", " +
+ "accesses=" + cacheStats.getRequestCount() + ", " +
+ "hits=" + cacheStats.getHitCount() + ", " +
+ "IOhitsPerSecond=" + cacheStats.getIOHitsPerSecond() + ", " +
+ "IOTimePerHit=" + String.format("%.2f", cacheStats.getIOTimePerHit())+ ", " +
+ "hitRatio=" + (cacheStats.getHitCount() == 0 ? "0," :
+ (StringUtils.formatPercent(cacheStats.getHitRatio(), 2)+ ", ")) +
+ "cachingAccesses=" + cacheStats.getRequestCachingCount() + ", " +
+ "cachingHits=" + cacheStats.getHitCachingCount() + ", " +
+ "cachingHitsRatio=" +(cacheStats.getHitCachingCount() == 0 ? "0," :
+ (StringUtils.formatPercent(cacheStats.getHitCachingRatio(), 2)+ ", ")) +
+ "evictions=" + cacheStats.getEvictionCount() + ", " +
+ "evicted=" + cacheStats.getEvictedCount() + ", " +
+ "evictedPerRun=" + cacheStats.evictedPerEviction());
+ cacheStats.reset();
+ }
+
+ private long acceptableSize() {
+ return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_ACCEPT_FACTOR);
+ }
+
+ private long minSize() {
+ return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_MIN_FACTOR);
+ }
+
+ private long singleSize() {
+ return (long) Math.floor(bucketAllocator.getTotalSize()
+ * DEFAULT_SINGLE_FACTOR * DEFAULT_MIN_FACTOR);
+ }
+
+ private long multiSize() {
+ return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_MULTI_FACTOR
+ * DEFAULT_MIN_FACTOR);
+ }
+
+ private long memorySize() {
+ return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_MEMORY_FACTOR
+ * DEFAULT_MIN_FACTOR);
+ }
+
+ /**
+ * Free the space if the used size reaches acceptableSize() or one size block
+ * couldn't be allocated. When freeing the space, we use the LRU algorithm and
+ * ensure there must be some blocks evicted
+ */
+ private void freeSpace() {
+ // Ensure only one freeSpace progress at a time
+ if (!freeSpaceLock.tryLock()) return;
+ try {
+ freeInProgress = true;
+ long bytesToFreeWithoutExtra = 0;
+ /*
+ * Calculate free byte for each bucketSizeinfo
+ */
+ StringBuffer msgBuffer = new StringBuffer();
+ BucketAllocator.IndexStatistics[] stats = bucketAllocator.getIndexStatistics();
+ long[] bytesToFreeForBucket = new long[stats.length];
+ for (int i = 0; i < stats.length; i++) {
+ bytesToFreeForBucket[i] = 0;
+ long freeGoal = (long) Math.floor(stats[i].totalCount()
+ * (1 - DEFAULT_MIN_FACTOR));
+ freeGoal = Math.max(freeGoal, 1);
+ if (stats[i].freeCount() < freeGoal) {
+ bytesToFreeForBucket[i] = stats[i].itemSize()
+ * (freeGoal - stats[i].freeCount());
+ bytesToFreeWithoutExtra += bytesToFreeForBucket[i];
+ msgBuffer.append("Free for bucketSize(" + stats[i].itemSize() + ")="
+ + StringUtils.byteDesc(bytesToFreeForBucket[i]) + ", ");
+ }
+ }
+ msgBuffer.append("Free for total="
+ + StringUtils.byteDesc(bytesToFreeWithoutExtra) + ", ");
+
+ if (bytesToFreeWithoutExtra <= 0) {
+ return;
+ }
+ long currentSize = bucketAllocator.getUsedSize();
+ long totalSize=bucketAllocator.getTotalSize();
+ LOG.debug("Bucket cache free space started; Attempting to " + msgBuffer.toString()
+ + " of current used=" + StringUtils.byteDesc(currentSize)
+ + ",actual cacheSize=" + StringUtils.byteDesc(realCacheSize.get())
+ + ",total=" + StringUtils.byteDesc(totalSize));
+
+ long bytesToFreeWithExtra = (long) Math.floor(bytesToFreeWithoutExtra
+ * (1 + DEFAULT_EXTRA_FREE_FACTOR));
+
+ // Instantiate priority buckets
+ BucketEntryGroup bucketSingle = new BucketEntryGroup(bytesToFreeWithExtra,
+ blockSize, singleSize());
+ BucketEntryGroup bucketMulti = new BucketEntryGroup(bytesToFreeWithExtra,
+ blockSize, multiSize());
+ BucketEntryGroup bucketMemory = new BucketEntryGroup(bytesToFreeWithExtra,
+ blockSize, memorySize());
+
+ // Scan entire map putting bucket entry into appropriate bucket entry
+ // group
+ for (Map.Entry<BlockCacheKey, BucketEntry> bucketEntryWithKey : backingMap.entrySet()) {
+ switch (bucketEntryWithKey.getValue().getPriority()) {
+ case SINGLE: {
+ bucketSingle.add(bucketEntryWithKey);
+ break;
+ }
+ case MULTI: {
+ bucketMulti.add(bucketEntryWithKey);
+ break;
+ }
+ case MEMORY: {
+ bucketMemory.add(bucketEntryWithKey);
+ break;
+ }
+ }
+ }
+
+ PriorityQueue<BucketEntryGroup> bucketQueue = new PriorityQueue<BucketEntryGroup>(3);
+
+ bucketQueue.add(bucketSingle);
+ bucketQueue.add(bucketMulti);
+ bucketQueue.add(bucketMemory);
+
+ int remainingBuckets = 3;
+ long bytesFreed = 0;
+
+ BucketEntryGroup bucketGroup;
+ while ((bucketGroup = bucketQueue.poll()) != null) {
+ long overflow = bucketGroup.overflow();
+ if (overflow > 0) {
+ long bucketBytesToFree = Math.min(overflow,
+ (bytesToFreeWithoutExtra - bytesFreed) / remainingBuckets);
+ bytesFreed += bucketGroup.free(bucketBytesToFree);
+ }
+ remainingBuckets--;
+ }
+
+ /**
+ * Check whether need extra free because some bucketSizeinfo still needs
+ * free space
+ */
+ stats = bucketAllocator.getIndexStatistics();
+ boolean needFreeForExtra = false;
+ for (int i = 0; i < stats.length; i++) {
+ long freeGoal = (long) Math.floor(stats[i].totalCount()
+ * (1 - DEFAULT_MIN_FACTOR));
+ freeGoal = Math.max(freeGoal, 1);
+ if (stats[i].freeCount() < freeGoal) {
+ needFreeForExtra = true;
+ break;
+ }
+ }
+
+ if (needFreeForExtra) {
+ bucketQueue.clear();
+ remainingBuckets = 2;
+
+ bucketQueue.add(bucketSingle);
+ bucketQueue.add(bucketMulti);
+
+ while ((bucketGroup = bucketQueue.poll()) != null) {
+ long bucketBytesToFree = (bytesToFreeWithExtra - bytesFreed)
+ / remainingBuckets;
+ bytesFreed += bucketGroup.free(bucketBytesToFree);
+ remainingBuckets--;
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ long single = bucketSingle.totalSize();
+ long multi = bucketMulti.totalSize();
+ long memory = bucketMemory.totalSize();
+ LOG.debug("Bucket cache free space completed; " + "freed="
+ + StringUtils.byteDesc(bytesFreed) + ", " + "total="
+ + StringUtils.byteDesc(totalSize) + ", " + "single="
+ + StringUtils.byteDesc(single) + ", " + "multi="
+ + StringUtils.byteDesc(multi) + ", " + "memory="
+ + StringUtils.byteDesc(memory));
+ }
+
+ } finally {
+ cacheStats.evict();
+ freeInProgress = false;
+ freeSpaceLock.unlock();
+ }
+ }
+
+ // This handles flushing the RAM cache to IOEngine.
+ private class WriterThread extends HasThread {
+ BlockingQueue<RAMQueueEntry> inputQueue;
+ final int threadIdx;
+ boolean writerEnabled = true;
+
+ WriterThread(BlockingQueue<RAMQueueEntry> queue, int threadNO) {
+ super();
+ this.inputQueue = queue;
+ this.threadIdx = threadNO;
+ setDaemon(true);
+ }
+
+ // Used for test
+ void disableWriter() {
+ this.writerEnabled = false;
+ }
+
+ public void run() {
+ List<RAMQueueEntry> entries = new ArrayList<RAMQueueEntry>();
+ try {
+ while (cacheEnabled && writerEnabled) {
+ try {
+ // Perform a block take first in order to avoid a drainTo()
+ // on an empty queue looping around and causing starvation.
+ entries.add(inputQueue.take());
+ inputQueue.drainTo(entries);
+ synchronized (cacheWaitSignals[threadIdx]) {
+ cacheWaitSignals[threadIdx].notifyAll();
+ }
+ } catch (InterruptedException ie) {
+ if (!cacheEnabled) break;
+ }
+ doDrain(entries);
+ }
+ } catch (Throwable t) {
+ LOG.warn("Failed doing drain", t);
+ }
+ LOG.info(this.getName() + " exiting, cacheEnabled=" + cacheEnabled);
+ }
+
+ /**
+ * Flush the entries in ramCache to IOEngine and add bucket entry to
+ * backingMap
+ * @param entries
+ * @throws InterruptedException
+ */
+ private void doDrain(List<RAMQueueEntry> entries)
+ throws InterruptedException {
+ BucketEntry[] bucketEntries = new BucketEntry[entries.size()];
+ RAMQueueEntry[] ramEntries = new RAMQueueEntry[entries.size()];
+ int done = 0;
+ while (entries.size() > 0 && cacheEnabled) {
+ // Keep going in case we throw...
+ RAMQueueEntry ramEntry = null;
+ try {
+ ramEntry = entries.remove(entries.size() - 1);
+ if (ramEntry == null) {
+ LOG.warn("Couldn't get the entry from RAM queue, who steals it?");
+ continue;
+ }
+ BucketEntry bucketEntry = ramEntry.writeToCache(ioEngine,
+ bucketAllocator, realCacheSize);
+ ramEntries[done] = ramEntry;
+ bucketEntries[done++] = bucketEntry;
+ if (ioErrorStartTime > 0) {
+ ioErrorStartTime = -1;
+ }
+ } catch (BucketAllocatorException fle) {
+ LOG.warn("Failed allocating for block "
+ + (ramEntry == null ? "" : ramEntry.getKey()), fle);
+ } catch (CacheFullException cfe) {
+ if (!freeInProgress) {
+ freeSpace();
+ } else {
+ Thread.sleep(50);
+ }
+ } catch (IOException ioex) {
+ LOG.error("Failed writing to bucket cache", ioex);
+ checkIOErrorIsTolerated();
+ }
+ }
+
+ // Make sure that the data pages we have written are on the media before
+ // we update the map.
+ try {
+ ioEngine.sync();
+ } catch (IOException ioex) {
+ LOG.error("Faild syncing IO engine", ioex);
+ checkIOErrorIsTolerated();
+ // Since we failed sync, free the blocks in bucket allocator
+ for (int i = 0; i < done; ++i) {
+ if (bucketEntries[i] != null) {
+ bucketAllocator.freeBlock(bucketEntries[i].offset());
+ }
+ }
+ done = 0;
+ }
+
+ for (int i = 0; i < done; ++i) {
+ if (bucketEntries[i] != null) {
+ backingMap.put(ramEntries[i].getKey(), bucketEntries[i]);
+ }
+ RAMQueueEntry ramCacheEntry = ramCache.remove(ramEntries[i].getKey());
+ if (ramCacheEntry != null) {
+ heapSize.addAndGet(-1 * ramEntries[i].getData().length);
+ }
+ }
+
+ if (bucketAllocator.getUsedSize() > acceptableSize()) {
+ freeSpace();
+ }
+ }
+ }
+
+ /**
+ * Check whether we tolerate IO error this time. If the duration of IOEngine
+ * throwing errors exceeds ioErrorsDurationTimeTolerated, we will disable the
+ * cache
+ */
+ private void checkIOErrorIsTolerated() {
+ long now = EnvironmentEdgeManager.currentTimeMillis();
+ if (this.ioErrorStartTime > 0) {
+ if (cacheEnabled
+ && (now - ioErrorStartTime) > this.ioErrorsTolerationDuration) {
+ LOG.error("IO errors duration time has exceeded "
+ + ioErrorsTolerationDuration
+ + "ms, disabing cache, please check your IOEngine");
+ disableCache();
+ }
+ } else {
+ this.ioErrorStartTime = now;
+ }
+ }
+
+ /**
+ * Used to shut down the cache -or- turn it off in the case of something
+ * broken.
+ */
+ private void disableCache() {
+ if (!cacheEnabled)
+ return;
+ cacheEnabled = false;
+ ioEngine.shutdown();
+ this.scheduleThreadPool.shutdown();
+ for (int i = 0; i < writerThreads.length; ++i)
+ writerThreads[i].interrupt();
+ this.ramCache.clear();
+ this.backingMap.clear();
+ }
+
+ private void join() throws InterruptedException {
+ for (int i = 0; i < writerThreads.length; ++i)
+ writerThreads[i].join();
+ }
+
+ public void shutdown() {
+ disableCache();
+ LOG.info("Shutting down bucket cache");
+ }
+
+ public LruBlockCache.CacheStats getStats() {
+ return cacheStats;
+ }
+
+ BucketAllocator getAllocator() {
+ return this.bucketAllocator;
+ }
+
+ public long heapSize() {
+ return this.heapSize.get();
+ }
+
+ /**
+ * Returns the total size of the block cache, in bytes.
+ * @return size of cache, in bytes
+ */
+ public long size() {
+ return this.realCacheSize.get();
+ }
+
+ public long getFreeSize() {
+ return this.bucketAllocator.getTotalSize() - this.bucketAllocator.getUsedSize();
+ }
+
+ public long getBlockCount() {
+ return this.blockNumber.get();
+ }
+
+ public long getEvictedCount() {
+ return cacheStats.getEvictedCount();
+ }
+
+ /**
+ * Evicts all blocks for a specific HFile. This is an expensive operation
+ * implemented as a linear-time search through all blocks in the cache.
+ * Ideally this should be a search in a log-access-time map.
+ *
+ * <p>
+ * This is used for evict-on-close to remove all blocks of a specific HFile.
+ *
+ * @return the number of blocks evicted
+ */
+ public int evictBlocksByHfileName(String hfileName) {
+ // Copy the list to avoid ConcurrentModificationException
+ // as evictBlockKey removes the key from the index
+ Set<BlockCacheKey> keySet = blocksByHFile.values(hfileName);
+ if (keySet == null) {
+ return 0;
+ }
+ int numEvicted = 0;
+ List<BlockCacheKey> keysForHFile =
+ ImmutableList.copyOf(keySet);
+ for (BlockCacheKey key : keysForHFile) {
+ if (evictBlock(key)) {
+ ++numEvicted;
+ }
+ }
+ return numEvicted;
+ }
+
+
+ /**
+ * Item in cache. We expect this to be where most memory goes. Java uses 8
+ * bytes just for object headers; after this, we want to use as little as
+ * possible - so we only use 8 bytes, but in order to do so we end up messing
+ * around with all this Java casting stuff. Offset stored as 5 bytes that make
+ * up the long. Doubt we'll see devices this big for ages. Offsets are divided
+ * by 256. So 5 bytes gives us 256TB or so.
+ */
+ static class BucketEntry implements Serializable, Comparable<BucketEntry> {
+ private static final long serialVersionUID = -6741504807982257534L;
+ private int offsetBase;
+ private int length;
+ private byte offset1;
+ private volatile long accessTime;
+ private CachedBlock.BlockPriority priority;
+
+ BucketEntry(long offset, int length, long accessTime, boolean inMemory) {
+ setOffset(offset);
+ this.length = length;
+ this.accessTime = accessTime;
+ if (inMemory) {
+ this.priority = CachedBlock.BlockPriority.MEMORY;
+ } else {
+ this.priority = CachedBlock.BlockPriority.SINGLE;
+ }
+ }
+
+ long offset() { // Java has no unsigned numbers
+ long o = ((long) offsetBase) & 0xFFFFFFFF;
+ o += (((long) (offset1)) & 0xFF) << 32;
+ return o << 8;
+ }
+
+ private void setOffset(long value) {
+ Preconditions.checkArgument((value & 0xFF) == 0);
+ value >>= 8;
+ offsetBase = (int) value;
+ offset1 = (byte) (value >> 32);
+ }
+
+ public int getLength() {
+ return length;
+ }
+
+ /**
+ * Block has been accessed. Update its local access time.
+ */
+ public void access(long accessTime) {
+ this.accessTime = accessTime;
+ if (this.priority == CachedBlock.BlockPriority.SINGLE) {
+ this.priority = CachedBlock.BlockPriority.MULTI;
+ }
+ }
+
+ public CachedBlock.BlockPriority getPriority() {
+ return this.priority;
+ }
+
+ @Override
+ public int compareTo(BucketEntry that) {
+ if(this.accessTime == that.accessTime) return 0;
+ return this.accessTime < that.accessTime ? 1 : -1;
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ return this == that;
+ }
+ }
+
+ /**
+ * Used to group bucket entries into priority buckets. There will be a
+ * BucketEntryGroup for each priority (single, multi, memory). Once bucketed,
+ * the eviction algorithm takes the appropriate number of elements out of each
+ * according to configuration parameters and their relative sizes.
+ */
+ private class BucketEntryGroup implements Comparable<BucketEntryGroup> {
+
+ private CachedEntryQueue queue;
+ private long totalSize = 0;
+ private long bucketSize;
+
+ public BucketEntryGroup(long bytesToFree, long blockSize, long bucketSize) {
+ this.bucketSize = bucketSize;
+ queue = new CachedEntryQueue(bytesToFree, blockSize);
+ totalSize = 0;
+ }
+
+ public void add(Map.Entry<BlockCacheKey, BucketEntry> block) {
+ totalSize += block.getValue().getLength();
+ queue.add(block);
+ }
+
+ /**
+ * Free specified number of bytes by freeing the least recently used
+ * buckets entries.
+ * @param toFree Number of bytes we want to free
+ * @return Number of bytes freed
+ */
+ public long free(long toFree) {
+ Map.Entry<BlockCacheKey, BucketEntry> entry;
+ long freedBytes = 0;
+ while ((entry = queue.pollLast()) != null) {
+ evictBlock(entry.getKey());
+ freedBytes += entry.getValue().getLength();
+ if (freedBytes >= toFree) {
+ return freedBytes;
+ }
+ }
+ return freedBytes;
+ }
+
+ public long overflow() {
+ return totalSize - bucketSize;
+ }
+
+ public long totalSize() {
+ return totalSize;
+ }
+
+ @Override
+ public int compareTo(BucketEntryGroup that) {
+ if (this.overflow() == that.overflow())
+ return 0;
+ return this.overflow() > that.overflow() ? 1 : -1;
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ return this == that;
+ }
+
+ }
+
+ /**
+ * Only used in test
+ * @throws InterruptedException
+ */
+ void stopWriterThreads() throws InterruptedException {
+ for (WriterThread writerThread : writerThreads) {
+ writerThread.disableWriter();
+ writerThread.interrupt();
+ writerThread.join();
+ }
+ }
+}
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCacheStats.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCacheStats.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCacheStats.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCacheStats.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,61 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hadoop.hbase.io.hfile.LruBlockCache;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+
+/**
+ * Class that implements cache metrics for bucket cache.
+ */
+public class BucketCacheStats extends LruBlockCache.CacheStats {
+ private static final int NS_PER_SECOND = 1000000;
+
+ private final AtomicLong ioHitCount = new AtomicLong(0);
+ private final AtomicLong ioHitTime = new AtomicLong(0);
+
+ private volatile long lastLogTime = EnvironmentEdgeManager.currentTimeMillis();
+
+ public void ioHit(long time) {
+ ioHitCount.incrementAndGet();
+ ioHitTime.addAndGet(time);
+ }
+
+ public long getIOHitsPerSecond() {
+ long now = EnvironmentEdgeManager.currentTimeMillis();
+ long took = (now - lastLogTime) / 1000;
+ lastLogTime = now;
+ return took == 0 ? 0 : ioHitCount.get() / took;
+ }
+
+ public double getIOTimePerHit() {
+ long time = ioHitTime.get() / NS_PER_SECOND;
+ long count = ioHitCount.get();
+ return count == 0 ? 0.0 : ((double) time / (double) count);
+ }
+
+ public void reset() {
+ ioHitCount.set(0);
+ ioHitTime.set(0);
+ }
+}
+
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ByteBufferIOEngine.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,83 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.util.ByteBufferArray;
+
+/**
+ * IO engine that stores data on the memory using an array of ByteBuffers
+ * {@link ByteBufferArray}
+ */
+public class ByteBufferIOEngine implements IOEngine {
+
+ private ByteBufferArray bufferArray;
+
+ /**
+ * Construct the ByteBufferIOEngine with the given capacity
+ * @param capacity
+ * @param direct true if allocate direct buffer
+ * @throws IOException
+ */
+ public ByteBufferIOEngine(long capacity, int bufferSize, boolean direct)
+ throws IOException {
+ bufferArray = new ByteBufferArray(capacity, bufferSize, direct);
+ }
+
+ /**
+ * Transfers data from the buffer array to the given byte buffer
+ * @param dst the given byte array into which bytes are to be written
+ * @param offset The offset in the ByteBufferArray of the first byte to be
+ * read
+ * @throws IOException
+ */
+ @Override
+ public void read(byte[] dst, long offset) throws IOException {
+ bufferArray.getMultiple(offset, dst.length, dst, 0);
+ }
+
+ /**
+ * Transfers data from the given byte buffer to the buffer array
+ * @param src the given byte array from which bytes are to be read
+ * @param offset The offset in the ByteBufferArray of the first byte to be
+ * written
+ * @throws IOException
+ */
+ @Override
+ public void write(byte[] src, long offset) throws IOException {
+ bufferArray.putMultiple(offset, src.length, src, 0);
+ }
+
+ /**
+ * No operation for the sync in the memory IO engine
+ */
+ @Override
+ public void sync() {
+
+ }
+
+ /**
+ * No operation for the shutdown in the memory IO engine
+ */
+ @Override
+ public void shutdown() {
+
+ }
+}
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CacheFullException.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CacheFullException.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CacheFullException.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CacheFullException.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,54 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import java.io.IOException;
+
+
+/**
+ * Thrown by {@link BucketAllocator#allocateBlock(int)} when cache is full for
+ * the requested size
+ */
+public class CacheFullException extends IOException {
+ private static final long serialVersionUID = 3265127301824638920L;
+ private int requestedSize, bucketIndex;
+
+ CacheFullException(int requestedSize, int bucketIndex) {
+ super();
+ this.requestedSize = requestedSize;
+ this.bucketIndex = bucketIndex;
+ }
+
+ public int bucketIndex() {
+ return bucketIndex;
+ }
+
+ public int requestedSize() {
+ return requestedSize;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder(1024);
+ sb.append("Allocator requested size ").append(requestedSize);
+ sb.append(" for bucket ").append(bucketIndex);
+ return sb.toString();
+ }
+}
+
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CachedEntryQueue.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CachedEntryQueue.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CachedEntryQueue.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/CachedEntryQueue.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,120 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
+import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.BucketEntry;
+
+import com.google.common.collect.MinMaxPriorityQueue;
+
+/**
+ * A memory-bound queue that will grow until an element brings total size larger
+ * than maxSize. From then on, only entries that are sorted larger than the
+ * smallest current entry will be inserted/replaced.
+ *
+ * <p>
+ * Use this when you want to find the largest elements (according to their
+ * ordering, not their heap size) that consume as close to the specified maxSize
+ * as possible. Default behavior is to grow just above rather than just below
+ * specified max.
+ */
+public class CachedEntryQueue {
+
+ private MinMaxPriorityQueue<Map.Entry<BlockCacheKey, BucketEntry>> queue;
+
+ private long cacheSize;
+ private long maxSize;
+
+ /**
+ * @param maxSize the target size of elements in the queue
+ * @param blockSize expected average size of blocks
+ */
+ public CachedEntryQueue(long maxSize, long blockSize) {
+ int initialSize = (int) (maxSize / blockSize);
+ if (initialSize == 0)
+ initialSize++;
+ queue = MinMaxPriorityQueue
+ .orderedBy(new Comparator<Map.Entry<BlockCacheKey, BucketEntry>>() {
+ public int compare(Entry<BlockCacheKey, BucketEntry> entry1,
+ Entry<BlockCacheKey, BucketEntry> entry2) {
+ return entry1.getValue().compareTo(entry2.getValue());
+ }
+
+ }).expectedSize(initialSize).create();
+ cacheSize = 0;
+ this.maxSize = maxSize;
+ }
+
+ /**
+ * Attempt to add the specified entry to this queue.
+ *
+ * <p>
+ * If the queue is smaller than the max size, or if the specified element is
+ * ordered after the smallest element in the queue, the element will be added
+ * to the queue. Otherwise, there is no side effect of this call.
+ * @param entry a bucket entry with key to try to add to the queue
+ */
+ public void add(Map.Entry<BlockCacheKey, BucketEntry> entry) {
+ if (cacheSize < maxSize) {
+ queue.add(entry);
+ cacheSize += entry.getValue().getLength();
+ } else {
+ BucketEntry head = queue.peek().getValue();
+ if (entry.getValue().compareTo(head) > 0) {
+ cacheSize += entry.getValue().getLength();
+ cacheSize -= head.getLength();
+ if (cacheSize > maxSize) {
+ queue.poll();
+ } else {
+ cacheSize += head.getLength();
+ }
+ queue.add(entry);
+ }
+ }
+ }
+
+ /**
+ * @return The next element in this queue, or {@code null} if the queue is
+ * empty.
+ */
+ public Map.Entry<BlockCacheKey, BucketEntry> poll() {
+ return queue.poll();
+ }
+
+ /**
+ * @return The last element in this queue, or {@code null} if the queue is
+ * empty.
+ */
+ public Map.Entry<BlockCacheKey, BucketEntry> pollLast() {
+ return queue.pollLast();
+ }
+
+ /**
+ * Total size of all elements in this queue.
+ * @return size of all elements currently in queue, in bytes
+ */
+ public long cacheSize() {
+ return cacheSize;
+ }
+}
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ConcurrentIndex.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ConcurrentIndex.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ConcurrentIndex.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/ConcurrentIndex.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,171 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+
+import com.google.common.base.Supplier;
+import com.google.common.collect.Multiset;
+
+import java.util.Comparator;
+import java.util.ConcurrentModificationException;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+/**
+ * A simple concurrent map of sets. This is similar in concept to
+ * {@link Multiset}, with the following exceptions:
+ * <ul>
+ * <li>The set is thread-safe and concurrent: no external locking or
+ * synchronization is required. This is important for the use case where
+ * this class is used to index cached blocks by filename for their
+ * efficient eviction from cache when the file is closed or compacted.</li>
+ * <li>The expectation is that all entries may only be removed for a key
+ * once no more additions of values are being made under that key.</li>
+ * </ul>
+ * @see BucketCache#evictBlocksByHfileName(String) for example use of this
+ * class.
+ * @param <K> Key type
+ * @param <V> Value type
+ */
+public class ConcurrentIndex<K, V> {
+
+ /** Container for the sets, indexed by key */
+ private final ConcurrentMap<K, Set<V>> container;
+
+ /**
+ * A factory that constructs new instances of the sets if no set is
+ * associated with a given key.
+ */
+ private final Supplier<Set<V>> valueSetFactory;
+
+ /**
+ * Creates an instance with a specified factory object for sets to be
+ * associated with a given key.
+ * @param valueSetFactory The factory instance
+ */
+ public ConcurrentIndex(Supplier<Set<V>> valueSetFactory) {
+ this.valueSetFactory = valueSetFactory;
+ this.container = new ConcurrentHashMap<K, Set<V>>();
+ }
+
+ /**
+ * Creates an instance using the {@link DefaultValueSetFactory} for sets,
+ * which in turn creates instances of {@link ConcurrentSkipListSet}
+ * @param valueComparator A {@link Comparator} for value types
+ */
+ public ConcurrentIndex(Comparator<V> valueComparator) {
+ this(new DefaultValueSetFactory<V>(valueComparator));
+ }
+
+ /**
+ * Associate a new unique value with a specified key. Under the covers, the
+ * method employs optimistic concurrency: if no set is associated with a
+ * given key, we create a new set; if another thread comes in, creates,
+ * and associates a set with the same key in the mean-time, we simply add
+ * the value to the already created set.
+ * @param key The key
+ * @param value An additional unique value we want to associate with a key
+ */
+ public void put(K key, V value) {
+ Set<V> set = container.get(key);
+ if (set != null) {
+ set.add(value);
+ } else {
+ set = valueSetFactory.get();
+ set.add(value);
+ Set<V> existing = container.putIfAbsent(key, set);
+ if (existing != null) {
+ // If a set is already associated with a key, that means another
+ // writer has already come in and created the set for the given key.
+ // Pursuant to an optimistic concurrency policy, in this case we will
+ // simply add the value to the existing set associated with the key.
+ existing.add(value);
+ }
+ }
+ }
+
+ /**
+ * Get all values associated with a specified key or null if no values are
+ * associated. <b>Note:</b> if the caller wishes to add or removes values
+ * to under the specified as they're iterating through the returned value,
+ * they should make a defensive copy; otherwise, a
+ * {@link ConcurrentModificationException} may be thrown.
+ * @see BucketCache#evictBlocksByHfileName(String) for example usage
+ * @param key The key
+ * @return All values associated with the specified key or null if no values
+ * are associated with the key.
+ */
+ public Set<V> values(K key) {
+ return container.get(key);
+ }
+
+ /**
+ * Removes the association between a specified key and value. If as a
+ * result of removing a value a set becomes empty, we remove the given
+ * set from the mapping as well.
+ * @param key The specified key
+ * @param value The value to disassociate with the key
+ */
+ public boolean remove(K key, V value) {
+ Set<V> set = container.get(key);
+ boolean success = false;
+ if (set != null) {
+ success = set.remove(value);
+ if (set.isEmpty()) {
+ container.remove(key);
+ }
+ }
+ return success;
+ }
+
+ /**
+ * Default factory class for the sets associated with given keys. Creates
+ * a {@link ConcurrentSkipListSet} using the comparator passed into the
+ * constructor.
+ * @see ConcurrentSkipListSet
+ * @see Supplier
+ * @param <V> The value type. Should match value type of the
+ * ConcurrentIndex instances of this object are passed to.
+ */
+ static class DefaultValueSetFactory<V> implements Supplier<Set<V>> {
+ private final Comparator<V> comparator;
+
+ /**
+ * Creates an instance that passes a specified comparator to the
+ * {@link ConcurrentSkipListSet}
+ * @param comparator The specified comparator
+ */
+ public DefaultValueSetFactory(Comparator<V> comparator) {
+ this.comparator = comparator;
+ }
+
+ /**
+ * Creates a new {@link ConcurrentSkipListSet} instance using the
+ * comparator specified when the class instance was constructor.
+ * @return The instantiated {@link ConcurrentSkipListSet} object
+ */
+ @Override
+ public Set<V> get() {
+ return new ConcurrentSkipListSet<V>(comparator);
+ }
+ }
+}
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/IOEngine.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/IOEngine.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/IOEngine.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/IOEngine.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,59 @@
+/**
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * A class implementing IOEngine interface could support data services for
+ * {@link BucketCache}.
+ */
+public interface IOEngine {
+
+ /**
+ * Transfers data from IOEngine to the given byte buffer
+ * @param dst the given byte array into which bytes are to be written
+ * @param offset The offset in the IO engine where the first byte to be read
+ * @throws IOException
+ */
+ void read(byte[] dst, long offset) throws IOException;
+
+ /**
+ * Transfers data from the given byte buffer to IOEngine
+ * @param src the given byte array from which bytes are to be read
+ * @param offset The offset in the IO engine where the first byte to be
+ * written
+ * @throws IOException
+ */
+ void write(byte[] src, long offset) throws IOException;
+
+ /**
+ * Sync the data to IOEngine after writing
+ * @throws IOException
+ */
+ void sync() throws IOException;
+
+ /**
+ * Shutdown the IOEngine
+ */
+ void shutdown();
+}
Added: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/RAMQueueEntry.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/RAMQueueEntry.java?rev=1492797&view=auto
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/RAMQueueEntry.java (added)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/RAMQueueEntry.java Thu Jun 13 18:18:20 2013
@@ -0,0 +1,79 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.io.hfile.bucket;
+
+import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Block Entry stored in the memory with key,data and so on
+ */
+class RAMQueueEntry {
+ private BlockCacheKey key;
+ private byte[] data;
+ private long accessTime;
+ private boolean inMemory;
+
+ public RAMQueueEntry(BlockCacheKey bck, byte[] data, long accessTime,
+ boolean inMemory) {
+ this.key = bck;
+ this.data = data;
+ this.accessTime = accessTime;
+ this.inMemory = inMemory;
+ }
+
+ public byte[] getData() {
+ return data;
+ }
+
+ public BlockCacheKey getKey() {
+ return key;
+ }
+
+ public void access(long accessTime) {
+ this.accessTime = accessTime;
+ }
+
+ public BucketCache.BucketEntry writeToCache(final IOEngine ioEngine,
+ final BucketAllocator bucketAllocator,
+ final AtomicLong realCacheSize) throws CacheFullException, IOException,
+ BucketAllocatorException {
+ int len = data.length;
+ if (len == 0) {
+ return null;
+ }
+ long offset = bucketAllocator.allocateBlock(len);
+ BucketCache.BucketEntry bucketEntry = new BucketCache.BucketEntry(offset, len, accessTime,
+ inMemory);
+ try {
+ ioEngine.write(data, offset);
+ } catch (IOException ioe) {
+ // free it in bucket allocator
+ bucketAllocator.freeBlock(offset);
+ throw ioe;
+ }
+
+ realCacheSize.addAndGet(len);
+ return bucketEntry;
+ }
+}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1492797&r1=1492796&r2=1492797&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Thu Jun 13 18:18:20 2013
@@ -82,6 +82,7 @@ import org.apache.hadoop.hbase.io.HeapSi
import org.apache.hadoop.hbase.io.Reference.Range;
import org.apache.hadoop.hbase.io.hfile.BlockCache;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.L2Cache;
import org.apache.hadoop.hbase.ipc.HRegionInterface;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.monitoring.TaskMonitor;
@@ -3901,7 +3902,12 @@ public class HRegion implements HeapSize
log.close();
// TODO: is this still right?
BlockCache bc = new CacheConfig(c).getBlockCache();
- if (bc != null) bc.shutdown();
+ L2Cache l2c = new CacheConfig(c).getL2Cache();
+ try {
+ if (bc != null) bc.shutdown();
+ } finally {
+ if (l2c != null) l2c.shutdown();
+ }
}
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1492797&r1=1492796&r2=1492797&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java Thu Jun 13 18:18:20 2013
@@ -110,6 +110,7 @@ import org.apache.hadoop.hbase.client.Se
import org.apache.hadoop.hbase.client.ServerConnectionManager;
import org.apache.hadoop.hbase.conf.ConfigurationManager;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.L2BucketCache;
import org.apache.hadoop.hbase.io.hfile.LruBlockCache;
import org.apache.hadoop.hbase.io.hfile.LruBlockCache.CacheStats;
import org.apache.hadoop.hbase.ipc.HBaseRPC;
@@ -1052,6 +1053,10 @@ public class HRegionServer implements HR
cacheConfig.getBlockCache().shutdown();
}
+ if (cacheConfig.isL2CacheEnabled()) {
+ cacheConfig.getL2Cache().shutdown();
+ }
+
// Send interrupts to wake up threads if sleeping so they notice shutdown.
// TODO: Should we check they are alive? If OOME could have exited already
cacheFlusher.interruptIfNecessary();
@@ -1630,7 +1635,21 @@ public class HRegionServer implements HR
int percent = (int) (ratio * 100);
this.metrics.blockCacheHitRatio.set(percent);
}
-
+
+ L2BucketCache l2BucketCache = (L2BucketCache)cacheConfig.getL2Cache();
+ if (l2BucketCache != null) {
+ this.metrics.l2CacheCount.set(l2BucketCache.size());
+ this.metrics.l2CacheFree.set(l2BucketCache.getFreeSize());
+ this.metrics.l2CacheSize.set(l2BucketCache.getCurrentSize());
+ this.metrics.l2CacheEvictedCount.set(l2BucketCache.getEvictedCount());
+ CacheStats l2CacheStats = l2BucketCache.getStats();
+ this.metrics.l2CacheHitCount.set(l2CacheStats.getHitCount());
+ this.metrics.l2CacheMissCount.set(l2CacheStats.getMissCount());
+ double ratio = l2CacheStats.getIncrementalHitRatio();
+ int percent = (int) (ratio * 100);
+ this.metrics.l2CacheHitRatio.set(percent);
+ }
+
long bytesRead = 0;
long bytesLocalRead = 0;
long bytesRackLocalRead = 0;
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1492797&r1=1492796&r2=1492797&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Thu Jun 13 18:18:20 2013
@@ -252,8 +252,10 @@ public class Store extends SchemaConfigu
this.storeNameStr = getColumnFamilyName();
// Setting up cache configuration for this family
- this.cacheConf = new CacheConfig(conf, family);
-
+ this.cacheConf = new CacheConfig.CacheConfigBuilder(conf)
+ .withCacheDataOnRead(family.isBlockCacheEnabled())
+ .withInMemory(family.isInMemory())
+ .build();
// By default we split region if a file > HConstants.DEFAULT_MAX_FILE_SIZE.
long maxFileSize = info.getTableDesc().getMaxFileSize();
if (maxFileSize == HConstants.DEFAULT_MAX_FILE_SIZE) {
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java?rev=1492797&r1=1492796&r2=1492797&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java Thu Jun 13 18:18:20 2013
@@ -573,8 +573,13 @@ public class StoreFile extends SchemaCon
*/
public synchronized void closeReader(boolean evictOnClose)
throws IOException {
+ closeReader(evictOnClose, evictOnClose);
+ }
+
+ public synchronized void closeReader(boolean evictL1OnClose,
+ boolean evictL2OnClose) throws IOException {
if (this.reader != null) {
- this.reader.close(evictOnClose);
+ this.reader.close(evictL1OnClose, evictL2OnClose);
this.reader = null;
}
}
@@ -1339,6 +1344,11 @@ public class StoreFile extends SchemaCon
reader.close(evictOnClose);
}
+ public void close(boolean evictL1OnClose, boolean evictL2OnClose)
+ throws IOException {
+ reader.close(evictL1OnClose, evictL2OnClose);
+ }
+
public void close() throws IOException {
reader.close();
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerMetrics.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerMetrics.java?rev=1492797&r1=1492796&r2=1492797&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerMetrics.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerMetrics.java Thu Jun 13 18:18:20 2013
@@ -110,7 +110,50 @@ public class RegionServerMetrics impleme
/**
* Block hit ratio.
*/
- public final MetricsIntValue blockCacheHitRatio = new MetricsIntValue("blockCacheHitRatio", registry);
+ public final MetricsIntValue blockCacheHitRatio =
+ new MetricsIntValue("blockCacheHitRatio", registry);
+
+ /**
+ * L2 cache size
+ */
+ public final MetricsLongValue l2CacheSize =
+ new MetricsLongValue("l2CacheSize", registry);
+
+ /**
+ * L2 cache free size
+ */
+ public final MetricsLongValue l2CacheFree =
+ new MetricsLongValue("l2CacheFree", registry);
+
+ /**
+ * L2 cache item count
+ */
+ public final MetricsLongValue l2CacheCount =
+ new MetricsLongValue("l2CacheCount", registry);
+
+ /**
+ * L2 cache hit count
+ */
+ public final MetricsLongValue l2CacheHitCount =
+ new MetricsLongValue("l2CacheHitCount", registry);
+
+ /**
+ * L2 cache miss count
+ */
+ public final MetricsLongValue l2CacheMissCount =
+ new MetricsLongValue("l2CacheMissCount", registry);
+
+ /**
+ * L2 evicted item count
+ */
+ public final MetricsLongValue l2CacheEvictedCount =
+ new MetricsLongValue("l2CacheEvictedCount", registry);
+
+ /**
+ * L2 cache hit ratio
+ */
+ public final MetricsIntValue l2CacheHitRatio =
+ new MetricsIntValue("l2CacheHitRatio", registry);
/*
* Count of rows read or updated to the regionservers since last call to metrics update
@@ -372,6 +415,13 @@ public class RegionServerMetrics impleme
this.blockCacheEvictedMultiCount.pushMetric(this.metricsRecord);
this.blockCacheEvictedMemoryCount.pushMetric(this.metricsRecord);
this.blockCacheHitRatio.pushMetric(this.metricsRecord);
+ this.l2CacheSize.pushMetric(this.metricsRecord);
+ this.l2CacheFree.pushMetric(this.metricsRecord);
+ this.l2CacheCount.pushMetric(this.metricsRecord);
+ this.l2CacheHitCount.pushMetric(this.metricsRecord);
+ this.l2CacheMissCount.pushMetric(this.metricsRecord);
+ this.l2CacheEvictedCount.pushMetric(this.metricsRecord);
+ this.l2CacheHitRatio.pushMetric(this.metricsRecord);
this.rowReadCnt.pushMetric(this.metricsRecord);
this.rowUpdatedCnt.pushMetric(this.metricsRecord);
this.numOptimizedSeeks.pushMetric(this.metricsRecord);
@@ -598,7 +648,22 @@ public class RegionServerMetrics impleme
Long.valueOf(this.blockCacheEvictedMemoryCount.get()));
sb = Strings.appendKeyValue(sb, this.blockCacheHitRatio.getName(),
Long.valueOf(this.blockCacheHitRatio.get()));
-
+
+ sb = Strings.appendKeyValue(sb, this.l2CacheSize.getName(),
+ this.l2CacheSize.get());
+ sb = Strings.appendKeyValue(sb, this.l2CacheFree.getName(),
+ this.l2CacheFree.get());
+ sb = Strings.appendKeyValue(sb, this.l2CacheCount.getName(),
+ this.l2CacheCount.get());
+ sb = Strings.appendKeyValue(sb, this.l2CacheHitCount.getName(),
+ this.l2CacheHitCount.get());
+ sb = Strings.appendKeyValue(sb, this.l2CacheMissCount.getName(),
+ this.l2CacheMissCount.get());
+ sb = Strings.appendKeyValue(sb, this.l2CacheEvictedCount.getName(),
+ this.l2CacheEvictedCount.get());
+ sb = Strings.appendKeyValue(sb, this.l2CacheHitRatio.getName(),
+ this.l2CacheHitRatio.get());
+
sb = Strings.appendKeyValue(sb, this.bytesRead.getName(),
Long.valueOf(this.bytesRead.get()));
sb = Strings.appendKeyValue(sb, this.bytesLocalRead.getName(),