hbase-external-blockcache/src/main/java/org/apache/hadoop/hbase/io/hfile/MemcachedBlockCache.java - hbase - Git at Google

 /**
  * 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;

 import net.spy.memcached.CachedData;
 import net.spy.memcached.ConnectionFactoryBuilder;
 import net.spy.memcached.FailureMode;
 import net.spy.memcached.MemcachedClient;
 import net.spy.memcached.transcoders.Transcoder;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.util.Addressing;
 import org.apache.htrace.Trace;
 import org.apache.htrace.TraceScope;

 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.concurrent.ExecutionException;

 /**
  * Class to store blocks into memcached.
  * This should only be used on a cluster of Memcached daemons that are tuned well and have a
  * good network connection to the HBase regionservers. Any other use will likely slow down HBase
  * greatly.
  */
 @InterfaceAudience.Private
 public class MemcachedBlockCache implements BlockCache {
   private static final Log LOG = LogFactory.getLog(MemcachedBlockCache.class.getName());

   // Some memcache versions won't take more than 1024 * 1024. So set the limit below
   // that just in case this client is used with those versions.
   public static final int MAX_SIZE = 1020 * 1024;

   // Config key for what memcached servers to use.
   // They should be specified in a comma sperated list with ports.
   // like:
   //
   // host1:11211,host3:8080,host4:11211
   public static final String MEMCACHED_CONFIG_KEY = "hbase.cache.memcached.servers";
   public static final String MEMCACHED_TIMEOUT_KEY = "hbase.cache.memcached.timeout";
   public static final String MEMCACHED_OPTIMEOUT_KEY = "hbase.cache.memcached.optimeout";
   public static final String MEMCACHED_OPTIMIZE_KEY = "hbase.cache.memcached.spy.optimze";
   public static final long MEMCACHED_DEFAULT_TIMEOUT = 500;
   public static final boolean MEMCACHED_OPTIMIZE_DEFAULT = false;

   private final MemcachedClient client;
   private final HFileBlockTranscoder tc = new HFileBlockTranscoder();
   private final CacheStats cacheStats = new CacheStats("MemcachedBlockCache");

   public MemcachedBlockCache(Configuration c) throws IOException {
     LOG.info("Creating MemcachedBlockCache");

     long opTimeout = c.getLong(MEMCACHED_OPTIMEOUT_KEY, MEMCACHED_DEFAULT_TIMEOUT);
     long queueTimeout = c.getLong(MEMCACHED_TIMEOUT_KEY, opTimeout + MEMCACHED_DEFAULT_TIMEOUT);
     boolean optimize = c.getBoolean(MEMCACHED_OPTIMIZE_KEY, MEMCACHED_OPTIMIZE_DEFAULT);

     ConnectionFactoryBuilder builder = new ConnectionFactoryBuilder()
         .setOpTimeout(opTimeout)
         .setOpQueueMaxBlockTime(queueTimeout) // Cap the max time before anything times out
         .setFailureMode(FailureMode.Redistribute)
         .setShouldOptimize(optimize)
         .setDaemon(true)                      // Don't keep threads around past the end of days.
         .setUseNagleAlgorithm(false)          // Ain't nobody got time for that
         .setReadBufferSize(HConstants.DEFAULT_BLOCKSIZE * 4 * 1024); // Much larger just in case

     // Assume only the localhost is serving memecached.
     // A la mcrouter or co-locating memcached with split regionservers.
     //
     // If this config is a pool of memecached servers they will all be used according to the
     // default hashing scheme defined by the memcache client. Spy Memecache client in this
     // case.
     String serverListString = c.get(MEMCACHED_CONFIG_KEY,"localhost:11211");
     String[] servers = serverListString.split(",");
     List<InetSocketAddress> serverAddresses = new ArrayList<InetSocketAddress>(servers.length);
     for (String s:servers) {
       serverAddresses.add(Addressing.createInetSocketAddressFromHostAndPortStr(s));
     }

     client = new MemcachedClient(builder.build(), serverAddresses);
   }

   @Override
   public void cacheBlock(BlockCacheKey cacheKey,
                          Cacheable buf,
                          boolean inMemory,
                          boolean cacheDataInL1) {
     cacheBlock(cacheKey, buf);
   }

   @Override
   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
     if (buf instanceof HFileBlock) {
       client.add(cacheKey.toString(), MAX_SIZE, (HFileBlock) buf, tc);
     } else {
       if (LOG.isDebugEnabled()) {
         LOG.debug("MemcachedBlockCache can not cache Cacheable's of type "
             + buf.getClass().toString());
       }
     }
   }

   @Override
   public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching,
                             boolean repeat, boolean updateCacheMetrics) {
     // Assume that nothing is the block cache
     HFileBlock result = null;

     try (TraceScope traceScope = Trace.startSpan("MemcachedBlockCache.getBlock")) {
       result = client.get(cacheKey.toString(), tc);
     } catch (Exception e) {
       // Catch a pretty broad set of exceptions to limit any changes in the memecache client
       // and how it handles failures from leaking into the read path.
       if (LOG.isDebugEnabled()) {
         LOG.debug("Exception pulling from memcached [ "
             + cacheKey.toString()
             + " ]. Treating as a miss.", e);
       }
       result = null;
     } finally {
       // Update stats if this request doesn't have it turned off 100% of the time
       if (updateCacheMetrics) {
         if (result == null) {
           cacheStats.miss(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
         } else {
           cacheStats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
         }
       }
     }

     return result;
   }

   @Override
   public boolean evictBlock(BlockCacheKey cacheKey) {
     try {
       cacheStats.evict();
       return client.delete(cacheKey.toString()).get();
     } catch (InterruptedException e) {
       LOG.warn("Error deleting " + cacheKey.toString(), e);
       Thread.currentThread().interrupt();
     } catch (ExecutionException e) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Error deleting " + cacheKey.toString(), e);
       }
     }
     return false;
   }

   /**
    * This method does nothing so that memcached can handle all evictions.
    */
   @Override
   public int evictBlocksByHfileName(String hfileName) {
     return 0;
   }

   @Override
   public CacheStats getStats() {
     return cacheStats;
   }

   @Override
   public void shutdown() {
     client.shutdown();
   }

   @Override
   public long size() {
     return 0;
   }

   @Override
   public long getMaxSize() {
     return 0;
   }

   @Override
   public long getFreeSize() {
     return 0;
   }

   @Override
   public long getCurrentSize() {
     return 0;
   }

   @Override
   public long getCurrentDataSize() {
     return 0;
   }

   @Override
   public long getBlockCount() {
     return 0;
   }

   @Override
   public long getDataBlockCount() {
     return 0;
   }

   @Override
   public Iterator<CachedBlock> iterator() {
     return new Iterator<CachedBlock>() {
       @Override
       public boolean hasNext() {
         return false;
       }

       @Override
       public CachedBlock next() {
         throw new NoSuchElementException("MemcachedBlockCache can't iterate over blocks.");
       }

       @Override
       public void remove() {

       }
     };
   }

   @Override
   public BlockCache[] getBlockCaches() {
     return null;
   }

   /**
    * Class to encode and decode an HFileBlock to and from memecached's resulting byte arrays.
    */
   private static class HFileBlockTranscoder implements Transcoder<HFileBlock> {

     @Override
     public boolean asyncDecode(CachedData d) {
       return false;
     }

     @Override
     public CachedData encode(HFileBlock block) {
       ByteBuffer bb = ByteBuffer.allocate(block.getSerializedLength());
       block.serialize(bb, true);
       return new CachedData(0, bb.array(), CachedData.MAX_SIZE);
     }

     @Override
     public HFileBlock decode(CachedData d) {
       try {
         ByteBuffer buf = ByteBuffer.wrap(d.getData());
         return (HFileBlock) HFileBlock.BLOCK_DESERIALIZER.deserialize(buf, true);
       } catch (IOException e) {
         LOG.warn("Error deserializing data from memcached",e);
       }
       return null;
     }

     @Override
     public int getMaxSize() {
       return MAX_SIZE;
     }
   }

 }
	/**
	* 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;

	import net.spy.memcached.CachedData;
	import net.spy.memcached.ConnectionFactoryBuilder;
	import net.spy.memcached.FailureMode;
	import net.spy.memcached.MemcachedClient;
	import net.spy.memcached.transcoders.Transcoder;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.hbase.HConstants;
	import org.apache.hadoop.hbase.classification.InterfaceAudience;
	import org.apache.hadoop.hbase.util.Addressing;
	import org.apache.htrace.Trace;
	import org.apache.htrace.TraceScope;

	import java.io.IOException;
	import java.net.InetSocketAddress;
	import java.nio.ByteBuffer;
	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.List;
	import java.util.NoSuchElementException;
	import java.util.concurrent.ExecutionException;

	/**
	* Class to store blocks into memcached.
	* This should only be used on a cluster of Memcached daemons that are tuned well and have a
	* good network connection to the HBase regionservers. Any other use will likely slow down HBase
	* greatly.
	*/
	@InterfaceAudience.Private
	public class MemcachedBlockCache implements BlockCache {
	private static final Log LOG = LogFactory.getLog(MemcachedBlockCache.class.getName());

	// Some memcache versions won't take more than 1024 * 1024. So set the limit below
	// that just in case this client is used with those versions.
	public static final int MAX_SIZE = 1020 * 1024;

	// Config key for what memcached servers to use.
	// They should be specified in a comma sperated list with ports.
	// like:
	//
	// host1:11211,host3:8080,host4:11211
	public static final String MEMCACHED_CONFIG_KEY = "hbase.cache.memcached.servers";
	public static final String MEMCACHED_TIMEOUT_KEY = "hbase.cache.memcached.timeout";
	public static final String MEMCACHED_OPTIMEOUT_KEY = "hbase.cache.memcached.optimeout";
	public static final String MEMCACHED_OPTIMIZE_KEY = "hbase.cache.memcached.spy.optimze";
	public static final long MEMCACHED_DEFAULT_TIMEOUT = 500;
	public static final boolean MEMCACHED_OPTIMIZE_DEFAULT = false;

	private final MemcachedClient client;
	private final HFileBlockTranscoder tc = new HFileBlockTranscoder();
	private final CacheStats cacheStats = new CacheStats("MemcachedBlockCache");

	public MemcachedBlockCache(Configuration c) throws IOException {
	LOG.info("Creating MemcachedBlockCache");

	long opTimeout = c.getLong(MEMCACHED_OPTIMEOUT_KEY, MEMCACHED_DEFAULT_TIMEOUT);
	long queueTimeout = c.getLong(MEMCACHED_TIMEOUT_KEY, opTimeout + MEMCACHED_DEFAULT_TIMEOUT);
	boolean optimize = c.getBoolean(MEMCACHED_OPTIMIZE_KEY, MEMCACHED_OPTIMIZE_DEFAULT);

	ConnectionFactoryBuilder builder = new ConnectionFactoryBuilder()
	.setOpTimeout(opTimeout)
	.setOpQueueMaxBlockTime(queueTimeout) // Cap the max time before anything times out
	.setFailureMode(FailureMode.Redistribute)
	.setShouldOptimize(optimize)
	.setDaemon(true) // Don't keep threads around past the end of days.
	.setUseNagleAlgorithm(false) // Ain't nobody got time for that
	.setReadBufferSize(HConstants.DEFAULT_BLOCKSIZE * 4 * 1024); // Much larger just in case

	// Assume only the localhost is serving memecached.
	// A la mcrouter or co-locating memcached with split regionservers.
	//
	// If this config is a pool of memecached servers they will all be used according to the
	// default hashing scheme defined by the memcache client. Spy Memecache client in this
	// case.
	String serverListString = c.get(MEMCACHED_CONFIG_KEY,"localhost:11211");
	String[] servers = serverListString.split(",");
	List<InetSocketAddress> serverAddresses = new ArrayList<InetSocketAddress>(servers.length);
	for (String s:servers) {
	serverAddresses.add(Addressing.createInetSocketAddressFromHostAndPortStr(s));
	}

	client = new MemcachedClient(builder.build(), serverAddresses);
	}

	@Override
	public void cacheBlock(BlockCacheKey cacheKey,
	Cacheable buf,
	boolean inMemory,
	boolean cacheDataInL1) {
	cacheBlock(cacheKey, buf);
	}

	@Override
	public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
	if (buf instanceof HFileBlock) {
	client.add(cacheKey.toString(), MAX_SIZE, (HFileBlock) buf, tc);
	} else {
	if (LOG.isDebugEnabled()) {
	LOG.debug("MemcachedBlockCache can not cache Cacheable's of type "
	+ buf.getClass().toString());
	}
	}
	}

	@Override
	public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching,
	boolean repeat, boolean updateCacheMetrics) {
	// Assume that nothing is the block cache
	HFileBlock result = null;

	try (TraceScope traceScope = Trace.startSpan("MemcachedBlockCache.getBlock")) {
	result = client.get(cacheKey.toString(), tc);
	} catch (Exception e) {
	// Catch a pretty broad set of exceptions to limit any changes in the memecache client
	// and how it handles failures from leaking into the read path.
	if (LOG.isDebugEnabled()) {
	LOG.debug("Exception pulling from memcached [ "
	+ cacheKey.toString()
	+ " ]. Treating as a miss.", e);
	}
	result = null;
	} finally {
	// Update stats if this request doesn't have it turned off 100% of the time
	if (updateCacheMetrics) {
	if (result == null) {
	cacheStats.miss(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
	} else {
	cacheStats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
	}
	}
	}

	return result;
	}

	@Override
	public boolean evictBlock(BlockCacheKey cacheKey) {
	try {
	cacheStats.evict();
	return client.delete(cacheKey.toString()).get();
	} catch (InterruptedException e) {
	LOG.warn("Error deleting " + cacheKey.toString(), e);
	Thread.currentThread().interrupt();
	} catch (ExecutionException e) {
	if (LOG.isDebugEnabled()) {
	LOG.debug("Error deleting " + cacheKey.toString(), e);
	}
	}
	return false;
	}

	/**
	* This method does nothing so that memcached can handle all evictions.
	*/
	@Override
	public int evictBlocksByHfileName(String hfileName) {
	return 0;
	}

	@Override
	public CacheStats getStats() {
	return cacheStats;
	}

	@Override
	public void shutdown() {
	client.shutdown();
	}

	@Override
	public long size() {
	return 0;
	}

	@Override
	public long getMaxSize() {
	return 0;
	}

	@Override
	public long getFreeSize() {
	return 0;
	}

	@Override
	public long getCurrentSize() {
	return 0;
	}

	@Override
	public long getCurrentDataSize() {
	return 0;
	}

	@Override
	public long getBlockCount() {
	return 0;
	}

	@Override
	public long getDataBlockCount() {
	return 0;
	}

	@Override
	public Iterator<CachedBlock> iterator() {
	return new Iterator<CachedBlock>() {
	@Override
	public boolean hasNext() {
	return false;
	}

	@Override
	public CachedBlock next() {
	throw new NoSuchElementException("MemcachedBlockCache can't iterate over blocks.");
	}

	@Override
	public void remove() {

	}
	};
	}

	@Override
	public BlockCache[] getBlockCaches() {
	return null;
	}

	/**
	* Class to encode and decode an HFileBlock to and from memecached's resulting byte arrays.
	*/
	private static class HFileBlockTranscoder implements Transcoder<HFileBlock> {

	@Override
	public boolean asyncDecode(CachedData d) {
	return false;
	}

	@Override
	public CachedData encode(HFileBlock block) {
	ByteBuffer bb = ByteBuffer.allocate(block.getSerializedLength());
	block.serialize(bb, true);
	return new CachedData(0, bb.array(), CachedData.MAX_SIZE);
	}

	@Override
	public HFileBlock decode(CachedData d) {
	try {
	ByteBuffer buf = ByteBuffer.wrap(d.getData());
	return (HFileBlock) HFileBlock.BLOCK_DESERIALIZER.deserialize(buf, true);
	} catch (IOException e) {
	LOG.warn("Error deserializing data from memcached",e);
	}
	return null;
	}

	@Override
	public int getMaxSize() {
	return MAX_SIZE;
	}
	}

	}