*/
-package com.meetup.memcached;
+package edu.usc.cs550.rejig.client;
+
+import edu.usc.cs550.rejig.client.configreader.RejigConfigReader;
+import edu.usc.cs550.rejig.interfaces.Fragment;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.*;
-import java.nio.*;
+import java.nio.*;
import java.net.InetAddress;
-import java.nio.charset.*;
+import java.nio.charset.*;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.io.*;
@@ -32,7 +37,7 @@
/**
* This is a Memcached client for the Java platform available from
* http://www.danga.com/memcached/.
- *
+ *
* Supports setting, adding, replacing, deleting compressed/uncompressed and
* serialized (can be stored as string if object is native class) objects to memcached.
*
@@ -42,116 +47,117 @@
*
* Some examples of use follow.
* To create cache client object and set params:
- *
+ *
* MemcachedClient mc = new MemcachedClient();
*
- * // compression is enabled by default
+ * // compression is enabled by default
* mc.setCompressEnable(true);
*
- * // set compression threshhold to 4 KB (default: 15 KB)
+ * // set compression threshhold to 4 KB (default: 15 KB)
* mc.setCompressThreshold(4096);
*
* // turn on storing primitive types as a string representation
- * // Should not do this in most cases.
+ * // Should not do this in most cases.
* mc.setPrimitiveAsString(true);
- *
+ *
* To store an object:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "cacheKey1";
- * Object value = SomeClass.getObject();
+ * String key = "cacheKey1";
+ * Object value = SomeClass.getObject();
* mc.set(key, value);
- *
+ *
* To store an object using a custom server hashCode:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "cacheKey1";
- * Object value = SomeClass.getObject();
- * Integer hash = new Integer(45);
+ * String key = "cacheKey1";
+ * Object value = SomeClass.getObject();
+ * Integer hash = new Integer(45);
* mc.set(key, value, hash);
- *
+ *
* The set method shown above will always set the object in the cache.
* The add and replace methods do the same, but with a slight difference.
*
* - add -- will store the object only if the server does not have an entry for this key
* - replace -- will store the object only if the server already has an entry for this key
- *
+ *
* To delete a cache entry:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "cacheKey1";
+ * String key = "cacheKey1";
* mc.delete(key);
- *
+ *
* To delete a cache entry using a custom hash code:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "cacheKey1";
- * Integer hash = new Integer(45);
+ * String key = "cacheKey1";
+ * Integer hash = new Integer(45);
* mc.delete(key, hashCode);
- *
+ *
* To store a counter and then increment or decrement that counter:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "counterKey";
+ * String key = "counterKey";
* mc.storeCounter(key, new Integer(100));
- * System.out.println("counter after adding 1: " mc.incr(key));
- * System.out.println("counter after adding 5: " mc.incr(key, 5));
- * System.out.println("counter after subtracting 4: " mc.decr(key, 4));
- * System.out.println("counter after subtracting 1: " mc.decr(key));
- *
+ * System.out.println("counter after adding 1: " mc.incr(key));
+ * System.out.println("counter after adding 5: " mc.incr(key, 5));
+ * System.out.println("counter after subtracting 4: " mc.decr(key, 4));
+ * System.out.println("counter after subtracting 1: " mc.decr(key));
+ *
* To store a counter and then increment or decrement that counter with custom hash:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "counterKey";
- * Integer hash = new Integer(45);
+ * String key = "counterKey";
+ * Integer hash = new Integer(45);
* mc.storeCounter(key, new Integer(100), hash);
- * System.out.println("counter after adding 1: " mc.incr(key, 1, hash));
- * System.out.println("counter after adding 5: " mc.incr(key, 5, hash));
- * System.out.println("counter after subtracting 4: " mc.decr(key, 4, hash));
- * System.out.println("counter after subtracting 1: " mc.decr(key, 1, hash));
- *
+ * System.out.println("counter after adding 1: " mc.incr(key, 1, hash));
+ * System.out.println("counter after adding 5: " mc.incr(key, 5, hash));
+ * System.out.println("counter after subtracting 4: " mc.decr(key, 4, hash));
+ * System.out.println("counter after subtracting 1: " mc.decr(key, 1, hash));
+ *
* To retrieve an object from the cache:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "key";
- * Object value = mc.get(key);
- *
+ * String key = "key";
+ * Object value = mc.get(key);
+ *
* To retrieve an object from the cache with custom hash:
*
* MemcachedClient mc = new MemcachedClient();
- * String key = "key";
- * Integer hash = new Integer(45);
- * Object value = mc.get(key, hash);
- *
+ * String key = "key";
+ * Integer hash = new Integer(45);
+ * Object value = mc.get(key, hash);
+ *
* To retrieve an multiple objects from the cache
*
* MemcachedClient mc = new MemcachedClient();
* String[] keys = { "key", "key1", "key2" };
* Map<Object> values = mc.getMulti(keys);
- *
+ *
* To retrieve an multiple objects from the cache with custom hashing
*
* MemcachedClient mc = new MemcachedClient();
* String[] keys = { "key", "key1", "key2" };
* Integer[] hashes = { new Integer(45), new Integer(32), new Integer(44) };
* Map<Object> values = mc.getMulti(keys, hashes);
- *
+ *
* To flush all items in server(s)
*
* MemcachedClient mc = new MemcachedClient();
* mc.flushAll();
- *
+ *
* To get stats from server(s)
*
* MemcachedClient mc = new MemcachedClient();
* Map stats = mc.stats();
- *
+ *
*
- * @author greg whalin
+ * @author greg whalin
* @author Richard 'toast' Russo
* @author Kevin Burton
* @author Robert Watts
* @author Vin Chawla
+ * @author Likhit Dharmapuri
* @version 1.5
*/
public class MemcachedClient {
@@ -170,6 +176,10 @@ public class MemcachedClient {
private static final String NOTSTORED = "NOT_STORED"; // data not stored
private static final String OK = "OK"; // success
private static final String END = "END"; // end of data from server
+ private static final String REFRESH_AND_RETRY = "REFRESH_AND_RETRY"; // client's config stale.
+ private static final String GRANTED = "GRANTED"; // lease granted
+ private static final String REVOKED = "REVOKED"; // lease revoked
+ private static final String REJIG_CONFIG_STORAGE_KEY = "REJIG_CONFIG_STORAGE_KEY"; // The identifier returned in the value line when REFRESH_AND_RETRY error occurs.
private static final String ERROR = "ERROR"; // invalid command name from client
private static final String CLIENT_ERROR = "CLIENT_ERROR"; // client error in input line - invalid protocol
@@ -182,8 +192,8 @@ public class MemcachedClient {
// default compression threshold
private static final int COMPRESS_THRESH = 30720;
-
- // values for cache flags
+
+ // values for cache flags
public static final int MARKER_BYTE = 1;
public static final int MARKER_BOOLEAN = 8192;
public static final int MARKER_INTEGER = 4;
@@ -199,7 +209,7 @@ public class MemcachedClient {
public static final int MARKER_BYTEARR = 4096;
public static final int F_COMPRESSED = 2;
public static final int F_SERIALIZED = 8;
-
+
// flags
private boolean sanitizeKeys;
private boolean primitiveAsString;
@@ -208,10 +218,13 @@ public class MemcachedClient {
private String defaultEncoding;
// pool instance
- private SockIOPool pool;
+ private AtomicReference currentPool;
+
+ // sockiopool initialization options.
+ SockIOPool.SockIOPoolOptions poolOptions;
// which pool to use
- private String poolName;
+ private String poolNamePrefix;
// optional passed in classloader
private ClassLoader classLoader;
@@ -219,28 +232,41 @@ public class MemcachedClient {
// optional error handler
private ErrorHandler errorHandler;
+ // config reader
+ RejigConfigReader configReader;
+
/**
* Creates a new instance of MemCachedClient.
+ * @param reader The reader to use to accquire and update the RejigConfig object used to create the SockIOPool.
+ * @param options The options on the SockIOPool maintained by this client.
*/
- public MemcachedClient() {
+ public MemcachedClient( RejigConfigReader reader, SockIOPool.SockIOPoolOptions options ) {
+ this.poolNamePrefix = "default";
+ this.configReader = reader;
+ this.poolOptions = options;
init();
}
- /**
+ /**
* Creates a new instance of MemCachedClient
- * accepting a passed in pool name.
- *
- * @param poolName name of SockIOPool
- */
- public MemcachedClient( String poolName ) {
- this.poolName = poolName;
+ * with the specified pool name as a prefix appended
+ * to the name of the SockIOPool created.
+ *
+ * @param poolNamePrefix prefix to name of SockIOPool
+ * @param reader The reader to use to accquire and update the RejigConfig object used to create the SockIOPool.
+ * @param options The options on the SockIOPool maintained by this client.
+ */
+ public MemcachedClient( String poolNamePrefix, RejigConfigReader reader, SockIOPool.SockIOPoolOptions options ) {
+ this.poolNamePrefix = poolNamePrefix;
+ this.configReader = reader;
+ this.poolOptions = options;
init();
}
- /**
+ /**
* Creates a new instance of MemCacheClient but
* acceptes a passed in ClassLoader.
- *
+ *
* @param classLoader ClassLoader object.
*/
public MemcachedClient( ClassLoader classLoader ) {
@@ -248,11 +274,11 @@ public MemcachedClient( ClassLoader classLoader ) {
init();
}
- /**
+ /**
* Creates a new instance of MemCacheClient but
* acceptes a passed in ClassLoader and a passed
* in ErrorHandler.
- *
+ *
* @param classLoader ClassLoader object.
* @param errorHandler ErrorHandler object.
*/
@@ -262,23 +288,27 @@ public MemcachedClient( ClassLoader classLoader, ErrorHandler errorHandler ) {
init();
}
- /**
+ /**
* Creates a new instance of MemCacheClient but
* acceptes a passed in ClassLoader, ErrorHandler,
- * and SockIOPool name.
- *
+ * and SockIOPool name prefix.
+ *
* @param classLoader ClassLoader object.
* @param errorHandler ErrorHandler object.
- * @param poolName SockIOPool name
+ * @param poolNamePrefix SockIOPool name's prefix.
+ * @param reader The reader to use to accquire and update the RejigConfig object used to create the SockIOPool.
+ * @param options The options on the SockIOPool maintained by this client.
*/
- public MemcachedClient( ClassLoader classLoader, ErrorHandler errorHandler, String poolName ) {
+ public MemcachedClient( ClassLoader classLoader, ErrorHandler errorHandler, String poolNamePrefix, RejigConfigReader reader, SockIOPool.SockIOPoolOptions options ) {
this.classLoader = classLoader;
this.errorHandler = errorHandler;
- this.poolName = poolName;
+ this.poolNamePrefix = poolNamePrefix;
+ this.configReader = reader;
+ this.poolOptions = options;
init();
}
- /**
+ /**
* Initializes client object to defaults.
*
* This enables compression and sets compression threshhold to 15 KB.
@@ -289,54 +319,95 @@ private void init() {
this.compressEnable = true;
this.compressThreshold = COMPRESS_THRESH;
this.defaultEncoding = "UTF-8";
- this.poolName = ( this.poolName == null ) ? "default" : this.poolName;
+ RejigConfig config = this.configReader.getConfig();
+ currentPool = new AtomicReference(createSockIOPool(config));
+ }
+
+ /**
+ * Create and initialize a new SockIOPool for the given
+ * config, and make the client use the new pool.
+ */
+ private SockIOPool createSockIOPool(RejigConfig config) {
+ String poolName = String.format("%s-%d", this.poolNamePrefix, config.getId());
+ SockIOPool pool = SockIOPool.getInstance(poolName)
+ .setRejigConfig(config)
+ .setPoolOptions(poolOptions)
+ .initialize();
+ return pool;
+ }
- // get a pool instance to work with for the life of this instance
- this.pool = SockIOPool.getInstance( poolName );
+ /**
+ * Atomically sets the current SockIOPool to the new value.
+ * If the set was successful, it shuts down the old pool.
+ * If the set failed, it retries infinitely with
+ * exponential backoff.
+ */
+ private void compareAndSetSockIOPool(SockIOPool oldPool, SockIOPool newPool) {
+ long sleepMillis = 2000;
+ while (true) {
+ boolean ret = currentPool.compareAndSet(oldPool, newPool);
+ if (ret) {
+ oldPool.shutDown();
+ break;
+ }
+ try {
+ Thread.sleep(sleepMillis);
+ sleepMillis *= 2;
+ } catch (InterruptedException e) {
+ log.warn("++++ Backoff sleep while setting SockIOPool interrupted.");
+ }
+ }
}
- /**
+ /**
+ * Returns the SockIOPool currently being used by the client.
+ */
+ public SockIOPool getSockIOPool() {
+ return currentPool.get();
+ }
+
+ /**
* Sets an optional ClassLoader to be used for
* serialization.
- *
- * @param classLoader
+ *
+ * @param classLoader
*/
public void setClassLoader( ClassLoader classLoader ) {
this.classLoader = classLoader;
}
- /**
+ /**
* Sets an optional ErrorHandler.
- *
- * @param errorHandler
+ *
+ * @param errorHandler
*/
public void setErrorHandler( ErrorHandler errorHandler ) {
this.errorHandler = errorHandler;
}
- /**
+ /**
* Enables/disables sanitizing keys by URLEncoding.
- *
+ *
* @param sanitizeKeys if true, then URLEncode all keys
*/
public void setSanitizeKeys( boolean sanitizeKeys ) {
this.sanitizeKeys = sanitizeKeys;
}
- /**
- * Enables storing primitive types as their String values.
- *
+ /**
+ * Enables storing primitive types as their String values.
+ *
* @param primitiveAsString if true, then store all primitives as their string value.
*/
public void setPrimitiveAsString( boolean primitiveAsString ) {
this.primitiveAsString = primitiveAsString;
}
- /**
- * Sets default String encoding when storing primitives as Strings.
+ /**
+ * Sets default String encoding when storing primitives as Strings.
* Default is UTF-8.
- *
- * @param defaultEncoding
+ *
+ * @param defaultEncoding
*/
public void setDefaultEncoding( String defaultEncoding ) {
this.defaultEncoding = defaultEncoding;
@@ -358,7 +429,7 @@ public void setDefaultEncoding( String defaultEncoding ) {
public void setCompressEnable( boolean compressEnable ) {
this.compressEnable = compressEnable;
}
-
+
/**
* Sets the required length for data to be considered for compression.
*
@@ -373,9 +444,9 @@ public void setCompressThreshold( long compressThreshold ) {
this.compressThreshold = compressThreshold;
}
- /**
- * Checks to see if key exists in cache.
- *
+ /**
+ * Checks to see if key exists in cache.
+ *
* @param key the key to look for
* @return true if key found in cache, false if not (or if cache is down)
*/
@@ -393,9 +464,9 @@ public boolean delete( String key ) {
return delete( key, null, null );
}
- /**
- * Deletes an object from cache given cache key and expiration date.
- *
+ /**
+ * Deletes an object from cache given cache key and expiration date.
+ *
* @param key the key to be removed
* @param expiry when to expire the record.
* @return true, if the data was deleted successfully
@@ -418,8 +489,12 @@ public boolean delete( String key, Date expiry ) {
* @param expiry when to expire the record.
* @return true, if the data was deleted successfully
*/
- public boolean delete( String key, Integer hashCode, Date expiry ) {
+ public boolean delete(
+ final String originalKey,
+ final Integer hashCode,
+ final Date expiry ) {
+ String key = originalKey;
if ( key == null ) {
log.error( "null value for key passed to delete()" );
return false;
@@ -439,29 +514,41 @@ public boolean delete( String key, Integer hashCode, Date expiry ) {
}
// get SockIO obj from hash or from key
- SockIOPool.SockIO sock = pool.getSock( key, hashCode );
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getSockAndFragmentId( key, hashCode );
// return false if unable to get SockIO obj
- if ( sock == null ) {
+ if ( sockAndId == null || sockAndId.sock() == null ) {
if ( errorHandler != null )
errorHandler.handleErrorOnDelete( this, new IOException( "no socket to server available" ), key );
return false;
}
+ SockIOPool.SockIO sock = sockAndId.sock();
// build command
- StringBuilder command = new StringBuilder( "delete " ).append( key );
+ StringBuilder command = new StringBuilder( "rj " )
+ .append( client_config_id )
+ .append( " " )
+ .append( sockAndId.fragmentNum() )
+ .append( " delete " )
+ .append( key );
if ( expiry != null )
command.append( " " + expiry.getTime() / 1000 );
command.append( "\r\n" );
-
+
try {
sock.write( command.toString().getBytes() );
sock.flush();
-
+
// if we get appropriate response back, then we return true
String line = sock.readLine();
- if ( DELETED.equals( line ) ) {
+ if ( REFRESH_AND_RETRY.equals(line) ) {
+ handleRefreshAndRetry(pool, sock);
+ return delete(originalKey, hashCode, expiry);
+ }
+ else if ( DELETED.equals( line ) ) {
if ( log.isInfoEnabled() )
log.info( "++++ deletion of key: " + key + " from cache was a success" );
@@ -506,7 +593,7 @@ else if ( NOTFOUND.equals( line ) ) {
return false;
}
-
+
/**
* Stores data on the server; only the key and the value are specified.
*
@@ -651,7 +738,7 @@ public boolean replace( String key, Object value, Date expiry, Integer hashCode
return set( "replace", key, value, expiry, hashCode, primitiveAsString );
}
- /**
+ /**
* Stores data to cache.
*
* If data does not already exist for this key on the server, or if the key is being
@@ -662,7 +749,7 @@ public boolean replace( String key, Object value, Date expiry, Integer hashCode
* the data will be stored in compressed form.
*
* As of the current release, all objects stored will use java serialization.
- *
+ *
* @param cmdname action to take (set, add, replace)
* @param key key to store cache under
* @param value object to cache
@@ -671,8 +758,15 @@ public boolean replace( String key, Object value, Date expiry, Integer hashCode
* @param asString store this object as a string?
* @return true/false indicating success
*/
- private boolean set( String cmdname, String key, Object value, Date expiry, Integer hashCode, boolean asString ) {
+ private boolean set(
+ final String cmdname,
+ final String originalKey,
+ final Object value,
+ final Date originalExpiry,
+ final Integer hashCode,
+ final boolean asString ) {
+ String key = originalKey;
if ( cmdname == null || cmdname.trim().equals( "" ) || key == null ) {
log.error( "key is null or cmd is null/empty for set()" );
return false;
@@ -697,25 +791,29 @@ private boolean set( String cmdname, String key, Object value, Date expiry, Inte
}
// get SockIO obj
- SockIOPool.SockIO sock = pool.getSock( key, hashCode );
-
- if ( sock == null ) {
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getSockAndFragmentId( key, hashCode );
+
+ if ( sockAndId == null || sockAndId.sock() == null ) {
if ( errorHandler != null )
errorHandler.handleErrorOnSet( this, new IOException( "no socket to server available" ), key );
return false;
}
-
+ SockIOPool.SockIO sock = sockAndId.sock();
+
+ Date expiry = originalExpiry;
if ( expiry == null )
expiry = new Date(0);
// store flags
int flags = 0;
-
+
// byte array to hold data
byte[] val;
if ( NativeHandler.isHandled( value ) ) {
-
+
if ( asString ) {
// useful for sharing data between java and non-java
// and also for storing ints for the increment method
@@ -784,9 +882,9 @@ private boolean set( String cmdname, String key, Object value, Date expiry, Inte
return false;
}
}
-
+
// now try to compress if we want to
- // and if the length is over the threshold
+ // and if the length is over the threshold
if ( compressEnable && val.length > compressThreshold ) {
try {
@@ -799,7 +897,7 @@ private boolean set( String cmdname, String key, Object value, Date expiry, Inte
gos.write( val, 0, val.length );
gos.finish();
gos.close();
-
+
// store it and set compression flag
val = bos.toByteArray();
flags |= F_COMPRESSED;
@@ -820,7 +918,7 @@ private boolean set( String cmdname, String key, Object value, Date expiry, Inte
// now write the data to the cache server
try {
- String cmd = String.format( "%s %s %d %d %d\r\n", cmdname, key, flags, (expiry.getTime() / 1000), val.length );
+ String cmd = String.format( "rj %d %d %s %s %d %d %d\r\n", client_config_id, sockAndId.fragmentNum(), cmdname, key, flags, (expiry.getTime() / 1000), val.length );
sock.write( cmd.getBytes() );
sock.write( val );
sock.write( "\r\n".getBytes() );
@@ -831,7 +929,11 @@ private boolean set( String cmdname, String key, Object value, Date expiry, Inte
if ( log.isInfoEnabled() )
log.info( "++++ memcache cmd (result code): " + cmd + " (" + line + ")" );
- if ( STORED.equals( line ) ) {
+ if ( REFRESH_AND_RETRY.equals( line ) ) {
+ handleRefreshAndRetry(pool, sock);
+ return set(cmdname, originalKey, value, originalExpiry, hashCode, asString);
+ }
+ else if ( STORED.equals( line ) ) {
if ( log.isInfoEnabled() )
log.info("++++ data successfully stored for key: " + key );
sock.close();
@@ -875,9 +977,9 @@ else if ( NOTSTORED.equals( line ) ) {
return false;
}
- /**
+ /**
* Store a counter to memcached given a key
- *
+ *
* @param key cache key
* @param counter number to store
* @return true/false indicating success
@@ -886,9 +988,9 @@ public boolean storeCounter( String key, long counter ) {
return set( "set", key, new Long( counter ), null, null, true );
}
- /**
+ /**
* Store a counter to memcached given a key
- *
+ *
* @param key cache key
* @param counter number to store
* @return true/false indicating success
@@ -896,10 +998,10 @@ public boolean storeCounter( String key, long counter ) {
public boolean storeCounter( String key, Long counter ) {
return set( "set", key, counter, null, null, true );
}
-
- /**
+
+ /**
* Store a counter to memcached given a key
- *
+ *
* @param key cache key
* @param counter number to store
* @param hashCode if not null, then the int hashcode to use
@@ -909,8 +1011,8 @@ public boolean storeCounter( String key, Long counter, Integer hashCode ) {
return set( "set", key, counter, null, hashCode, true );
}
- /**
- * Returns value in counter at given key as long.
+ /**
+ * Returns value in counter at given key as long.
*
* @param key cache ket
* @return counter value or -1 if not found
@@ -919,8 +1021,8 @@ public long getCounter( String key ) {
return getCounter( key, null );
}
- /**
- * Returns value in counter at given key as long.
+ /**
+ * Returns value in counter at given key as long.
*
* @param key cache ket
* @param hashCode if not null, then the int hashcode to use
@@ -947,13 +1049,13 @@ public long getCounter( String key, Integer hashCode ) {
if ( log.isInfoEnabled() )
log.info( String.format( "Failed to parse Long value for key: %s", key ) );
}
-
+
return counter;
}
- /**
- * Thread safe way to initialize and increment a counter.
- *
+ /**
+ * Thread safe way to initialize and increment a counter.
+ *
* @param key key where the data is stored
* @return value of incrementer
*/
@@ -961,9 +1063,9 @@ public long addOrIncr( String key ) {
return addOrIncr( key, 0, null );
}
- /**
- * Thread safe way to initialize and increment a counter.
- *
+ /**
+ * Thread safe way to initialize and increment a counter.
+ *
* @param key key where the data is stored
* @param inc value to set or increment by
* @return value of incrementer
@@ -972,9 +1074,9 @@ public long addOrIncr( String key, long inc ) {
return addOrIncr( key, inc, null );
}
- /**
- * Thread safe way to initialize and increment a counter.
- *
+ /**
+ * Thread safe way to initialize and increment a counter.
+ *
* @param key key where the data is stored
* @param inc value to set or increment by
* @param hashCode if not null, then the int hashcode to use
@@ -991,9 +1093,9 @@ public long addOrIncr( String key, long inc, Integer hashCode ) {
}
}
- /**
- * Thread safe way to initialize and decrement a counter.
- *
+ /**
+ * Thread safe way to initialize and decrement a counter.
+ *
* @param key key where the data is stored
* @return value of incrementer
*/
@@ -1001,9 +1103,9 @@ public long addOrDecr( String key ) {
return addOrDecr( key, 0, null );
}
- /**
- * Thread safe way to initialize and decrement a counter.
- *
+ /**
+ * Thread safe way to initialize and decrement a counter.
+ *
* @param key key where the data is stored
* @param inc value to set or increment by
* @return value of incrementer
@@ -1012,9 +1114,9 @@ public long addOrDecr( String key, long inc ) {
return addOrDecr( key, inc, null );
}
- /**
- * Thread safe way to initialize and decrement a counter.
- *
+ /**
+ * Thread safe way to initialize and decrement a counter.
+ *
* @param key key where the data is stored
* @param inc value to set or increment by
* @param hashCode if not null, then the int hashcode to use
@@ -1041,9 +1143,9 @@ public long incr( String key ) {
return incrdecr( "incr", key, 1, null );
}
- /**
- * Increment the value at the specified key by passed in val.
- *
+ /**
+ * Increment the value at the specified key by passed in val.
+ *
* @param key key where the data is stored
* @param inc how much to increment by
* @return -1, if the key is not found, the value after incrementing otherwise
@@ -1063,7 +1165,7 @@ public long incr( String key, long inc ) {
public long incr( String key, long inc, Integer hashCode ) {
return incrdecr( "incr", key, inc, hashCode );
}
-
+
/**
* Decrement the value at the specified key by 1, and then return it.
*
@@ -1097,9 +1199,9 @@ public long decr( String key, long inc, Integer hashCode ) {
return incrdecr( "decr", key, inc, hashCode );
}
- /**
+ /**
* Increments/decrements the value at the specified key by inc.
- *
+ *
* Note that the server uses a 32-bit unsigned integer, and checks for
* underflow. In the event of underflow, the result will be zero. Because
* Java lacks unsigned types, the value is returned as a 64-bit integer.
@@ -1112,8 +1214,13 @@ public long decr( String key, long inc, Integer hashCode ) {
* @param hashCode if not null, then the int hashcode to use
* @return new value or -1 if not exist
*/
- private long incrdecr( String cmdname, String key, long inc, Integer hashCode ) {
+ private long incrdecr(
+ final String cmdname,
+ final String originalKey,
+ final long inc,
+ final Integer hashCode ) {
+ String key = originalKey;
if ( key == null ) {
log.error( "null key for incrdecr()" );
return -1;
@@ -1133,16 +1240,19 @@ private long incrdecr( String cmdname, String key, long inc, Integer hashCode )
}
// get SockIO obj for given cache key
- SockIOPool.SockIO sock = pool.getSock( key, hashCode );
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getSockAndFragmentId( key, hashCode );
- if ( sock == null ) {
+ if ( sockAndId == null || sockAndId.sock() == null ) {
if ( errorHandler != null )
errorHandler.handleErrorOnSet( this, new IOException( "no socket to server available" ), key );
return -1;
}
-
+ SockIOPool.SockIO sock = sockAndId.sock();
+
try {
- String cmd = String.format( "%s %s %d\r\n", cmdname, key, inc );
+ String cmd = String.format( "rj %d %d %s %s %d\r\n", client_config_id, sockAndId.fragmentNum(), cmdname, key, inc );
if ( log.isDebugEnabled() )
log.debug( "++++ memcache incr/decr command: " + cmd );
@@ -1152,7 +1262,11 @@ private long incrdecr( String cmdname, String key, long inc, Integer hashCode )
// get result back
String line = sock.readLine();
- if ( line.matches( "\\d+" ) ) {
+ if ( REFRESH_AND_RETRY.equals( line ) ) {
+ handleRefreshAndRetry(pool, sock);
+ return incrdecr(cmdname, originalKey, inc, hashCode);
+ }
+ else if ( line.matches( "\\d+" ) ) {
// return sock to pool and return result
sock.close();
@@ -1196,7 +1310,7 @@ else if ( NOTFOUND.equals( line ) ) {
sock = null;
}
-
+
if ( sock != null ) {
sock.close();
sock = null;
@@ -1221,7 +1335,7 @@ public Object get( String key ) {
return get( key, null, false );
}
- /**
+ /**
* Retrieve a key from the server, using a specific hash.
*
* If the data was compressed or serialized when compressed, it will automatically
@@ -1252,8 +1366,12 @@ public Object get( String key, Integer hashCode ) {
* @param asString if true, then return string val
* @return the object that was previously stored, or null if it was not previously stored
*/
- public Object get( String key, Integer hashCode, boolean asString ) {
+ public Object get(
+ final String originalKey,
+ final Integer hashCode,
+ final boolean asString ) {
+ String key = originalKey;
if ( key == null ) {
log.error( "key is null for get()" );
return null;
@@ -1273,20 +1391,22 @@ public Object get( String key, Integer hashCode, boolean asString ) {
}
// get SockIO obj using cache key
- SockIOPool.SockIO sock = pool.getSock( key, hashCode );
-
- if ( sock == null ) {
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getSockAndFragmentId( key, hashCode );
+
+ if ( sockAndId == null || sockAndId.sock() == null ) {
if ( errorHandler != null )
errorHandler.handleErrorOnGet( this, new IOException( "no socket to server available" ), key );
return null;
}
+ SockIOPool.SockIO sock = sockAndId.sock();
try {
- String cmd = "get " + key + "\r\n";
-
+ String cmd = String.format("rj %d %d get %s\r\n", client_config_id, sockAndId.fragmentNum(), key);
if ( log.isDebugEnabled() )
log.debug("++++ memcache get command: " + cmd);
-
+
sock.write( cmd.getBytes() );
sock.flush();
@@ -1299,17 +1419,32 @@ public Object get( String key, Integer hashCode, boolean asString ) {
if ( log.isDebugEnabled() )
log.debug( "++++ line: " + line );
- if ( line.startsWith( VALUE ) ) {
+ if ( REFRESH_AND_RETRY.equals(line) ) {
+ handleRefreshAndRetry(pool, sock);
+ return get(originalKey, hashCode, asString);
+ }
+ else if ( line.startsWith( VALUE ) ) {
String[] info = line.split(" ");
int flag = Integer.parseInt( info[2] );
int length = Integer.parseInt( info[3] );
+ int key_config_id = Integer.parseInt( info[4] );
if ( log.isDebugEnabled() ) {
log.debug( "++++ key: " + key );
log.debug( "++++ flags: " + flag );
log.debug( "++++ length: " + length );
+ log.debug( "++++ key config id: " + key_config_id );
}
-
+
+ // If fragment id is greater than the entry's config id
+ // it means that the entry is stale.
+ if ( sockAndId.fragmentId() > key_config_id ) {
+ sock.close();
+ sock = null;
+ delete( key );
+ return null;
+ }
+
// read obj into buffer
byte[] buf = new byte[length];
sock.read( buf );
@@ -1322,7 +1457,7 @@ public Object get( String key, Integer hashCode, boolean asString ) {
// will need to be, and we don't want to resize it a bunch
GZIPInputStream gzi = new GZIPInputStream( new ByteArrayInputStream( buf ) );
ByteArrayOutputStream bos = new ByteArrayOutputStream( buf.length );
-
+
int count;
byte[] tmp = new byte[2048];
while ( (count = gzi.read(tmp)) != -1 ) {
@@ -1355,7 +1490,7 @@ public Object get( String key, Integer hashCode, boolean asString ) {
else {
// decoding object
try {
- o = NativeHandler.decode( buf, flag );
+ o = NativeHandler.decode( buf, flag );
}
catch ( Exception e ) {
@@ -1392,11 +1527,11 @@ else if ( END.equals( line ) ) {
break;
}
}
-
+
sock.close();
sock = null;
return o;
- }
+ }
catch ( IOException e ) {
// if we have an errorHandler, use its hook
@@ -1421,7 +1556,7 @@ else if ( END.equals( line ) ) {
return null;
}
- /**
+ /**
* Retrieve multiple objects from the memcache.
*
* This is recommended over repeated calls to {@link #get(String) get()}, since it
@@ -1434,7 +1569,7 @@ public Object[] getMultiArray( String[] keys ) {
return getMultiArray( keys, null, false );
}
- /**
+ /**
* Retrieve multiple objects from the memcache.
*
* This is recommended over repeated calls to {@link #get(String) get()}, since it
@@ -1448,7 +1583,7 @@ public Object[] getMultiArray( String[] keys, Integer[] hashCodes ) {
return getMultiArray( keys, hashCodes, false );
}
- /**
+ /**
* Retrieve multiple objects from the memcache.
*
* This is recommended over repeated calls to {@link #get(String) get()}, since it
@@ -1488,7 +1623,7 @@ public Object[] getMultiArray( String[] keys, Integer[] hashCodes, boolean asStr
public Map getMulti( String[] keys ) {
return getMulti( keys, null, false );
}
-
+
/**
* Retrieve multiple keys from the memcache.
*
@@ -1555,7 +1690,8 @@ public Map getMulti( String[] keys, Integer[] hashCodes, boolean
}
// get SockIO obj from cache key
- SockIOPool.SockIO sock = pool.getSock( cleanKey, hash );
+ SockIOPool pool = currentPool.get();
+ SockIOPool.SockIO sock = pool.getSockAndFragmentId( cleanKey, hash ).sock();
if ( sock == null ) {
if ( errorHandler != null )
@@ -1572,7 +1708,7 @@ public Map getMulti( String[] keys, Integer[] hashCodes, boolean
// return to pool
sock.close();
}
-
+
if ( log.isInfoEnabled() )
log.info( "multi get socket count : " + cmdMap.size() );
@@ -1615,12 +1751,12 @@ public Map getMulti( String[] keys, Integer[] hashCodes, boolean
return ret;
}
- /**
+ /**
* This method loads the data from cache into a Map.
*
* Pass a SockIO object which is ready to receive data and a HashMap
* to store the results.
- *
+ *
* @param sock socket waiting to pass back data
* @param hm hashmap to store data into
* @param asString if true, and if we are using NativehHandler, return string val
@@ -1644,7 +1780,7 @@ private void loadMulti( LineInputStream input, Map hm, boolean as
log.debug( "++++ flags: " + flag );
log.debug( "++++ length: " + length );
}
-
+
// read obj into buffer
byte[] buf = new byte[length];
input.read( buf );
@@ -1652,7 +1788,7 @@ private void loadMulti( LineInputStream input, Map hm, boolean as
// ready object
Object o;
-
+
// check for compression
if ( (flag & F_COMPRESSED) == F_COMPRESSED ) {
try {
@@ -1661,7 +1797,7 @@ private void loadMulti( LineInputStream input, Map hm, boolean as
// will need to be, and we don't want to resize it a bunch
GZIPInputStream gzi = new GZIPInputStream( new ByteArrayInputStream( buf ) );
ByteArrayOutputStream bos = new ByteArrayOutputStream( buf.length );
-
+
int count;
byte[] tmp = new byte[2048];
while ( (count = gzi.read(tmp)) != -1 ) {
@@ -1694,7 +1830,7 @@ private void loadMulti( LineInputStream input, Map hm, boolean as
else {
// decoding object
try {
- o = NativeHandler.decode( buf, flag );
+ o = NativeHandler.decode( buf, flag );
}
catch ( Exception e ) {
@@ -1717,9 +1853,9 @@ private void loadMulti( LineInputStream input, Map hm, boolean as
log.info( "++++ deserializing " + o.getClass() );
}
catch ( InvalidClassException e ) {
- /* Errors de-serializing are to be expected in the case of a
- * long running server that spans client restarts with updated
- * classes.
+ /* Errors de-serializing are to be expected in the case of a
+ * long running server that spans client restarts with updated
+ * classes.
*/
// if we have an errorHandler, use its hook
if ( errorHandler != null )
@@ -1755,23 +1891,23 @@ private String sanitizeKey( String key ) throws UnsupportedEncodingException {
return ( sanitizeKeys ) ? URLEncoder.encode( key, "UTF-8" ) : key;
}
- /**
+ /**
* Invalidates the entire cache.
*
* Will return true only if succeeds in clearing all servers.
- *
+ *
* @return success true/false
*/
public boolean flushAll() {
return flushAll( null );
}
- /**
+ /**
* Invalidates the entire cache.
*
* Will return true only if succeeds in clearing all servers.
* If pass in null, then will try to flush all servers.
- *
+ *
* @param servers optional array of host(s) to flush (host:port)
* @return success true/false
*/
@@ -1779,15 +1915,20 @@ public boolean flushAll( String[] servers ) {
// get SockIOPool instance
// return false if unable to get SockIO obj
- if ( pool == null ) {
+ if ( currentPool == null || currentPool.get() == null ) {
log.error( "++++ unable to get SockIOPool instance" );
return false;
}
// get all servers and iterate over them
- servers = ( servers == null )
- ? pool.getServers()
- : servers;
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ if (servers == null) {
+ RejigConfig config = pool.getRejigConfig();
+ HashSet hosts = getAllHosts(config);
+ servers = new String[hosts.size()];
+ servers = hosts.toArray(servers);
+ }
// if no servers, then return early
if ( servers == null || servers.length <= 0 ) {
@@ -1798,15 +1939,15 @@ public boolean flushAll( String[] servers ) {
boolean success = true;
for ( int i = 0; i < servers.length; i++ ) {
-
- SockIOPool.SockIO sock = pool.getConnection( servers[i] );
- if ( sock == null ) {
+ SockIOPool.SockAndFragmentId sockAndFragmentId = pool.getHostSockAndFragmentId( servers[i] );
+ if ( sockAndFragmentId == null || sockAndFragmentId.sock() == null ) {
log.error( "++++ unable to get connection to : " + servers[i] );
success = false;
if ( errorHandler != null )
errorHandler.handleErrorOnFlush( this, new IOException( "no socket to server available" ) );
continue;
}
+ SockIOPool.SockIO sock = sockAndFragmentId.sock();
// build command
String command = "flush_all\r\n";
@@ -1851,108 +1992,108 @@ public boolean flushAll( String[] servers ) {
return success;
}
- /**
+ /**
* Retrieves stats for all servers.
*
* Returns a map keyed on the servername.
* The value is another map which contains stats
* with stat name as key and value as value.
- *
+ *
* @return Stats map
*/
public Map stats() {
return stats( null );
}
- /**
+ /**
* Retrieves stats for passed in servers (or all servers).
*
* Returns a map keyed on the servername.
* The value is another map which contains stats
* with stat name as key and value as value.
- *
- * @param servers string array of servers to retrieve stats from, or all if this is null
+ *
+ * @param servers string array of servers to retrieve stats from, or all if this is null
* @return Stats map
*/
public Map stats( String[] servers ) {
return stats( servers, "stats\r\n", STATS );
- }
+ }
- /**
+ /**
* Retrieves stats items for all servers.
*
* Returns a map keyed on the servername.
* The value is another map which contains item stats
* with itemname:number:field as key and value as value.
- *
+ *
* @return Stats map
*/
public Map statsItems() {
return statsItems( null );
}
-
- /**
+
+ /**
* Retrieves stats for passed in servers (or all servers).
*
* Returns a map keyed on the servername.
* The value is another map which contains item stats
* with itemname:number:field as key and value as value.
- *
+ *
* @param servers string array of servers to retrieve stats from, or all if this is null
* @return Stats map
*/
public Map statsItems( String[] servers ) {
return stats( servers, "stats items\r\n", STATS );
}
-
- /**
+
+ /**
* Retrieves stats items for all servers.
*
* Returns a map keyed on the servername.
* The value is another map which contains slabs stats
* with slabnumber:field as key and value as value.
- *
+ *
* @return Stats map
*/
public Map statsSlabs() {
return statsSlabs( null );
}
-
- /**
+
+ /**
* Retrieves stats for passed in servers (or all servers).
*
* Returns a map keyed on the servername.
* The value is another map which contains slabs stats
* with slabnumber:field as key and value as value.
- *
+ *
* @param servers string array of servers to retrieve stats from, or all if this is null
* @return Stats map
*/
public Map statsSlabs( String[] servers ) {
return stats( servers, "stats slabs\r\n", STATS );
}
-
- /**
+
+ /**
* Retrieves items cachedump for all servers.
*
* Returns a map keyed on the servername.
* The value is another map which contains cachedump stats
* with the cachekey as key and byte size and unix timestamp as value.
- *
+ *
* @param slabNumber the item number of the cache dump
* @return Stats map
*/
public Map statsCacheDump( int slabNumber, int limit ) {
return statsCacheDump( null, slabNumber, limit );
}
-
- /**
+
+ /**
* Retrieves stats for passed in servers (or all servers).
*
* Returns a map keyed on the servername.
* The value is another map which contains cachedump stats
* with the cachekey as key and byte size and unix timestamp as value.
- *
+ *
* @param servers string array of servers to retrieve stats from, or all if this is null
* @param slabNumber the item number of the cache dump
* @return Stats map
@@ -1960,7 +2101,7 @@ public Map statsCacheDump( int slabNumber, int limit ) {
public Map statsCacheDump( String[] servers, int slabNumber, int limit ) {
return stats( servers, String.format( "stats cachedump %d %d\r\n", slabNumber, limit ), ITEM );
}
-
+
private Map stats( String[] servers, String command, String lineStart ) {
if ( command == null || command.trim().equals( "" ) ) {
@@ -1969,9 +2110,13 @@ private Map stats( String[] servers, String command, String lineStart ) {
}
// get all servers and iterate over them
- servers = (servers == null)
- ? pool.getServers()
- : servers;
+ SockIOPool pool = currentPool.get();
+ if (servers == null) {
+ RejigConfig config = pool.getRejigConfig();
+ HashSet hosts = getAllHosts(config);
+ servers = new String[hosts.size()];
+ servers = hosts.toArray(servers);
+ }
// if no servers, then return early
if ( servers == null || servers.length <= 0 ) {
@@ -2008,7 +2153,7 @@ private Map stats( String[] servers, String command, String lineStart ) {
log.debug( "++++ line: " + line );
if ( line.startsWith( lineStart ) ) {
- String[] info = line.split( " ", 3 );
+ String[] info = line.split( " ", 3 );
String key = info[1];
String value = info[2];
@@ -2063,6 +2208,426 @@ else if ( line.startsWith( ERROR ) || line.startsWith( CLIENT_ERROR ) || line.st
return statsMaps;
}
+ /** Shuts down the SockIOPool. */
+ public void shutDown() {
+ currentPool.get().shutDown();
+ currentPool.set(null);
+ currentPool = null;
+ poolOptions = null;
+ poolNamePrefix = null;
+ classLoader = null;
+ errorHandler = null;
+ configReader = null;
+ }
+
+ /**
+ * Updates the rejig config on all servers to the specified version.
+ *
+ * @param value value to store
+ * @param expiry when to expire the record
+ * @return true, if the data was successfully stored in all servers.
+ */
+ public boolean setConfig( RejigConfig config, Date expiry ) {
+ for (String server : getAllHosts(config)) {
+ if (!setConfig(config, expiry, server)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Updates the rejig config on the server to the specified version.
+ *
+ * @param value value to store
+ * @param expiry when to expire the record
+ * @param server ip:port of the server to push the config to
+ * @return true, if the data was successfully stored
+ */
+ public boolean setConfig( RejigConfig config, Date expiry, String server ) {
+ // get SockIO obj
+ SockIOPool pool = currentPool.get();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getHostSockAndFragmentId( server );
+
+ if ( sockAndId == null || sockAndId.sock() == null ) {
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnConf( this, new IOException( "no socket to server available" ) );
+ return false;
+ }
+ SockIOPool.SockIO sock = sockAndId.sock();
+
+ if ( expiry == null )
+ expiry = new Date(0);
+
+ // store flags
+ int flags = 0;
+
+ byte[] value = config.toByteArray();
+ // byte array to hold data
+ byte[] val;
+
+ try {
+ if ( log.isInfoEnabled() ) {
+ log.info( "Storing with native handler..." );
+ }
+ flags |= NativeHandler.getMarkerFlag( value );
+ val = NativeHandler.encode( value );
+ }
+ catch ( Exception e ) {
+
+ // if we have an errorHandler, use its hook
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnConf( this, e );
+
+ log.error( "Failed to native handle obj", e );
+
+ sock.close();
+ sock = null;
+ return false;
+ }
+
+ // now try to compress if we want to
+ // and if the length is over the threshold
+ if ( compressEnable && val.length > compressThreshold ) {
+ try {
+ if ( log.isInfoEnabled() ) {
+ log.info( "++++ trying to compress data" );
+ log.info( "++++ size prior to compression: " + val.length );
+ }
+ ByteArrayOutputStream bos = new ByteArrayOutputStream( val.length );
+ GZIPOutputStream gos = new GZIPOutputStream( bos );
+ gos.write( val, 0, val.length );
+ gos.finish();
+ gos.close();
+
+ // store it and set compression flag
+ val = bos.toByteArray();
+ flags |= F_COMPRESSED;
+
+ if ( log.isInfoEnabled() )
+ log.info( "++++ compression succeeded, size after: " + val.length );
+ }
+ catch ( IOException e ) {
+ // if we have an errorHandler, use its hook
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnConf( this, e );
+
+ log.error( "IOException while compressing stream: " + e.getMessage() );
+ log.error( "storing data uncompressed" );
+ }
+ }
+
+ // now write the data to the cache server
+ try {
+ String cmd = String.format( "rj %d %d conf %d %d %d\r\n", config.getId(), config.getFragmentCount(), flags, (expiry.getTime() / 1000), val.length );
+ sock.write( cmd.getBytes() );
+ sock.write( val );
+ sock.write( "\r\n".getBytes() );
+ sock.flush();
+
+ // get result code
+ String line = sock.readLine();
+ if ( log.isInfoEnabled() )
+ log.info( "++++ memcache conf (result code): " + line );
+
+ if ( REFRESH_AND_RETRY.equals(line) ) {
+ if (errorHandler != null) {
+ errorHandler.handleErrorOnConf( this, new IllegalArgumentException("The config is older than the current config on the server."));
+ }
+ sock.close();
+ sock = null;
+ return false;
+ }
+ else if ( STORED.equals( line ) ) {
+ if ( log.isInfoEnabled() )
+ log.info("++++ config successfully stored. Config id: " + config.getId() );
+ sock.close();
+ sock = null;
+ return true;
+ }
+ else if ( NOTSTORED.equals( line ) ) {
+ if ( log.isInfoEnabled() )
+ log.info( "++++ config not stored. Config id: " + config.getId() );
+ }
+ else {
+ log.error( "++++ error storing config. Config id: " + config.getId() + " -- length: " + val.length );
+ log.error( "++++ server response: " + line );
+ }
+ }
+ catch ( IOException e ) {
+
+ // if we have an errorHandler, use its hook
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnConf( this, e );
+
+ // exception thrown
+ log.error( "++++ exception thrown while writing bytes to server on set" );
+ log.error( e.getMessage(), e );
+
+ try {
+ sock.trueClose();
+ }
+ catch ( IOException ioe ) {
+ log.error( "++++ failed to close socket : " + sock.toString() );
+ }
+
+ sock = null;
+ }
+
+ if ( sock != null ) {
+ sock.close();
+ sock = null;
+ }
+
+ return false;
+ }
+
+ /**
+ * Grans a lease on the fragment to the specified host.
+ *
+ * @param fragmentNum the fragment to grant the lease to
+ * @param expiry when to expire the record
+ * @param server ip:port of the server to push the config to
+ * @return true, if the lease was succesfully granted
+ */
+ public boolean grantLease( int fragmentNum, Date expiry, String server ) {
+ // get SockIO obj
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getHostSockAndFragmentId( server );
+
+ if ( sockAndId == null || sockAndId.sock() == null ) {
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnGrantLease( this, new IOException( "no socket to server available" ) );
+ return false;
+ }
+ SockIOPool.SockIO sock = sockAndId.sock();
+
+ if ( expiry == null ) {
+ expiry = new Date(0);
+ }
+
+ // now write the data to the cache server
+ try {
+ String cmd = String.format( "rj %d %d grant %d\r\n", client_config_id, fragmentNum, (expiry.getTime() / 1000) );
+ sock.write( cmd.getBytes() );
+ sock.flush();
+
+ // get result code
+ String line = sock.readLine();
+ if ( log.isInfoEnabled() )
+ log.info( "++++ memcache grant (result code): " + line );
+
+ if ( REFRESH_AND_RETRY.equals(line) ) {
+ if (errorHandler != null) {
+ errorHandler.handleErrorOnGrantLease( this, new IllegalArgumentException("The config id is lower than or the same as the current config id on the server. Host: " + sock.getHost()));
+ }
+ sock.close();
+ sock = null;
+ return false;
+ }
+ else if ( GRANTED.equals( line ) ) {
+ if ( log.isInfoEnabled() )
+ log.info("++++ lease successfully granted. Fragment num: " + fragmentNum );
+ sock.close();
+ sock = null;
+ return true;
+ }
+ else {
+ log.error( "++++ error granting lease. Fragment num: " + fragmentNum );
+ log.error( "++++ server response: " + line );
+ }
+ }
+ catch ( IOException e ) {
+
+ // if we have an errorHandler, use its hook
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnGrantLease( this, e );
+
+ // exception thrown
+ log.error( "++++ exception thrown while writing bytes to server on set" );
+ log.error( e.getMessage(), e );
+
+ try {
+ sock.trueClose();
+ }
+ catch ( IOException ioe ) {
+ log.error( "++++ failed to close socket : " + sock.toString() );
+ }
+
+ sock = null;
+ }
+
+ if ( sock != null ) {
+ sock.close();
+ sock = null;
+ }
+
+ return false;
+ }
+
+ /**
+ * Revokes the lease on the fragment to the specified host.
+ *
+ * @param fragmentNum the fragment to grant the lease to
+ * @param server ip:port of the server to push the config to
+ * @return true, if the lease was succesfully revoked.
+ */
+ public boolean revokeLease( int fragmentNum, String server ) {
+ // get SockIO obj
+ SockIOPool pool = currentPool.get();
+ int client_config_id = pool.getRejigConfig().getId();
+ SockIOPool.SockAndFragmentId sockAndId = pool.getHostSockAndFragmentId( server );
+
+ if ( sockAndId == null || sockAndId.sock() == null ) {
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnRevokeLease( this, new IOException( "no socket to server available" ) );
+ return false;
+ }
+ SockIOPool.SockIO sock = sockAndId.sock();
+
+ // now write the data to the cache server
+ try {
+ String cmd = String.format( "rj %d %d revoke\r\n", client_config_id, fragmentNum );
+ sock.write( cmd.getBytes() );
+ sock.flush();
+
+ // get result code
+ String line = sock.readLine();
+ if ( log.isInfoEnabled() )
+ log.info( "++++ memcache revoke (result code): " + line );
+
+ if ( REFRESH_AND_RETRY.equals(line) ) {
+ if (errorHandler != null) {
+ errorHandler.handleErrorOnRevokeLease( this, new IllegalArgumentException("The config id is lower than the current config id."));
+ }
+ sock.close();
+ sock = null;
+ return false;
+ }
+ else if ( REVOKED.equals( line ) ) {
+ if ( log.isInfoEnabled() )
+ log.info("++++ lease successfully revoked. Fragment num: " + fragmentNum );
+ sock.close();
+ sock = null;
+ return true;
+ }
+ else {
+ log.error( "++++ error revoking lease. Fragment num: " + fragmentNum );
+ log.error( "++++ server response: " + line );
+ }
+ }
+ catch ( IOException e ) {
+
+ // if we have an errorHandler, use its hook
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnRevokeLease( this, e );
+
+ // exception thrown
+ log.error( "++++ exception thrown while writing bytes to server on set" );
+ log.error( e.getMessage(), e );
+
+ try {
+ sock.trueClose();
+ }
+ catch ( IOException ioe ) {
+ log.error( "++++ failed to close socket : " + sock.toString() );
+ }
+
+ sock = null;
+ }
+
+ if ( sock != null ) {
+ sock.close();
+ sock = null;
+ }
+
+ return false;
+ }
+
+ private void handleRefreshAndRetry(SockIOPool pool, SockIOPool.SockIO sock) throws IOException {
+ if ( log.isDebugEnabled() ) {
+ log.debug("++++ refresh and retry. Host: " + sock.getHost());
+ }
+
+ RejigConfig newConfig = null;
+ String line = sock.readLine();
+ if ( END.equals(line) ) {
+ newConfig = configReader.getConfig();
+ }
+ else if ( line.startsWith( VALUE ) ) {
+ String[] info = line.split(" ");
+ String key = info[1];
+ int flag = Integer.parseInt( info[2] );
+ int length = Integer.parseInt( info[3] );
+
+ if ( !REJIG_CONFIG_STORAGE_KEY.equals( key )) {
+ log.error("++++ the key is not equal to the REJIG_CONFIG_STORAGE_KEY! This should never happen!.");
+ throw new IllegalStateException("Key in REFRESH_AND_RETRY error is wrong!");
+ }
+
+ if ( log.isDebugEnabled() ) {
+ log.debug( "++++ key: " + key );
+ log.debug( "++++ flags: " + flag );
+ log.debug( "++++ length: " + length );
+ }
+
+ // read the config bytes into buffer
+ byte[] buf = new byte[length];
+ sock.read( buf );
+ sock.clearEOL();
+
+ if ( (flag & F_COMPRESSED) == F_COMPRESSED ) {
+ try {
+ // read the input stream, and write to a byte array output stream since
+ // we have to read into a byte array, but we don't know how large it
+ // will need to be, and we don't want to resize it a bunch
+ GZIPInputStream gzi = new GZIPInputStream( new ByteArrayInputStream( buf ) );
+ ByteArrayOutputStream bos = new ByteArrayOutputStream( buf.length );
+
+ int count;
+ byte[] tmp = new byte[2048];
+ while ( (count = gzi.read(tmp)) != -1 ) {
+ bos.write( tmp, 0, count );
+ }
+
+ // store uncompressed back to buffer
+ buf = bos.toByteArray();
+ gzi.close();
+ }
+ catch ( IOException e ) {
+ // if we have an errorHandler, use its hook
+ if ( errorHandler != null )
+ errorHandler.handleErrorOnGet( this, e, key );
+
+ log.error( "++++ IOException thrown while trying to uncompress input stream for key: " + key + " -- " + e.getMessage() );
+ throw new NestedIOException( "++++ IOException thrown while trying to uncompress input stream for key: " + key, e );
+ }
+ }
+ newConfig = RejigConfig.parseFrom(buf);
+ log.info( "++++ succesfully parsed rejig config." );
+ }
+
+ if (newConfig.getId() <= pool.getRejigConfig().getId()) {
+ if (errorHandler != null) {
+ errorHandler.handleErrorOnRefreshAndRetry( this,
+ new IOException("++++ new config id is same as old one after REFRESH_AND_RETRY. New Config id: " + newConfig.getId() + ". Old Config id: " + pool.getRejigConfig().getId() + ". Host: " + sock.getHost()) );
+ }
+ return;
+ }
+ sock.close();
+ sock = null;
+ compareAndSetSockIOPool(pool, createSockIOPool(newConfig));
+ }
+
+ private static HashSet getAllHosts(RejigConfig config) {
+ HashSet serverSet = new HashSet<>();
+ for (Fragment f : config.getFragmentList()) {
+ serverSet.add(f.getAddress());
+ }
+ return serverSet;
+ }
+
protected final class NIOLoader {
protected Selector selector;
protected int numConns = 0;
@@ -2074,20 +2639,20 @@ public NIOLoader( MemcachedClient mc ) {
}
private final class Connection {
-
+
public List incoming = new ArrayList();
public ByteBuffer outgoing;
public SockIOPool.SockIO sock;
public SocketChannel channel;
private boolean isDone = false;
-
+
public Connection( SockIOPool.SockIO sock, StringBuilder request ) throws IOException {
if ( log.isDebugEnabled() )
log.debug( "setting up connection to "+sock.getHost() );
-
+
this.sock = sock;
outgoing = ByteBuffer.wrap( request.append( "\r\n" ).toString().getBytes() );
-
+
channel = sock.getChannel();
if ( channel == null )
throw new IOException( "dead connection to: " + sock.getHost() );
@@ -2095,14 +2660,14 @@ public Connection( SockIOPool.SockIO sock, StringBuilder request ) throws IOExce
channel.configureBlocking( false );
channel.register( selector, SelectionKey.OP_WRITE, this );
}
-
+
public void close() {
try {
if ( isDone ) {
// turn off non-blocking IO and return to pool
if ( log.isDebugEnabled() )
log.debug( "++++ gracefully closing connection to "+sock.getHost() );
-
+
channel.configureBlocking( true );
sock.close();
return;
@@ -2111,7 +2676,7 @@ public void close() {
catch ( IOException e ) {
log.warn( "++++ memcache: unexpected error closing normally" );
}
-
+
try {
if ( log.isDebugEnabled() )
log.debug("forcefully closing connection to "+sock.getHost());
@@ -2121,12 +2686,12 @@ public void close() {
}
catch ( IOException ignoreMe ) { }
}
-
+
public boolean isDone() {
// if we know we're done, just say so
- if ( isDone )
+ if ( isDone )
return true;
-
+
// else find out the hard way
int strPos = B_END.length-1;
@@ -2141,11 +2706,11 @@ public boolean isDone() {
bi--;
}
-
+
isDone = strPos < 0;
return isDone;
}
-
+
public ByteBuffer getBuffer() {
int last = incoming.size()-1;
if ( last >= 0 && incoming.get( last ).hasRemaining() ) {
@@ -2157,22 +2722,23 @@ public ByteBuffer getBuffer() {
return newBuf;
}
}
-
+
public String toString() {
return "Connection to " + sock.getHost() + " with " + incoming.size() + " bufs; done is " + isDone;
}
}
-
+
public void doMulti( boolean asString, Map sockKeys, String[] keys, Map ret ) {
-
+
long timeRemaining = 0;
try {
selector = Selector.open();
-
+
// get the sockets, flip them to non-blocking, and set up data
// structures
conns = new Connection[sockKeys.keySet().size()];
numConns = 0;
+ SockIOPool pool = currentPool.get();
for ( Iterator i = sockKeys.keySet().iterator(); i.hasNext(); ) {
// get SockIO obj from hostname
String host = i.next();
@@ -2187,15 +2753,15 @@ public void doMulti( boolean asString, Map sockKeys, Stri
conns[numConns++] = new Connection( sock, sockKeys.get( host ) );
}
-
+
// the main select loop; ends when
// 1) we've received data from all the servers, or
// 2) we time out
long startTime = System.currentTimeMillis();
- long timeout = pool.getMaxBusy();
+ long timeout = pool.getPoolOptions().maxBusyTime;
timeRemaining = timeout;
-
+
while ( numConns > 0 && timeRemaining > 0 ) {
int n = selector.select( Math.min( timeout, 5000 ) );
if ( n > 0 ) {
@@ -2212,7 +2778,7 @@ public void doMulti( boolean asString, Map sockKeys, Stri
// TODO: This seems like a problem area that we need to figure out how to handle.
log.error( "selector timed out waiting for activity" );
}
-
+
timeRemaining = timeout - (System.currentTimeMillis() - startTime);
}
}
@@ -2225,7 +2791,7 @@ public void doMulti( boolean asString, Map sockKeys, Stri
finally {
if ( log.isDebugEnabled() )
log.debug( "Disconnecting; numConns=" + numConns + " timeRemaining=" + timeRemaining );
-
+
// run through our conns and either return them to the pool
// or forcibly close them
try {
@@ -2233,13 +2799,13 @@ public void doMulti( boolean asString, Map sockKeys, Stri
selector.close();
}
catch ( IOException ignoreMe ) { }
-
+
for ( Connection c : conns ) {
if ( c != null )
c.close();
}
}
-
+
// Done! Build the list of results and return them. If we get
// here by a timeout, then some of the connections are probably
// not done. But we'll return what we've got...
@@ -2254,38 +2820,38 @@ public void doMulti( boolean asString, Map sockKeys, Stri
}
}
}
-
+
private void handleError( Throwable e, String[] keys ) {
// if we have an errorHandler, use its hook
if ( errorHandler != null )
errorHandler.handleErrorOnGet( MemcachedClient.this, e, keys );
-
+
// exception thrown
log.error( "++++ exception thrown while getting from cache on getMulti" );
log.error( e.getMessage() );
}
-
+
private void handleKey( SelectionKey key ) throws IOException {
if ( log.isDebugEnabled() )
log.debug( "handling selector op " + key.readyOps() + " for key " + key );
-
+
if ( key.isReadable() )
readResponse( key );
else if ( key.isWritable() )
writeRequest( key );
}
-
+
public void writeRequest( SelectionKey key ) throws IOException {
ByteBuffer buf = ((Connection) key.attachment()).outgoing;
SocketChannel sc = (SocketChannel)key.channel();
-
+
if ( buf.hasRemaining() ) {
if ( log.isDebugEnabled() )
log.debug( "writing " + buf.remaining() + "B to " + ((SocketChannel) key.channel()).socket().getInetAddress() );
sc.write( buf );
}
-
+
if ( !buf.hasRemaining() ) {
if ( log.isDebugEnabled() )
log.debug( "switching to read mode for server " + ((SocketChannel)key.channel()).socket().getInetAddress() );
@@ -2293,7 +2859,7 @@ public void writeRequest( SelectionKey key ) throws IOException {
key.interestOps( SelectionKey.OP_READ );
}
}
-
+
public void readResponse( SelectionKey key ) throws IOException {
Connection conn = (Connection)key.attachment();
ByteBuffer buf = conn.getBuffer();
@@ -2301,7 +2867,7 @@ public void readResponse( SelectionKey key ) throws IOException {
if ( count > 0 ) {
if ( log.isDebugEnabled() )
log.debug( "read " + count + " from " + conn.channel.socket().getInetAddress() );
-
+
if ( conn.isDone() ) {
if ( log.isDebugEnabled() )
log.debug( "connection done to " + conn.channel.socket().getInetAddress() );
diff --git a/src/com/meetup/memcached/NativeHandler.java b/src/main/java/edu/usc/cs550/rejig/client/NativeHandler.java
similarity index 94%
rename from src/com/meetup/memcached/NativeHandler.java
rename to src/main/java/edu/usc/cs550/rejig/client/NativeHandler.java
index 7f9a87e..f39e299 100644
--- a/src/com/meetup/memcached/NativeHandler.java
+++ b/src/main/java/edu/usc/cs550/rejig/client/NativeHandler.java
@@ -13,9 +13,9 @@
* You should have received a copy of the BSD License along with this
* library.
*
- * @author Greg Whalin
+ * @author Greg Whalin
*/
-package com.meetup.memcached;
+package edu.usc.cs550.rejig.client;
import java.util.Date;
import org.apache.log4j.Logger;
@@ -23,68 +23,68 @@
/**
* Handle encoding standard Java types directly which can result in significant
* memory savings:
- *
+ *
* Currently the Memcached driver for Java supports the setSerialize() option.
* This can increase performance in some situations but has a few issues:
- *
+ *
* Code that performs class casting will throw ClassCastExceptions when
* setSerialize is enabled. For example:
- *
+ *
* mc.set( "foo", new Integer( 1 ) ); Integer output = (Integer)mc.get("foo");
- *
+ *
* Will work just file when setSerialize is true but when its false will just throw
* a ClassCastException.
- *
+ *
* Also internally it doesn't support Boolean and since toString is called wastes a
* lot of memory and causes additional performance issue. For example an Integer
* can take anywhere from 1 byte to 10 bytes.
- *
+ *
* Due to the way the memcached slab allocator works it seems like a LOT of wasted
* memory to store primitive types as serialized objects (from a performance and
* memory perspective). In our applications we have millions of small objects and
* wasted memory would become a big problem.
- *
+ *
* For example a Serialized Boolean takes 47 bytes which means it will fit into the
* 64byte LRU. Using 1 byte means it will fit into the 8 byte LRU thus saving 8x
* the memory. This also saves the CPU performance since we don't have to
* serialize bytes back and forth and we can compute the byte[] value directly.
- *
+ *
* One problem would be when the user calls get() because doing so would require
* the app to know the type of the object stored as a bytearray inside memcached
* (since the user will probably cast).
- *
+ *
* If we assume the basic types are interned we could use the first byte as the
* type with the remaining bytes as the value. Then on get() we could read the
* first byte to determine the type and then construct the correct object for it.
* This would prevent the ClassCastException I talked about above.
- *
+ *
* We could remove the setSerialize() option and just assume that standard VM types
* are always internd in this manner.
- *
+ *
* mc.set( "foo", new Boolean.TRUE ); Boolean b = (Boolean)mc.get( "foo" );
- *
+ *
* And the type casts would work because internally we would create a new Boolean
* to return back to the client.
- *
+ *
* This would reduce memory footprint and allow for a virtual implementation of the
* Externalizable interface which is much faster than Serialzation.
- *
+ *
* Currently the memory improvements would be:
- *
+ *
* java.lang.Boolean - 8x performance improvement (now just two bytes)
* java.lang.Integer - 16x performance improvement (now just 5 bytes)
- *
+ *
* Most of the other primitive types would benefit from this optimization.
* java.lang.Character being another obvious example.
- *
+ *
* I know it seems like I'm being really picky here but for our application I'd
* save 1G of memory right off the bat. We'd go down from 1.152G of memory used
* down to 144M of memory used which is much better IMO.
- *
+ *
* http://java.sun.com/docs/books/tutorial/native1.1/integrating/types.html
*
* @author Kevin A. Burton
- * @author Greg Whalin
+ * @author Greg Whalin
*/
public class NativeHandler {
@@ -92,9 +92,9 @@ public class NativeHandler {
private static Logger log =
Logger.getLogger( NativeHandler.class.getName() );
- /**
- * Detemine of object can be natively serialized by this class.
- *
+ /**
+ * Detemine of object can be natively serialized by this class.
+ *
* @param value Object to test.
* @return true/false
*/
@@ -119,9 +119,9 @@ public static boolean isHandled( Object value ) {
: false;
}
- /**
- * Returns the flag for marking the type of the byte array.
- *
+ /**
+ * Returns the flag for marking the type of the byte array.
+ *
* @param value Object we are storing.
* @return int marker
*/
@@ -129,95 +129,95 @@ public static int getMarkerFlag( Object value ) {
if ( value instanceof Byte )
return MemcachedClient.MARKER_BYTE;
-
+
if ( value instanceof Boolean )
return MemcachedClient.MARKER_BOOLEAN;
-
- if ( value instanceof Integer )
+
+ if ( value instanceof Integer )
return MemcachedClient.MARKER_INTEGER;
-
+
if ( value instanceof Long )
return MemcachedClient.MARKER_LONG;
if ( value instanceof Character )
return MemcachedClient.MARKER_CHARACTER;
-
+
if ( value instanceof String )
return MemcachedClient.MARKER_STRING;
-
+
if ( value instanceof StringBuffer )
return MemcachedClient.MARKER_STRINGBUFFER;
-
+
if ( value instanceof Float )
return MemcachedClient.MARKER_FLOAT;
-
+
if ( value instanceof Short )
return MemcachedClient.MARKER_SHORT;
-
+
if ( value instanceof Double )
return MemcachedClient.MARKER_DOUBLE;
-
+
if ( value instanceof Date )
return MemcachedClient.MARKER_DATE;
-
+
if ( value instanceof StringBuilder )
return MemcachedClient.MARKER_STRINGBUILDER;
-
+
if ( value instanceof byte[] )
return MemcachedClient.MARKER_BYTEARR;
-
+
return -1;
}
- /**
- * Encodes supported types
- *
+ /**
+ * Encodes supported types
+ *
* @param value Object to encode.
* @return byte array
*
* @throws Exception If fail to encode.
*/
public static byte[] encode( Object value ) throws Exception {
-
+
if ( value instanceof Byte )
return encode( (Byte)value );
-
+
if ( value instanceof Boolean )
return encode( (Boolean)value );
-
- if ( value instanceof Integer )
+
+ if ( value instanceof Integer )
return encode( ((Integer)value).intValue() );
-
+
if ( value instanceof Long )
return encode( ((Long)value).longValue() );
-
+
if ( value instanceof Character )
return encode( (Character)value );
-
+
if ( value instanceof String )
return encode( (String)value );
-
+
if ( value instanceof StringBuffer )
return encode( (StringBuffer)value );
-
+
if ( value instanceof Float )
return encode( ((Float)value).floatValue() );
-
+
if ( value instanceof Short )
return encode( (Short)value );
-
+
if ( value instanceof Double )
return encode( ((Double)value).doubleValue() );
-
+
if ( value instanceof Date )
return encode( (Date)value);
-
+
if ( value instanceof StringBuilder )
return encode( (StringBuilder)value );
-
+
if ( value instanceof byte[] )
return encode( (byte[])value );
-
+
return null;
}
@@ -241,43 +241,43 @@ protected static byte[] encode( Boolean value ) {
protected static byte[] encode( int value ) {
return getBytes( value );
}
-
+
protected static byte[] encode( long value ) throws Exception {
return getBytes( value );
}
-
+
protected static byte[] encode( Date value ) {
return getBytes( value.getTime() );
}
-
+
protected static byte[] encode( Character value ) {
return encode( value.charValue() );
}
-
+
protected static byte[] encode( String value ) throws Exception {
return value.getBytes( "UTF-8" );
}
-
+
protected static byte[] encode( StringBuffer value ) throws Exception {
return encode( value.toString() );
}
-
+
protected static byte[] encode( float value ) throws Exception {
return encode( (int)Float.floatToIntBits( value ) );
}
-
+
protected static byte[] encode( Short value ) throws Exception {
return encode( (int)value.shortValue() );
}
-
+
protected static byte[] encode( double value ) throws Exception {
return encode( (long)Double.doubleToLongBits( value ) );
}
-
+
protected static byte[] encode( StringBuilder value ) throws Exception {
return encode( value.toString() );
}
-
+
protected static byte[] encode( byte[] value ) {
return value;
}
@@ -294,7 +294,7 @@ protected static byte[] getBytes( long value ) {
b[7] = (byte)((value >> 0) & 0xFF);
return b;
}
-
+
protected static byte[] getBytes( int value ) {
byte[] b = new byte[4];
b[0] = (byte)((value >> 24) & 0xFF);
@@ -303,133 +303,133 @@ protected static byte[] getBytes( int value ) {
b[3] = (byte)((value >> 0) & 0xFF);
return b;
}
-
- /**
+
+ /**
* Decodes byte array using memcache flag to determine type.
- *
- * @param b
- * @param marker
- * @return
- * @throws Exception
+ *
+ * @param b
+ * @param marker
+ * @return
+ * @throws Exception
*/
public static Object decode( byte[] b, int flag ) throws Exception {
if ( b.length < 1 )
return null;
-
+
if ( ( flag & MemcachedClient.MARKER_BYTE ) == MemcachedClient.MARKER_BYTE )
return decodeByte( b );
-
+
if ( ( flag & MemcachedClient.MARKER_BOOLEAN ) == MemcachedClient.MARKER_BOOLEAN )
return decodeBoolean( b );
-
+
if ( ( flag & MemcachedClient.MARKER_INTEGER ) == MemcachedClient.MARKER_INTEGER )
return decodeInteger( b );
-
+
if ( ( flag & MemcachedClient.MARKER_LONG ) == MemcachedClient.MARKER_LONG )
return decodeLong( b );
-
+
if ( ( flag & MemcachedClient.MARKER_CHARACTER ) == MemcachedClient.MARKER_CHARACTER )
return decodeCharacter( b );
-
+
if ( ( flag & MemcachedClient.MARKER_STRING ) == MemcachedClient.MARKER_STRING )
return decodeString( b );
-
+
if ( ( flag & MemcachedClient.MARKER_STRINGBUFFER ) == MemcachedClient.MARKER_STRINGBUFFER )
return decodeStringBuffer( b );
-
+
if ( ( flag & MemcachedClient.MARKER_FLOAT ) == MemcachedClient.MARKER_FLOAT )
return decodeFloat( b );
-
+
if ( ( flag & MemcachedClient.MARKER_SHORT ) == MemcachedClient.MARKER_SHORT )
return decodeShort( b );
-
+
if ( ( flag & MemcachedClient.MARKER_DOUBLE ) == MemcachedClient.MARKER_DOUBLE )
return decodeDouble( b );
-
+
if ( ( flag & MemcachedClient.MARKER_DATE ) == MemcachedClient.MARKER_DATE )
return decodeDate( b );
-
+
if ( ( flag & MemcachedClient.MARKER_STRINGBUILDER ) == MemcachedClient.MARKER_STRINGBUILDER )
return decodeStringBuilder( b );
-
+
if ( ( flag & MemcachedClient.MARKER_BYTEARR ) == MemcachedClient.MARKER_BYTEARR )
return decodeByteArr( b );
-
+
return null;
}
-
+
// decode methods
protected static Byte decodeByte( byte[] b ) {
return new Byte( b[0] );
}
-
+
protected static Boolean decodeBoolean( byte[] b ) {
boolean value = b[0] == 1;
return ( value ) ? Boolean.TRUE : Boolean.FALSE;
}
-
+
protected static Integer decodeInteger( byte[] b ) {
return new Integer( toInt( b ) );
}
-
+
protected static Long decodeLong( byte[] b ) throws Exception {
return new Long( toLong( b ) );
}
-
+
protected static Character decodeCharacter( byte[] b ) {
return new Character( (char)decodeInteger( b ).intValue() );
}
-
+
protected static String decodeString( byte[] b ) throws Exception {
return new String( b, "UTF-8" );
}
-
+
protected static StringBuffer decodeStringBuffer( byte[] b ) throws Exception {
return new StringBuffer( decodeString( b ) );
}
-
+
protected static Float decodeFloat( byte[] b ) throws Exception {
Integer l = decodeInteger( b );
return new Float( Float.intBitsToFloat( l.intValue() ) );
}
-
+
protected static Short decodeShort( byte[] b ) throws Exception {
return new Short( (short)decodeInteger( b ).intValue() );
}
-
+
protected static Double decodeDouble( byte[] b ) throws Exception {
Long l = decodeLong( b );
return new Double( Double.longBitsToDouble( l.longValue() ) );
}
-
+
protected static Date decodeDate( byte[] b ) {
return new Date( toLong( b ) );
}
-
+
protected static StringBuilder decodeStringBuilder( byte[] b ) throws Exception {
return new StringBuilder( decodeString( b ) );
}
-
+
protected static byte[] decodeByteArr( byte[] b ) {
return b;
}
-
- /**
+
+ /**
* This works by taking each of the bit patterns and converting them to
* ints taking into account 2s complement and then adding them..
- *
- * @param b
- * @return
+ *
+ * @param b
+ * @return
*/
protected static int toInt( byte[] b ) {
return (((((int) b[3]) & 0xFF) << 32) +
((((int) b[2]) & 0xFF) << 40) +
((((int) b[1]) & 0xFF) << 48) +
((((int) b[0]) & 0xFF) << 56));
- }
-
+ }
+
protected static long toLong( byte[] b ) {
return ((((long) b[7]) & 0xFF) +
((((long) b[6]) & 0xFF) << 8) +
@@ -439,5 +439,5 @@ protected static long toLong( byte[] b ) {
((((long) b[2]) & 0xFF) << 40) +
((((long) b[1]) & 0xFF) << 48) +
((((long) b[0]) & 0xFF) << 56));
- }
+ }
}
diff --git a/src/com/meetup/memcached/NestedIOException.java b/src/main/java/edu/usc/cs550/rejig/client/NestedIOException.java
similarity index 92%
rename from src/com/meetup/memcached/NestedIOException.java
rename to src/main/java/edu/usc/cs550/rejig/client/NestedIOException.java
index 264605e..a4b01d0 100644
--- a/src/com/meetup/memcached/NestedIOException.java
+++ b/src/main/java/edu/usc/cs550/rejig/client/NestedIOException.java
@@ -13,16 +13,16 @@
* You should have received a copy of the BSD License along with this
* library.
*
- * @author Kevin A. Burton
+ * @author Kevin A. Burton
*/
-package com.meetup.memcached;
+package edu.usc.cs550.rejig.client;
import java.io.*;
/**
* Bridge class to provide nested Exceptions with IOException which has
* constructors that don't take Throwables.
- *
+ *
* @author Kevin Burton
* @version 1.2
*/
diff --git a/src/com/meetup/memcached/SockIOPool.java b/src/main/java/edu/usc/cs550/rejig/client/SockIOPool.java
similarity index 66%
rename from src/com/meetup/memcached/SockIOPool.java
rename to src/main/java/edu/usc/cs550/rejig/client/SockIOPool.java
index 6fe0c92..806cb14 100644
--- a/src/com/meetup/memcached/SockIOPool.java
+++ b/src/main/java/edu/usc/cs550/rejig/client/SockIOPool.java
@@ -13,9 +13,12 @@
* You should have received a copy of the BSD License along with this
* library.
*
- * @author greg whalin
+ * @author greg whalin
*/
-package com.meetup.memcached;
+package edu.usc.cs550.rejig.client;
+import edu.usc.cs550.rejig.interfaces.Fragment;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
+
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
@@ -42,7 +45,7 @@
import java.util.concurrent.locks.ReentrantLock;
import org.apache.log4j.Logger;
-/**
+/**
* This class is a connection pool for maintaning a pool of persistent connections
* to memcached servers.
*
@@ -57,41 +60,41 @@
*
* SockIOPool pool = SockIOPool.getInstance();
* pool.setServers(serverlist);
- * pool.initialize();
+ * pool.initialize();
* }
- *
+ *
* An example of initializing using defaults and providing weights for servers:
*
* static {
* String[] serverlist = { "cache0.server.com:12345", "cache1.server.com:12345" };
* Integer[] weights = { new Integer(5), new Integer(2) };
- *
+ *
* SockIOPool pool = SockIOPool.getInstance();
* pool.setServers(serverlist);
- * pool.setWeights(weights);
- * pool.initialize();
+ * pool.setWeights(weights);
+ * pool.initialize();
* }
- *
+ *
* An example of initializing overriding defaults:
*
* static {
* String[] serverlist = { "cache0.server.com:12345", "cache1.server.com:12345" };
- * Integer[] weights = { new Integer(5), new Integer(2) };
+ * Integer[] weights = { new Integer(5), new Integer(2) };
* int initialConnections = 10;
* int minSpareConnections = 5;
- * int maxSpareConnections = 50;
+ * int maxSpareConnections = 50;
* long maxIdleTime = 1000 * 60 * 30; // 30 minutes
* long maxBusyTime = 1000 * 60 * 5; // 5 minutes
* long maintThreadSleep = 1000 * 5; // 5 seconds
* int socketTimeOut = 1000 * 3; // 3 seconds to block on reads
* int socketConnectTO = 1000 * 3; // 3 seconds to block on initial connections. If 0, then will use blocking connect (default)
- * boolean failover = false; // turn off auto-failover in event of server down
- * boolean nagleAlg = false; // turn off Nagle's algorithm on all sockets in pool
+ * boolean failover = false; // turn off auto-failover in event of server down
+ * boolean nagleAlg = false; // turn off Nagle's algorithm on all sockets in pool
* boolean aliveCheck = false; // disable health check of socket on checkout
*
* SockIOPool pool = SockIOPool.getInstance();
* pool.setServers( serverlist );
- * pool.setWeights( weights );
+ * pool.setWeights( weights );
* pool.setInitConn( initialConnections );
* pool.setMinConn( minSpareConnections );
* pool.setMaxConn( maxSpareConnections );
@@ -99,29 +102,30 @@
* pool.setMaxBusyTime( maxBusyTime );
* pool.setMaintSleep( maintThreadSleep );
* pool.setSocketTO( socketTimeOut );
- * pool.setNagle( nagleAlg );
+ * pool.setNagle( nagleAlg );
* pool.setHashingAlg( SockIOPool.NEW_COMPAT_HASH );
* pool.setAliveCheck( true );
- * pool.initialize();
+ * pool.initialize();
* }
- *
- * The easiest manner in which to initialize the pool is to set the servers and rely on defaults as in the first example.
+ *
+ * The easiest manner in which to initialize the pool is to set the servers and rely on defaults as in the first example.
* After pool is initialized, a client will request a SockIO object by calling getSock with the cache key
- * The client must always close the SockIO object when finished, which will return the connection back to the pool.
+ * The client must always close the SockIO object when finished, which will return the connection back to the pool.
* An example of retrieving a SockIO object:
*
* SockIOPool.SockIO sock = SockIOPool.getInstance().getSock( key );
* try {
- * sock.write( "version\r\n" );
- * sock.flush();
- * System.out.println( "Version: " + sock.readLine() );
+ * sock.write( "version\r\n" );
+ * sock.flush();
+ * System.out.println( "Version: " + sock.readLine() );
* }
- * catch (IOException ioe) { System.out.println( "io exception thrown" ) };
+ * catch (IOException ioe) { System.out.println( "io exception thrown" ) };
*
- * sock.close();
- *
+ * sock.close();
+ *
*
- * @author greg whalin
+ * @author greg whalin
+ * @author Likhit Dharmapuri
* @version 1.5
*/
public class SockIOPool {
@@ -134,78 +138,101 @@ public class SockIOPool {
private static Map pools =
new HashMap();
- // avoid recurring construction
- private static ThreadLocal MD5 = new ThreadLocal() {
- @Override
- protected MessageDigest initialValue() {
- try {
- return MessageDigest.getInstance( "MD5" );
- }
- catch ( NoSuchAlgorithmException e ) {
- log.error( "++++ no md5 algorithm found" );
- throw new IllegalStateException( "++++ no md5 algorythm found");
- }
- }
- };
-
// Constants
private static final Integer ZERO = new Integer( 0 );
- public static final int NATIVE_HASH = 0; // native String.hashCode();
- public static final int OLD_COMPAT_HASH = 1; // original compatibility hashing algorithm (works with other clients)
- public static final int NEW_COMPAT_HASH = 2; // new CRC32 based compatibility hashing algorithm (works with other clients)
- public static final int CONSISTENT_HASH = 3; // MD5 Based -- Stops thrashing when a server added or removed
-
public static final long MAX_RETRY_DELAY = 10 * 60 * 1000; // max of 10 minute delay for fall off
+ // Hashing algorithm to use to select the fragment to assign a key to.
+ public static enum FragmentHashingAlgo {
+ NATIVE_HASH, // native String.hashCode();
+ OLD_COMPAT_HASH, // original compatibility hashing algorithm (works with other clients)
+ NEW_COMPAT_HASH // new CRC32 based compatibility hashing algorithm (works with other clients)
+ }
+
// Pool data
private MaintThread maintThread;
private boolean initialized = false;
private int maxCreate = 1; // this will be initialized by pool when the pool is initialized
- // initial, min and max pool sizes
- private int poolMultiplier = 3;
- private int initConn = 10;
- private int minConn = 5;
- private int maxConn = 100;
- private long maxIdle = 1000 * 60 * 5; // max idle time for avail sockets
- private long maxBusyTime = 1000 * 30; // max idle time for avail sockets
- private long maintSleep = 1000 * 30; // maintenance thread sleep time
- private int socketTO = 1000 * 3; // default timeout of socket reads
- private int socketConnectTO = 1000 * 3; // default timeout of socket connections
- private boolean aliveCheck = false; // default to not check each connection for being alive
- private boolean failover = true; // default to failover in event of cache server dead
- private boolean failback = true; // only used if failover is also set ... controls putting a dead server back into rotation
- private boolean nagle = false; // enable/disable Nagle's algorithm
- private int hashingAlg = NATIVE_HASH; // default to using the native hash as it is the fastest
+ /**
+ * All the configurable parameters used by
+ * the SockIOPool instance.
+ */
+ public static class SockIOPoolOptions {
+ /** The initial number of connections per server in the available pool. */
+ public int initConn = 10;
+ /** The min number of connections per server in the available pool. */
+ public int minConn = 5;
+ /** The max number of connections per server in the available pool. */
+ public int maxConn = 100;
+ /** The max idle time for avail sockets */
+ public long maxIdle = 1000 * 60 * 5;
+ /** The max idle time for avail sockets */
+ public long maxBusyTime = 1000 * 30;
+ /** Maintenance thread sleep time */
+ public long maintSleep = 1000 * 30;
+ /** Default timeout of socket reads */
+ public int socketTO = 1000 * 3;
+ /** Default timeout of socket connections */
+ public int socketConnectTO = 1000 * 3;
+ /** Default to not check each connection for being alive */
+ public boolean aliveCheck = false;
+ /** Default to failover in event of cache server dead */
+ public boolean failover = true;
+ /** Only used if failover is also set ... controls putting a dead server back into rotation */
+ public boolean failback = true;
+ /** Enable/disable Nagle's algorithm */
+ public boolean nagle = false;
+ /** Default to using the native hash as it is the fastest */
+ public FragmentHashingAlgo hashingAlg = FragmentHashingAlgo.NATIVE_HASH;
+
+ public SockIOPoolOptions copy() {
+ SockIOPoolOptions copy = new SockIOPoolOptions();
+ copy.initConn = initConn;
+ copy.minConn = minConn;
+ copy.maxConn = maxConn;
+ copy.maxIdle = maxIdle;
+ copy.maxBusyTime = maxBusyTime;
+ copy.maintSleep = maintSleep;
+ copy.socketTO = socketTO;
+ copy.socketConnectTO = socketConnectTO;
+ copy.aliveCheck = aliveCheck;
+ copy.failover = failover;
+ copy.failback = failback;
+ copy.nagle = nagle;
+ copy.hashingAlg = hashingAlg;
+ return copy;
+ }
+ }
+
+ private SockIOPoolOptions options;
+
+ private int poolMultiplier = 3;
+
// locks
private final ReentrantLock hostDeadLock = new ReentrantLock();
// list of all servers
- private String[] servers;
- private Integer[] weights;
- private Integer totalWeight = 0;
-
- private List buckets;
- private TreeMap consistentBuckets;
+ private RejigConfig config;
// dead server map
private Map hostDead;
private Map hostDeadDur;
-
+
// map to hold all available sockets
// map to hold busy sockets
// set to hold sockets to close
private Map> availPool;
private Map> busyPool;
private Map deadPool;
-
+
// empty constructor
protected SockIOPool() { }
-
- /**
- * Factory to create/retrieve new pools given a unique poolName.
- *
+
+ /**
+ * Factory to create/retrieve new pools given a unique poolName.
+ *
* @param poolName unique name of the pool
* @return instance of SockIOPool
*/
@@ -219,261 +246,42 @@ public static synchronized SockIOPool getInstance( String poolName ) {
return pool;
}
- /**
- * Single argument version of factory used for back compat.
- * Simply creates a pool named "default".
- *
- * @return instance of SockIOPool
- */
- public static SockIOPool getInstance() {
- return getInstance( "default" );
- }
-
- /**
- * Sets the list of all cache servers.
- *
- * @param servers String array of servers [host:port]
- */
- public void setServers( String[] servers ) { this.servers = servers; }
-
- /**
- * Returns the current list of all cache servers.
- *
- * @return String array of servers [host:port]
- */
- public String[] getServers() { return this.servers; }
-
- /**
- * Sets the list of weights to apply to the server list.
- *
- * This is an int array with each element corresponding to an element
- * in the same position in the server String array.
- *
- * @param weights Integer array of weights
- */
- public void setWeights( Integer[] weights ) { this.weights = weights; }
-
- /**
- * Returns the current list of weights.
- *
- * @return int array of weights
- */
- public Integer[] getWeights() { return this.weights; }
-
- /**
- * Sets the initial number of connections per server in the available pool.
- *
- * @param initConn int number of connections
- */
- public void setInitConn( int initConn ) { this.initConn = initConn; }
-
- /**
- * Returns the current setting for the initial number of connections per server in
- * the available pool.
- *
- * @return number of connections
- */
- public int getInitConn() { return this.initConn; }
-
- /**
- * Sets the minimum number of spare connections to maintain in our available pool.
- *
- * @param minConn number of connections
- */
- public void setMinConn( int minConn ) { this.minConn = minConn; }
-
- /**
- * Returns the minimum number of spare connections in available pool.
- *
- * @return number of connections
- */
- public int getMinConn() { return this.minConn; }
-
- /**
- * Sets the maximum number of spare connections allowed in our available pool.
- *
- * @param maxConn number of connections
- */
- public void setMaxConn( int maxConn ) { this.maxConn = maxConn; }
-
- /**
- * Returns the maximum number of spare connections allowed in available pool.
- *
- * @return number of connections
- */
- public int getMaxConn() { return this.maxConn; }
-
- /**
- * Sets the max idle time for threads in the available pool.
- *
- * @param maxIdle idle time in ms
- */
- public void setMaxIdle( long maxIdle ) { this.maxIdle = maxIdle; }
-
- /**
- * Returns the current max idle setting.
- *
- * @return max idle setting in ms
- */
- public long getMaxIdle() { return this.maxIdle; }
-
- /**
- * Sets the max busy time for threads in the busy pool.
- *
- * @param maxBusyTime idle time in ms
- */
- public void setMaxBusyTime( long maxBusyTime ) { this.maxBusyTime = maxBusyTime; }
-
- /**
- * Returns the current max busy setting.
- *
- * @return max busy setting in ms
- */
- public long getMaxBusy() { return this.maxBusyTime; }
-
- /**
- * Set the sleep time between runs of the pool maintenance thread.
- * If set to 0, then the maint thread will not be started.
- *
- * @param maintSleep sleep time in ms
- */
- public void setMaintSleep( long maintSleep ) { this.maintSleep = maintSleep; }
-
- /**
- * Returns the current maint thread sleep time.
- *
- * @return sleep time in ms
- */
- public long getMaintSleep() { return this.maintSleep; }
-
- /**
- * Sets the socket timeout for reads.
- *
- * @param socketTO timeout in ms
- */
- public void setSocketTO( int socketTO ) { this.socketTO = socketTO; }
-
- /**
- * Returns the socket timeout for reads.
- *
- * @return timeout in ms
- */
- public int getSocketTO() { return this.socketTO; }
-
- /**
- * Sets the socket timeout for connect.
- *
- * @param socketConnectTO timeout in ms
- */
- public void setSocketConnectTO( int socketConnectTO ) { this.socketConnectTO = socketConnectTO; }
-
- /**
- * Returns the socket timeout for connect.
- *
- * @return timeout in ms
- */
- public int getSocketConnectTO() { return this.socketConnectTO; }
-
- /**
- * Sets the failover flag for the pool.
- *
- * If this flag is set to true, and a socket fails to connect,
- * the pool will attempt to return a socket from another server
- * if one exists. If set to false, then getting a socket
- * will return null if it fails to connect to the requested server.
- *
- * @param failover true/false
- */
- public void setFailover( boolean failover ) { this.failover = failover; }
-
- /**
- * Returns current state of failover flag.
- *
- * @return true/false
- */
- public boolean getFailover() { return this.failover; }
-
- /**
- * Sets the failback flag for the pool.
- *
- * If this is true and we have marked a host as dead,
- * will try to bring it back. If it is false, we will never
- * try to resurrect a dead host.
- *
- * @param failback true/false
- */
- public void setFailback( boolean failback ) { this.failback = failback; }
-
- /**
- * Returns current state of failover flag.
- *
- * @return true/false
- */
- public boolean getFailback() { return this.failback; }
-
/**
- * Sets the aliveCheck flag for the pool.
- *
- * When true, this will attempt to talk to the server on
- * every connection checkout to make sure the connection is
- * still valid. This adds extra network chatter and thus is
- * defaulted off. May be useful if you want to ensure you do
- * not have any problems talking to the server on a dead connection.
+ * Sets the RejigConfig containing the list of all servers.
*
- * @param aliveCheck true/false
+ * @param config RejigConfig object
*/
- public void setAliveCheck( boolean aliveCheck ) { this.aliveCheck = aliveCheck; }
-
+ public SockIOPool setRejigConfig( RejigConfig config ) {
+ this.config = config;
+ return this;
+ }
/**
- * Returns the current status of the aliveCheck flag.
- *
- * @return true / false
- */
- public boolean getAliveCheck() { return this.aliveCheck; }
-
- /**
- * Sets the Nagle alg flag for the pool.
- *
- * If false, will turn off Nagle's algorithm on all sockets created.
- *
- * @param nagle true/false
- */
- public void setNagle( boolean nagle ) { this.nagle = nagle; }
-
- /**
- * Returns current status of nagle flag
- *
- * @return true/false
+ * Returns the current RejigConfig containing the list of all servers.
*/
- public boolean getNagle() { return this.nagle; }
+ public RejigConfig getRejigConfig() { return this.config; }
- /**
- * Sets the hashing algorithm we will use.
- *
- * The types are as follows.
- *
- * SockIOPool.NATIVE_HASH (0) - native String.hashCode() - fast (cached) but not compatible with other clients
- * SockIOPool.OLD_COMPAT_HASH (1) - original compatibility hashing alg (works with other clients)
- * SockIOPool.NEW_COMPAT_HASH (2) - new CRC32 based compatibility hashing algorithm (fast and works with other clients)
- *
- * @param alg int value representing hashing algorithm
+ /**
+ * Sets the SockIOPoolOptions that the object.
*/
- public void setHashingAlg( int alg ) { this.hashingAlg = alg; }
+ public SockIOPool setPoolOptions(SockIOPoolOptions options) {
+ this.options = options;
+ return this;
+ }
- /**
- * Returns current status of customHash flag
- *
- * @return true/false
+ /**
+ * Returns a copy of the SockIOPoolOptions that the object.
*/
- public int getHashingAlg() { return this.hashingAlg; }
+ public SockIOPoolOptions getPoolOptions() {
+ return this.options.copy();
+ }
- /**
+ /**
* Internal private hashing method.
*
* This is the original hashing algorithm from other clients.
* Found to be slow and have poor distribution.
- *
+ *
* @param key String to hash
* @return hashCode for this string using our own hashing algorithm
*/
@@ -488,16 +296,16 @@ private static long origCompatHashingAlg( String key ) {
return hash;
}
- /**
+ /**
* Internal private hashing method.
*
* This is the new hashing algorithm from other clients.
- * Found to be fast and have very good distribution.
+ * Found to be fast and have very good distribution.
*
* UPDATE: This is dog slow under java
- *
- * @param key
- * @return
+ *
+ * @param key
+ * @return
*/
private static long newCompatHashingAlg( String key ) {
CRC32 checksum = new CRC32();
@@ -506,51 +314,28 @@ private static long newCompatHashingAlg( String key ) {
return (crc >> 16) & 0x7fff;
}
- /**
- * Internal private hashing method.
+ /**
+ * Returns a bucket to check for a given key.
*
- * MD5 based hash algorithm for use in the consistent
- * hashing approach.
- *
- * @param key
- * @return
- */
- private static long md5HashingAlg( String key ) {
- MessageDigest md5 = MD5.get();
- md5.reset();
- md5.update( key.getBytes() );
- byte[] bKey = md5.digest();
- long res = ((long)(bKey[3]&0xFF) << 24) | ((long)(bKey[2]&0xFF) << 16) | ((long)(bKey[1]&0xFF) << 8) | (long)(bKey[0]&0xFF);
- return res;
- }
-
- /**
- * Returns a bucket to check for a given key.
- *
* @param key String key cache is stored under
* @return int bucket
*/
private long getHash( String key, Integer hashCode ) {
if ( hashCode != null ) {
- if ( hashingAlg == CONSISTENT_HASH )
- return hashCode.longValue() & 0xffffffffL;
- else
return hashCode.longValue();
}
else {
- switch ( hashingAlg ) {
+ switch ( options.hashingAlg ) {
case NATIVE_HASH:
return (long)key.hashCode();
case OLD_COMPAT_HASH:
return origCompatHashingAlg( key );
case NEW_COMPAT_HASH:
return newCompatHashingAlg( key );
- case CONSISTENT_HASH:
- return md5HashingAlg( key );
default:
// use the native hash as a default
- hashingAlg = NATIVE_HASH;
+ options.hashingAlg = FragmentHashingAlgo.NATIVE_HASH;
return (long)key.hashCode();
}
}
@@ -559,198 +344,103 @@ private long getHash( String key, Integer hashCode ) {
private long getBucket( String key, Integer hashCode ) {
long hc = getHash( key, hashCode );
- if ( this.hashingAlg == CONSISTENT_HASH ) {
- return findPointFor( hc );
- }
- else {
- long bucket = hc % buckets.size();
- if ( bucket < 0 ) bucket *= -1;
- return bucket;
- }
+ long bucket = hc % config.getFragmentCount();
+ if ( bucket < 0 ) bucket *= -1;
+ return bucket;
}
/**
- * Gets the first available key equal or above the given one, if none found,
- * returns the first k in the bucket
- * @param k key
- * @return
- */
- private Long findPointFor( Long hv ) {
- // this works in java 6, but still want to release support for java5
- //Long k = this.consistentBuckets.ceilingKey( hv );
- //return ( k == null ) ? this.consistentBuckets.firstKey() : k;
-
- SortedMap tmap =
- this.consistentBuckets.tailMap( hv );
-
- return ( tmap.isEmpty() ) ? this.consistentBuckets.firstKey() : tmap.firstKey();
- }
-
- /**
- * Initializes the pool.
+ * Initializes the pool.
*/
- public void initialize() {
+ public SockIOPool initialize() {
synchronized( this ) {
// check to see if already initialized
if ( initialized
- && ( buckets != null || consistentBuckets != null )
&& ( availPool != null )
&& ( busyPool != null ) ) {
- log.error( "++++ trying to initialize an already initialized pool" );
- return;
+ log.error( "++++ trying to initialize an already initialized pool" );
+ return this;
}
// pools
- availPool = new HashMap>( servers.length * initConn );
- busyPool = new HashMap>( servers.length * initConn );
+ availPool = new HashMap>( config.getFragmentCount() * options.initConn );
+ busyPool = new HashMap>( config.getFragmentCount() * options.initConn );
deadPool = new IdentityHashMap();
hostDeadDur = new HashMap();
hostDead = new HashMap();
- maxCreate = (poolMultiplier > minConn) ? minConn : minConn / poolMultiplier; // only create up to maxCreate connections at once
+ maxCreate = (poolMultiplier > options.minConn) ? options.minConn : options.minConn / poolMultiplier; // only create up to maxCreate connections at once
if ( log.isDebugEnabled() ) {
log.debug( "++++ initializing pool with following settings:" );
- log.debug( "++++ initial size: " + initConn );
- log.debug( "++++ min spare : " + minConn );
- log.debug( "++++ max spare : " + maxConn );
+ log.debug( "++++ initial size: " + options.initConn );
+ log.debug( "++++ min spare : " + options.minConn );
+ log.debug( "++++ max spare : " + options.maxConn );
}
// if servers is not set, or it empty, then
// throw a runtime exception
- if ( servers == null || servers.length <= 0 ) {
+ if ( config == null || config.getFragmentCount() <= 0 ) {
log.error( "++++ trying to initialize with no servers" );
throw new IllegalStateException( "++++ trying to initialize with no servers" );
}
// initalize our internal hashing structures
- if ( this.hashingAlg == CONSISTENT_HASH )
- populateConsistentBuckets();
- else
- populateBuckets();
+ populateBuckets();
// mark pool as initialized
this.initialized = true;
// start maint thread
- if ( this.maintSleep > 0 )
+ if ( this.options.maintSleep > 0 )
this.startMaintThread();
}
+
+ return this;
}
private void populateBuckets() {
if ( log.isDebugEnabled() )
log.debug( "++++ initializing internal hashing structure for consistent hashing" );
- // store buckets in tree map
- this.buckets = new ArrayList();
-
- for ( int i = 0; i < servers.length; i++ ) {
- if ( this.weights != null && this.weights.length > i ) {
- for ( int k = 0; k < this.weights[i].intValue(); k++ ) {
- this.buckets.add( servers[i] );
- if ( log.isDebugEnabled() )
- log.debug( "++++ added " + servers[i] + " to server bucket" );
- }
- }
- else {
- this.buckets.add( servers[i] );
- if ( log.isDebugEnabled() )
- log.debug( "++++ added " + servers[i] + " to server bucket" );
- }
-
+ for ( int i = 0; i < config.getFragmentCount(); i++ ) {
+ String server = config.getFragment(i).getAddress();
// create initial connections
if ( log.isDebugEnabled() )
- log.debug( "+++ creating initial connections (" + initConn + ") for host: " + servers[i] );
+ log.debug( "+++ creating initial connections (" + options.initConn + ") for host: " + server);
- for ( int j = 0; j < initConn; j++ ) {
- SockIO socket = createSocket( servers[i] );
+ for ( int j = 0; j < options.initConn; j++ ) {
+ SockIO socket = createSocket( server );
if ( socket == null ) {
- log.error( "++++ failed to create connection to: " + servers[i] + " -- only " + j + " created." );
+ log.error( "++++ failed to create connection to: " + server + " -- only " + j + " created." );
break;
}
- addSocketToPool( availPool, servers[i], socket );
+ addSocketToPool( availPool, server, socket );
if ( log.isDebugEnabled() )
- log.debug( "++++ created and added socket: " + socket.toString() + " for host " + servers[i] );
+ log.debug( "++++ created and added socket: " + socket.toString() + " for host " + server );
}
}
}
- private void populateConsistentBuckets() {
- if ( log.isDebugEnabled() )
- log.debug( "++++ initializing internal hashing structure for consistent hashing" );
-
- // store buckets in tree map
- this.consistentBuckets = new TreeMap();
-
- MessageDigest md5 = MD5.get();
- if ( this.totalWeight <= 0 && this.weights != null ) {
- for ( int i = 0; i < this.weights.length; i++ )
- this.totalWeight += ( this.weights[i] == null ) ? 1 : this.weights[i];
- }
- else if ( this.weights == null ) {
- this.totalWeight = this.servers.length;
- }
-
- for ( int i = 0; i < servers.length; i++ ) {
- int thisWeight = 1;
- if ( this.weights != null && this.weights[i] != null )
- thisWeight = this.weights[i];
-
- double factor = Math.floor( ((double)(40 * this.servers.length * thisWeight)) / (double)this.totalWeight );
-
- for ( long j = 0; j < factor; j++ ) {
- byte[] d = md5.digest( ( servers[i] + "-" + j ).getBytes() );
- for ( int h = 0 ; h < 4; h++ ) {
- Long k =
- ((long)(d[3+h*4]&0xFF) << 24)
- | ((long)(d[2+h*4]&0xFF) << 16)
- | ((long)(d[1+h*4]&0xFF) << 8)
- | ((long)(d[0+h*4]&0xFF));
-
- consistentBuckets.put( k, servers[i] );
- if ( log.isDebugEnabled() )
- log.debug( "++++ added " + servers[i] + " to server bucket" );
- }
- }
-
- // create initial connections
- if ( log.isDebugEnabled() )
- log.debug( "+++ creating initial connections (" + initConn + ") for host: " + servers[i] );
-
- for ( int j = 0; j < initConn; j++ ) {
- SockIO socket = createSocket( servers[i] );
- if ( socket == null ) {
- log.error( "++++ failed to create connection to: " + servers[i] + " -- only " + j + " created." );
- break;
- }
-
- addSocketToPool( availPool, servers[i], socket );
- if ( log.isDebugEnabled() )
- log.debug( "++++ created and added socket: " + socket.toString() + " for host " + servers[i] );
- }
- }
- }
-
- /**
- * Returns state of pool.
- *
+ /**
+ * Returns state of pool.
+ *
* @return true if initialized.
*/
public boolean isInitialized() {
return initialized;
}
- /**
+ /**
* Creates a new SockIO obj for the given server.
*
* If server fails to connect, then return null and do not try
* again until a duration has passed. This duration will grow
- * by doubling after each failed attempt to connect.
- *
+ * by doubling after each failed attempt to connect.
+ *
* @param host host:port to connect to
* @return SockIO obj or null if failed to create
*/
@@ -763,7 +453,7 @@ protected SockIO createSocket( String host ) {
// we do not try to put back in if failback is off
hostDeadLock.lock();
try {
- if ( failover && failback && hostDead.containsKey( host ) && hostDeadDur.containsKey( host ) ) {
+ if ( options.failover && options.failback && hostDead.containsKey( host ) && hostDeadDur.containsKey( host ) ) {
Date store = hostDead.get( host );
long expire = hostDeadDur.get( host ).longValue();
@@ -777,7 +467,7 @@ protected SockIO createSocket( String host ) {
}
try {
- socket = new SockIO( this, host, this.socketTO, this.socketConnectTO, this.nagle );
+ socket = new SockIO( this, host, this.options.socketTO, this.options.socketConnectTO, this.options.nagle );
if ( !socket.isConnected() ) {
log.error( "++++ failed to get SockIO obj for: " + host + " -- new socket is not connected" );
@@ -827,51 +517,64 @@ protected SockIO createSocket( String host ) {
return socket;
}
- /**
- * @param key
- * @return
+ /**
+ * @param key
+ * @return
*/
public String getHost( String key ) {
return getHost( key, null );
}
- /**
- * Gets the host that a particular key / hashcode resides on.
- *
- * @param key
- * @param hashcode
- * @return
+ /**
+ * Gets the host that a particular key / hashcode resides on.
+ *
+ * @param key
+ * @param hashcode
+ * @return
*/
public String getHost( String key, Integer hashcode ) {
- SockIO socket = getSock( key, hashcode );
+ SockIO socket = getSockAndFragmentId( key, hashcode ).sock();
String host = socket.getHost();
socket.close();
return host;
}
- /**
+ public SockAndFragmentId getHostSockAndFragmentId( String host ) {
+ int i = 0;
+ for (Fragment f: config.getFragmentList()) {
+ if (f.getAddress().equals(host)) {
+ return getSockAndFragmentId( null, i );
+ }
+ i++;
+ }
+ return null;
+ }
+
+ /**
* Returns appropriate SockIO object given
* string cache key.
- *
+ *
* @param key hashcode for cache key
* @return SockIO obj connected to server
*/
- public SockIO getSock( String key ) {
- return getSock( key, null );
+ public SockAndFragmentId getSockAndFragmentId( String key ) {
+ return getSockAndFragmentId( key, null );
}
- /**
- * Returns appropriate SockIO object given
- * string cache key and optional hashcode.
+ /**
+ * Returns appropriate SockAndFragmentId object
+ * containing the SockIO object given string cache
+ * key and optional hashcode, and the fragment id
+ * of the fragment containing the host.
*
* Trys to get SockIO from pool. Fails over
* to additional pools in event of server failure.
- *
+ *
* @param key hashcode for cache key
* @param hashCode if not null, then the int hashcode to use
- * @return SockIO obj connected to server
+ * @return SockAndFragmentId obj connected to server
*/
- public SockIO getSock( String key, Integer hashCode ) {
+ public SockAndFragmentId getSockAndFragmentId( String key, Integer hashCode ) {
if ( log.isDebugEnabled() )
log.debug( "cache socket pick " + key + " " + hashCode );
@@ -882,20 +585,16 @@ public SockIO getSock( String key, Integer hashCode ) {
}
// if no servers return null
- if ( ( this.hashingAlg == CONSISTENT_HASH && consistentBuckets.size() == 0 )
- || ( buckets != null && buckets.size() == 0 ) )
+ if ( config != null && config.getFragmentCount() == 0 )
return null;
// if only one server, return it
- if ( ( this.hashingAlg == CONSISTENT_HASH && consistentBuckets.size() == 1 )
- || ( buckets != null && buckets.size() == 1 ) ) {
+ if ( config != null && config.getFragmentCount() == 1 ) {
- SockIO sock = ( this.hashingAlg == CONSISTENT_HASH )
- ? getConnection( consistentBuckets.get( consistentBuckets.firstKey() ) )
- : getConnection( buckets.get( 0 ) );
+ SockIO sock = getConnection( config.getFragment( 0 ).getAddress() );
if ( sock != null && sock.isConnected() ) {
- if ( aliveCheck ) {
+ if ( options.aliveCheck ) {
if ( !sock.isAlive() ) {
sock.close();
try { sock.trueClose(); } catch ( IOException ioe ) { log.error( "failed to close dead socket" ); }
@@ -910,32 +609,33 @@ public SockIO getSock( String key, Integer hashCode ) {
}
}
- return sock;
+ return new SockAndFragmentId(
+ sock, config.getFragment(0).getId(), 1);
}
-
+
// from here on, we are working w/ multiple servers
// keep trying different servers until we find one
// making sure we only try each server one time
- Set tryServers = new HashSet( Arrays.asList( servers ) );
+ Set tryServers = new HashSet( config.getFragmentList() );
// get initial bucket
long bucket = getBucket( key, hashCode );
- String server = ( this.hashingAlg == CONSISTENT_HASH )
- ? consistentBuckets.get( bucket )
- : buckets.get( (int)bucket );
+ Fragment fragment = config.getFragment( (int)bucket );
+ int fragmentNum = ((int)bucket) + 1;
while ( !tryServers.isEmpty() ) {
// try to get socket from bucket
+ String server = fragment.getAddress();
SockIO sock = getConnection( server );
if ( log.isDebugEnabled() )
log.debug( "cache choose " + server + " for " + key );
if ( sock != null && sock.isConnected() ) {
- if ( aliveCheck ) {
+ if ( options.aliveCheck ) {
if ( sock.isAlive() ) {
- return sock;
+ return new SockAndFragmentId(sock, fragment.getId(), fragmentNum);
}
else {
sock.close();
@@ -944,7 +644,7 @@ public SockIO getSock( String key, Integer hashCode ) {
}
}
else {
- return sock;
+ return new SockAndFragmentId(sock, fragment.getId(), fragmentNum);
}
}
else {
@@ -955,29 +655,27 @@ public SockIO getSock( String key, Integer hashCode ) {
}
// if we do not want to failover, then bail here
- if ( !failover )
+ if ( !options.failover )
return null;
// log that we tried
- tryServers.remove( server );
+ tryServers.remove( fragment );
if ( tryServers.isEmpty() )
break;
// if we failed to get a socket from this server
// then we try again by adding an incrementer to the
- // current key and then rehashing
+ // current key and then rehashing
int rehashTries = 0;
- while ( !tryServers.contains( server ) ) {
+ while ( !tryServers.contains( fragment ) ) {
String newKey = String.format( "%s%s", rehashTries, key );
if ( log.isDebugEnabled() )
log.debug( "rehashing with: " + newKey );
bucket = getBucket( newKey, null );
- server = ( this.hashingAlg == CONSISTENT_HASH )
- ? consistentBuckets.get( bucket )
- : buckets.get( (int)bucket );
+ fragment = config.getFragment( (int)bucket );
rehashTries++;
}
@@ -986,13 +684,13 @@ public SockIO getSock( String key, Integer hashCode ) {
return null;
}
- /**
+ /**
* Returns a SockIO object from the pool for the passed in host.
*
* Meant to be called from a more intelligent method
* which handles choosing appropriate server
- * and failover.
- *
+ * and failover.
+ *
* @param host host from which to retrieve object
* @return SockIO object or null if fail to retrieve one
*/
@@ -1044,7 +742,7 @@ public SockIO getConnection( String host ) {
}
}
}
-
+
// create one socket -- let the maint thread take care of creating more
SockIO socket = createSocket( host );
if ( socket != null ) {
@@ -1056,12 +754,12 @@ public SockIO getConnection( String host ) {
return socket;
}
- /**
+ /**
* Adds a socket to a given pool for the given host.
* THIS METHOD IS NOT THREADSAFE, SO BE CAREFUL WHEN USING!
*
- * Internal utility method.
- *
+ * Internal utility method.
+ *
* @param pool pool to add to
* @param host host this socket is connected to
* @param socket socket to add
@@ -1084,12 +782,12 @@ protected void addSocketToPool( Map> pool, String host,
pool.put( host, sockets );
}
- /**
+ /**
* Removes a socket from specified pool for host.
* THIS METHOD IS NOT THREADSAFE, SO BE CAREFUL WHEN USING!
*
- * Internal utility method.
- *
+ * Internal utility method.
+ *
* @param pool pool to remove from
* @param host host pool
* @param socket socket to remove
@@ -1102,11 +800,11 @@ protected void removeSocketFromPool( Map> pool, String h
}
}
- /**
- * Closes and removes all sockets from specified pool for host.
+ /**
+ * Closes and removes all sockets from specified pool for host.
* THIS METHOD IS NOT THREADSAFE, SO BE CAREFUL WHEN USING!
- *
- * Internal utility method.
+ *
+ * Internal utility method.
*
* @param pool pool to clear
* @param host host to clear
@@ -1133,7 +831,7 @@ protected void clearHostFromPool( Map> pool, String host
}
}
- /**
+ /**
* Checks a SockIO object in with the pool.
*
* This will remove SocketIO from busy pool, and optionally
@@ -1167,24 +865,24 @@ private void checkIn( SockIO socket, boolean addToAvail ) {
}
}
- /**
+ /**
* Returns a socket to the avail pool.
*
* This is called from SockIO.close(). Calling this method
* directly without closing the SockIO object first
* will cause an IOException to be thrown.
- *
+ *
* @param socket socket to return
*/
private void checkIn( SockIO socket ) {
checkIn( socket, true );
}
- /**
+ /**
* Closes all sockets in the passed in pool.
*
- * Internal utility method.
- *
+ * Internal utility method.
+ *
* @param pool pool to close
*/
protected void closePool( Map> pool ) {
@@ -1208,7 +906,7 @@ protected void closePool( Map> pool ) {
}
}
- /**
+ /**
* Shuts down the pool.
*
* Cleanly closes all sockets.
@@ -1238,8 +936,6 @@ public void shutDown() {
closePool( busyPool );
availPool = null;
busyPool = null;
- buckets = null;
- consistentBuckets = null;
hostDeadDur = null;
hostDead = null;
maintThread = null;
@@ -1249,7 +945,7 @@ public void shutDown() {
}
}
- /**
+ /**
* Starts the maintenance thread.
*
* This thread will manage the size of the active pool
@@ -1269,12 +965,12 @@ protected void startMaintThread() {
}
else {
maintThread = new MaintThread( this );
- maintThread.setInterval( this.maintSleep );
+ maintThread.setInterval( this.options.maintSleep );
maintThread.start();
}
}
- /**
+ /**
* Stops the maintenance thread.
*/
protected void stopMaintThread() {
@@ -1282,10 +978,10 @@ protected void stopMaintThread() {
maintThread.stopThread();
}
- /**
+ /**
* Runs self maintenance on all internal pools.
*
- * This is typically called by the maintenance thread to manage pool size.
+ * This is typically called by the maintenance thread to manage pool size.
*/
protected void selfMaint() {
if ( log.isDebugEnabled() )
@@ -1306,9 +1002,9 @@ protected void selfMaint() {
log.debug( "++++ Size of avail pool for host (" + host + ") = " + sockets.size() );
// if pool is too small (n < minSpare)
- if ( sockets.size() < minConn ) {
+ if ( sockets.size() < options.minConn ) {
// need to create new sockets
- int need = minConn - sockets.size();
+ int need = options.minConn - sockets.size();
needSockets.put( host, need );
}
}
@@ -1354,9 +1050,9 @@ protected void selfMaint() {
if ( log.isDebugEnabled() )
log.debug( "++++ Size of avail pool for host (" + host + ") = " + sockets.size() );
- if ( sockets.size() > maxConn ) {
+ if ( sockets.size() > options.maxConn ) {
// need to close down some sockets
- int diff = sockets.size() - maxConn;
+ int diff = sockets.size() - options.maxConn;
int needToClose = (diff <= poolMultiplier)
? diff
: (diff) / poolMultiplier;
@@ -1375,7 +1071,7 @@ protected void selfMaint() {
// if past idle time
// then close socket
// and remove from pool
- if ( (expire + maxIdle) < System.currentTimeMillis() ) {
+ if ( (expire + options.maxIdle) < System.currentTimeMillis() ) {
if ( log.isDebugEnabled() )
log.debug( "+++ removing stale entry from pool as it is past its idle timeout and pool is over max spare" );
@@ -1407,7 +1103,7 @@ protected void selfMaint() {
// if past max busy time
// then close socket
// and remove from pool
- if ( (hungTime + maxBusyTime) < System.currentTimeMillis() ) {
+ if ( (hungTime + options.maxBusyTime) < System.currentTimeMillis() ) {
log.error( "+++ removing potentially hung connection from busy pool ... socket in pool for " + (System.currentTimeMillis() - hungTime) + "ms" );
// remove from the busy pool
@@ -1440,10 +1136,10 @@ protected void selfMaint() {
if ( log.isDebugEnabled() )
log.debug( "+++ ending self maintenance." );
}
-
- /**
+
+ /**
* Class which extends thread and handles maintenance of the pool.
- *
+ *
* @author greg whalin
* @version 1.5
*/
@@ -1465,21 +1161,21 @@ protected MaintThread( SockIOPool pool ) {
}
public void setInterval( long interval ) { this.interval = interval; }
-
+
public boolean isRunning() {
return this.running;
}
- /**
+ /**
* sets stop variable
- * and interupts any wait
+ * and interupts any wait
*/
public void stopThread() {
this.stopThread = true;
this.interrupt();
}
- /**
+ /**
* Start the thread.
*/
public void run() {
@@ -1504,12 +1200,12 @@ public void run() {
}
}
- /**
+ /**
* MemCached client for Java, utility class for Socket IO.
*
* This class is a wrapper around a Socket and its streams.
*
- * @author greg whalin
+ * @author greg whalin
* @author Richard 'toast' Russo
* @version 1.5
*/
@@ -1529,10 +1225,10 @@ public static class SockIO implements LineInputStream {
private DataInputStream in;
private BufferedOutputStream out;
- /**
+ /**
* creates a new SockIO object wrapping a socket
* connection to host:port, and its input and output streams
- *
+ *
* @param pool Pool this object is tied to
* @param host host to connect to
* @param port port to connect to
@@ -1548,7 +1244,7 @@ public SockIO( SockIOPool pool, String host, int port, int timeout, int connectT
// get a socket channel
sock = getSocket( host, port, connectTimeout );
-
+
if ( timeout >= 0 )
sock.setSoTimeout( timeout );
@@ -1562,10 +1258,10 @@ public SockIO( SockIOPool pool, String host, int port, int timeout, int connectT
this.host = host + ":" + port;
}
- /**
+ /**
* creates a new SockIO object wrapping a socket
* connection to host:port, and its input and output streams
- *
+ *
* @param host hostname:port
* @param timeout read timeout value for connected socket
* @param connectTimeout timeout for initial connections
@@ -1595,7 +1291,7 @@ public SockIO( SockIOPool pool, String host, int timeout, int connectTimeout, bo
this.host = host;
}
- /**
+ /**
* Method which gets a connection from SocketChannel.
*
* @param host host to establish connection to
@@ -1611,22 +1307,22 @@ protected static Socket getSocket( String host, int port, int timeout ) throws I
return sock.socket();
}
- /**
- * Lets caller get access to underlying channel.
- *
+ /**
+ * Lets caller get access to underlying channel.
+ *
* @return the backing SocketChannel
*/
public SocketChannel getChannel() { return sock.getChannel(); }
- /**
- * returns the host this socket is connected to
- *
+ /**
+ * returns the host this socket is connected to
+ *
* @return String representation of host (hostname:port)
*/
public String getHost() { return this.host; }
- /**
- * closes socket and all streams connected to it
+ /**
+ * closes socket and all streams connected to it
*
* @throws IOException if fails to close streams or socket
*/
@@ -1634,8 +1330,8 @@ public void trueClose() throws IOException {
trueClose( true );
}
- /**
- * closes socket and all streams connected to it
+ /**
+ * closes socket and all streams connected to it
*
* @throws IOException if fails to close streams or socket
*/
@@ -1697,7 +1393,7 @@ public void trueClose( boolean addToDeadPool ) throws IOException {
throw new IOException( errMsg.toString() );
}
- /**
+ /**
* sets closed flag and checks in to connection pool
* but does not close connections
*/
@@ -1707,10 +1403,10 @@ void close() {
log.debug("++++ marking socket (" + this.toString() + ") as closed and available to return to avail pool");
pool.checkIn( this );
}
-
- /**
- * checks if the connection is open
- *
+
+ /**
+ * checks if the connection is open
+ *
* @return true if connected
*/
boolean isConnected() {
@@ -1740,10 +1436,10 @@ boolean isAlive() {
return true;
}
- /**
+ /**
* reads a line
- * intentionally not using the deprecated readLine method from DataInputStream
- *
+ * intentionally not using the deprecated readLine method from DataInputStream
+ *
* @return String that was read in
* @throws IOException if io problems during read
*/
@@ -1783,9 +1479,9 @@ public String readLine() throws IOException {
return bos.toString().trim();
}
- /**
- * reads up to end of line and returns nothing
- *
+ /**
+ * reads up to end of line and returns nothing
+ *
* @throws IOException if io problems during read
*/
public void clearEOL() throws IOException {
@@ -1814,9 +1510,9 @@ public void clearEOL() throws IOException {
}
}
- /**
+ /**
* reads length bytes into the passed in byte array from dtream
- *
+ *
* @param b byte array
* @throws IOException if io problems during read
*/
@@ -1835,9 +1531,9 @@ public int read( byte[] b ) throws IOException {
return count;
}
- /**
- * flushes output stream
- *
+ /**
+ * flushes output stream
+ *
* @throws IOException if io problems during read
*/
void flush() throws IOException {
@@ -1847,10 +1543,10 @@ void flush() throws IOException {
}
out.flush();
}
-
- /**
+
+ /**
* writes a byte array to the output stream
- *
+ *
* @param b byte array to write
* @throws IOException if an io error happens
*/
@@ -1862,27 +1558,27 @@ void write( byte[] b ) throws IOException {
out.write( b );
}
- /**
+ /**
* use the sockets hashcode for this object
- * so we can key off of SockIOs
- *
+ * so we can key off of SockIOs
+ *
* @return int hashcode
*/
public int hashCode() {
return ( sock == null ) ? 0 : sock.hashCode();
}
- /**
- * returns the string representation of this socket
- *
+ /**
+ * returns the string representation of this socket
+ *
* @return string
*/
public String toString() {
return ( sock == null ) ? "" : sock.toString();
}
- /**
- * Hack to reap any leaking children.
+ /**
+ * Hack to reap any leaking children.
*/
protected void finalize() throws Throwable {
try {
@@ -1900,4 +1596,31 @@ protected void finalize() throws Throwable {
}
}
}
+
+ public static class SockAndFragmentId {
+ // The socket.
+ private SockIO sock;
+ // The fragment id of the fragment that the socket belongs to.
+ private int fragmentId;
+ // The fragment number (index in the fragments array + 1).
+ private int fragmentNum;
+
+ SockAndFragmentId(SockIO sock, int id, int fragmentNum) {
+ this.sock = sock;
+ this.fragmentId = id;
+ this.fragmentNum = fragmentNum;
+ }
+
+ public SockIO sock() {
+ return sock;
+ }
+
+ public int fragmentId() {
+ return fragmentId;
+ }
+
+ public int fragmentNum() {
+ return fragmentNum;
+ }
+ }
}
diff --git a/src/main/java/edu/usc/cs550/rejig/client/configreader/RejigConfigReader.java b/src/main/java/edu/usc/cs550/rejig/client/configreader/RejigConfigReader.java
new file mode 100644
index 0000000..ebf9fa0
--- /dev/null
+++ b/src/main/java/edu/usc/cs550/rejig/client/configreader/RejigConfigReader.java
@@ -0,0 +1,11 @@
+package edu.usc.cs550.rejig.client.configreader;
+
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
+
+/**
+ * Implement this interface to manage reading the RegjigConfig
+ * object from a local, or remote store.
+ */
+public interface RejigConfigReader {
+ RejigConfig getConfig();
+}
\ No newline at end of file
diff --git a/src/com/meetup/memcached/test/MemcachedBench.java b/src/test/java/edu/usc/cs550/rejig/client/test/MemcachedBench.java
similarity index 54%
rename from src/com/meetup/memcached/test/MemcachedBench.java
rename to src/test/java/edu/usc/cs550/rejig/client/test/MemcachedBench.java
index 9f0491f..05a79ad 100644
--- a/src/com/meetup/memcached/test/MemcachedBench.java
+++ b/src/test/java/edu/usc/cs550/rejig/client/test/MemcachedBench.java
@@ -1,23 +1,26 @@
-/**
- * Copyright (c) 2008 Greg Whalin
- * All rights reserved.
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the BSD license
- *
- * This library is distributed in the hope that it will be
- * useful, but WITHOUT ANY WARRANTY; without even the implied
- * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
- * PURPOSE.
- *
- * You should have received a copy of the BSD License along with this
- * library.
- *
- * @author Greg Whalin
- */
-package com.meetup.memcached.test;
-
-import com.meetup.memcached.*;
+// /**
+// * Copyright (c) 2008 Greg Whalin
+// * All rights reserved.
+// *
+// * This library is free software; you can redistribute it and/or
+// * modify it under the terms of the BSD license
+// *
+// * This library is distributed in the hope that it will be
+// * useful, but WITHOUT ANY WARRANTY; without even the implied
+// * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+// * PURPOSE.
+// *
+// * You should have received a copy of the BSD License along with this
+// * library.
+// *
+// * @author Greg Whalin
+// */
+package edu.usc.cs550.rejig.client.test;
+
+import edu.usc.cs550.rejig.client.*;
+import edu.usc.cs550.rejig.interfaces.Fragment;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
+
import java.util.*;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
@@ -37,24 +40,34 @@ public static void main(String[] args) {
int runs = Integer.parseInt(args[0]);
int start = Integer.parseInt(args[1]);
- String[] serverlist = { "192.168.1.50:1624" };
-
- // initialize the pool for memcache servers
- SockIOPool pool = SockIOPool.getInstance( "test" );
- pool.setServers(serverlist);
+ String[] servers = { "localhost:11211" };
- pool.setInitConn( 100 );
- pool.setMinConn( 100 );
- pool.setMaxConn( 500 );
- pool.setMaintSleep( 20 );
-
- pool.setNagle( false );
- pool.initialize();
+ // initialize the pool options.
+ SockIOPool.SockIOPoolOptions options = new SockIOPool.SockIOPoolOptions();
+ options.initConn = 100;
+ options.minConn = 100;
+ options.maxConn = 500;
+ options.maintSleep = 20 ;
+ options.nagle = false;
// get client instance
- MemcachedClient mc = new MemcachedClient( "test" );
+ MockRejigConfigReader configReader = new MockRejigConfigReader();
+ configReader.setConfig(RejigConfig.newBuilder()
+ .setId(1)
+ .addFragment(Fragment.newBuilder()
+ .setId(1)
+ .setAddress(servers[0])
+ .build()
+ ).build());
+ MemcachedClient mc = new MemcachedClient(
+ null, new MockErrorHandler(),
+ "test", configReader, options);
mc.setCompressEnable( false );
+ Date expiry = new Date(System.currentTimeMillis() * 10*60*100);
+ mc.setConfig(configReader.getConfig(), null, servers[0]);
+ mc.grantLease(1, expiry, servers[0]);
+
String keyBase = "testKey";
String object = "This is a test of an object blah blah es, serialization does not seem to slow things down so much. The gzip compression is horrible horrible performance, so we only use it for very large objects. I have not done any heavy benchmarking recently";
@@ -94,6 +107,6 @@ public static void main(String[] args) {
time = end - begin;
System.out.println(runs + " deletes: " + time + "ms");
- SockIOPool.getInstance( "test" ).shutDown();
+ mc.shutDown();
}
}
diff --git a/src/com/meetup/memcached/test/MemcachedTest.java b/src/test/java/edu/usc/cs550/rejig/client/test/MemcachedTest.java
similarity index 64%
rename from src/com/meetup/memcached/test/MemcachedTest.java
rename to src/test/java/edu/usc/cs550/rejig/client/test/MemcachedTest.java
index f695e96..b845c5e 100644
--- a/src/com/meetup/memcached/test/MemcachedTest.java
+++ b/src/test/java/edu/usc/cs550/rejig/client/test/MemcachedTest.java
@@ -13,11 +13,14 @@
* You should have received a copy of the BSD License along with this
* library.
*
- * @author Greg Whalin
+ * @author Greg Whalin
*/
-package com.meetup.memcached.test;
+package edu.usc.cs550.rejig.client.test;
+
+import edu.usc.cs550.rejig.client.*;
+import edu.usc.cs550.rejig.interfaces.Fragment;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
-import com.meetup.memcached.*;
import java.util.*;
public class MemcachedTest {
@@ -25,50 +28,73 @@ public class MemcachedTest {
// store results from threads
private static Hashtable threadInfo =
new Hashtable();
-
+
/**
* This runs through some simple tests of the MemcacheClient.
*
* Command line args:
* args[0] = number of threads to spawn
* args[1] = number of runs per thread
- * args[2] = size of object to store
+ * args[2] = size of object to store
*
* @param args the command line arguments
*/
public static void main(String[] args) {
- String[] serverlist = { "cache1.int.meetup.com:12345", "cache0.int.meetup.com:12345" };
-
- // initialize the pool for memcache servers
- SockIOPool pool = SockIOPool.getInstance();
- pool.setServers( serverlist );
-
- pool.setInitConn(5);
- pool.setMinConn(5);
- pool.setMaxConn(50);
- pool.setMaintSleep(30);
+ String[] servers = { "localhost:11211", "localhost:11212" };
- pool.setNagle(false);
- pool.initialize();
+ // initialize the pool options
+ SockIOPool.SockIOPoolOptions options = new SockIOPool.SockIOPoolOptions();
+ options.initConn = 5;
+ options.minConn = 5;
+ options.maxConn = 50;
+ options.maintSleep = 30;
+ options.nagle = false;
int threads = Integer.parseInt(args[0]);
int runs = Integer.parseInt(args[1]);
int size = 1024 * Integer.parseInt(args[2]); // how many kilobytes
+ // set RejigConfig
+ MockRejigConfigReader configReader = new MockRejigConfigReader();
+ configReader.setConfig(RejigConfig.newBuilder()
+ .setId(1)
+ .addFragment(Fragment.newBuilder()
+ .setId(1)
+ .setAddress(servers[0])
+ .build()
+ ).addFragment(Fragment.newBuilder()
+ .setId(1)
+ .setAddress(servers[1])
+ .build()
+ ).build());
+
+ // get client instance
+ MemcachedClient mc = new MemcachedClient(
+ null, new MockErrorHandler(),
+ "test", configReader, options);
+ mc.setCompressEnable( false );
+ mc.setCompressThreshold(0);
+
+ Date expiry = new Date(System.currentTimeMillis() * 10*60*100);
+ mc.setConfig(configReader.getConfig(), null, servers[0]);
+ mc.grantLease(1, expiry, servers[0]);
+ mc.setConfig(configReader.getConfig(), null, servers[1]);
+ mc.grantLease(2, expiry, servers[1]);
+
// get object to store
int[] obj = new int[size];
for (int i = 0; i < size; i++) {
obj[i] = i;
}
- String[] keys = new String[size];
- for (int i = 0; i < size; i++) {
+ String[] keys = new String[runs];
+ for (int i = 0; i < runs; i++) {
keys[i] = "test_key" + i;
}
for (int i = 0; i < threads; i++) {
- bench b = new bench(runs, i, obj, keys);
+ bench b = new bench(mc, runs, i, obj, keys);
b.start();
}
@@ -88,12 +114,11 @@ public static void main(String[] args) {
}
}
- pool.shutDown();
- System.exit(1);
+ mc.shutDown();
}
- /**
- * Test code per thread.
+ /**
+ * Test code per thread.
*/
private static class bench extends Thread {
private int runs;
@@ -101,24 +126,21 @@ private static class bench extends Thread {
private int[] object;
private String[] keys;
private int size;
+ private MemcachedClient mc;
- public bench(int runs, int threadNum, int[] object, String[] keys) {
+ public bench(MemcachedClient client, int runs, int threadNum, int[] object, String[] keys) {
this.runs = runs;
this.threadNum = threadNum;
this.object = object;
this.keys = keys;
this.size = object.length;
+ this.mc = client;
}
public void run() {
StringBuilder result = new StringBuilder();
- // get client instance
- MemcachedClient mc = new MemcachedClient();
- mc.setCompressEnable(false);
- mc.setCompressThreshold(0);
-
// time deletes
long start = System.currentTimeMillis();
for (int i = 0; i < runs; i++) {
diff --git a/src/test/java/edu/usc/cs550/rejig/client/test/MockErrorHandler.java b/src/test/java/edu/usc/cs550/rejig/client/test/MockErrorHandler.java
new file mode 100644
index 0000000..d6b3cf6
--- /dev/null
+++ b/src/test/java/edu/usc/cs550/rejig/client/test/MockErrorHandler.java
@@ -0,0 +1,66 @@
+package edu.usc.cs550.rejig.client.test;
+
+import edu.usc.cs550.rejig.client.ErrorHandler;
+import edu.usc.cs550.rejig.client.MemcachedClient;
+
+class MockErrorHandler implements ErrorHandler {
+
+ public void handleErrorOnInit(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnGet(
+ final MemcachedClient client, final Throwable error,
+ final String cacheKey) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnGet(
+ final MemcachedClient client, final Throwable error,
+ final String[] cacheKeys) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnSet(
+ final MemcachedClient client, final Throwable error,
+ final String cacheKey) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnDelete(
+ final MemcachedClient client, final Throwable error,
+ final String cacheKey) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnFlush(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnStats(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnConf(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnGrantLease(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnRevokeLease(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+
+ public void handleErrorOnRefreshAndRetry(
+ final MemcachedClient client, final Throwable error) {
+ throw new RuntimeException(error);
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/edu/usc/cs550/rejig/client/test/MockRejigConfigReader.java b/src/test/java/edu/usc/cs550/rejig/client/test/MockRejigConfigReader.java
new file mode 100644
index 0000000..0962f05
--- /dev/null
+++ b/src/test/java/edu/usc/cs550/rejig/client/test/MockRejigConfigReader.java
@@ -0,0 +1,17 @@
+package edu.usc.cs550.rejig.client.test;
+
+import edu.usc.cs550.rejig.client.configreader.RejigConfigReader;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
+
+class MockRejigConfigReader implements RejigConfigReader {
+ private RejigConfig config;
+
+ public RejigConfig getConfig() {
+ return this.config;
+ }
+
+ public MockRejigConfigReader setConfig(RejigConfig config) {
+ this.config = config;
+ return this;
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/edu/usc/cs550/rejig/client/test/TestMemcached.java b/src/test/java/edu/usc/cs550/rejig/client/test/TestMemcached.java
new file mode 100644
index 0000000..058a607
--- /dev/null
+++ b/src/test/java/edu/usc/cs550/rejig/client/test/TestMemcached.java
@@ -0,0 +1,90 @@
+/**
+ * Copyright (c) 2008 Greg Whalin
+ * All rights reserved.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the BSD license
+ *
+ * This library is distributed in the hope that it will be
+ * useful, but WITHOUT ANY WARRANTY; without even the implied
+ * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.
+ *
+ * You should have received a copy of the BSD License along with this
+ * library.
+ *
+ * @author greg whalin
+ */
+package edu.usc.cs550.rejig.client.test;
+
+import edu.usc.cs550.rejig.client.*;
+import edu.usc.cs550.rejig.interfaces.Fragment;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
+
+import org.apache.log4j.*;
+
+import java.util.Date;
+
+public class TestMemcached {
+ public static void main(String[] args) {
+ // memcached should be running on port 11211 but NOT on 11212
+
+ BasicConfigurator.configure();
+ String[] servers = { "localhost:11211", "localhost:11212" };
+ // initialize the pool options
+ SockIOPool.SockIOPoolOptions options = new SockIOPool.SockIOPoolOptions();
+ options.failover = true;
+ options.initConn = 10;
+ options.minConn = 5;
+ options.maxConn = 250;
+ options.maintSleep = 30;
+ options.nagle = false;
+ options.socketTO = 3000;
+ options.aliveCheck = true;
+
+ // set RejigConfig
+ MockRejigConfigReader configReader = new MockRejigConfigReader();
+ configReader.setConfig(RejigConfig.newBuilder()
+ .setId(1)
+ .addFragment(Fragment.newBuilder()
+ .setId(1)
+ .setAddress(servers[0])
+ .build()
+ ).addFragment(Fragment.newBuilder()
+ .setId(1)
+ .setAddress(servers[1])
+ .build()
+ ).build());
+
+ // get client instance
+ MemcachedClient mcc = new MemcachedClient(
+ null, new MockErrorHandler(),
+ "test", configReader, options);
+
+ Date expiry = new Date(System.currentTimeMillis() * 10*60*100);
+ mcc.setConfig(configReader.getConfig(), null, servers[0]);
+ mcc.grantLease(1, expiry, servers[0]);
+ mcc.setConfig(configReader.getConfig(), null, servers[1]);
+ mcc.grantLease(2, expiry, servers[1]);
+
+ // turn off most memcached client logging:
+ edu.usc.cs550.rejig.client.Logger.getLogger( MemcachedClient.class.getName() ).setLevel( edu.usc.cs550.rejig.client.Logger.LEVEL_WARN );
+
+ for ( int i = 0; i < 10; i++ ) {
+ boolean success = mcc.set( "" + i, "Hello!" );
+ String result = (String)mcc.get( "" + i );
+ System.out.println( String.format( "set( %d ): %s", i, success ) );
+ System.out.println( String.format( "get( %d ): %s", i, result ) );
+ }
+
+ System.out.println( "\n\t -- sleeping --\n" );
+ try { Thread.sleep( 10000 ); } catch ( Exception ex ) { }
+
+ for ( int i = 0; i < 10; i++ ) {
+ boolean success = mcc.set( "" + i, "Hello!" );
+ String result = (String)mcc.get( "" + i );
+ System.out.println( String.format( "set( %d ): %s", i, success ) );
+ System.out.println( String.format( "get( %d ): %s", i, result ) );
+ }
+ }
+}
diff --git a/src/test/java/edu/usc/cs550/rejig/client/test/UnitTests.java b/src/test/java/edu/usc/cs550/rejig/client/test/UnitTests.java
new file mode 100644
index 0000000..5b97209
--- /dev/null
+++ b/src/test/java/edu/usc/cs550/rejig/client/test/UnitTests.java
@@ -0,0 +1,531 @@
+/**
+ * Copyright (c) 2008 Greg Whalin
+ * All rights reserved.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the BSD license
+ *
+ * This library is distributed in the hope that it will be
+ * useful, but WITHOUT ANY WARRANTY; without even the implied
+ * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.
+ *
+ * You should have received a copy of the BSD License along with this
+ * library.
+ *
+ * @author Kevin Burton
+ * @author greg whalin
+ */
+package edu.usc.cs550.rejig.client.test;
+
+import edu.usc.cs550.rejig.client.*;
+import edu.usc.cs550.rejig.interfaces.Fragment;
+import edu.usc.cs550.rejig.interfaces.RejigConfig;
+
+import java.util.*;
+import java.io.Serializable;
+
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.BasicConfigurator;
+
+public class UnitTests {
+
+ // logger
+ private static Logger log =
+ Logger.getLogger( UnitTests.class.getName() );
+
+ public static MemcachedClient mc = null;
+
+ public static void test1() {
+ mc.set( "foo", Boolean.TRUE );
+ Boolean b = (Boolean)mc.get( "foo" );
+ assertion(
+ b.booleanValue(),
+ "+ store/retrieve Boolean type test failed"
+ );
+ }
+
+ public static void test2() {
+ mc.set( "foo", new Integer( Integer.MAX_VALUE ) );
+ Integer i = (Integer)mc.get( "foo" );
+ assertion(
+ i.intValue() == Integer.MAX_VALUE,
+ "+ store/retrieve Integer type test failed"
+ );
+ }
+
+ public static void test3() {
+ String input = "test of string encoding";
+ mc.set( "foo", input );
+ String s = (String)mc.get( "foo" );
+ assertion(
+ s.equals( input ),
+ "+ store/retrieve String type test failed"
+ );
+ }
+
+ public static void test4() {
+ mc.set( "foo", new Character( 'z' ) );
+ Character c = (Character)mc.get( "foo" );
+ assertion(
+ c.charValue() == 'z',
+ "+ store/retrieve Character type test failed"
+ );
+ }
+
+ public static void test5() {
+ mc.set( "foo", new Byte( (byte)127 ) );
+ Byte b = (Byte)mc.get( "foo" );
+ assertion(
+ b.byteValue() == 127,
+ "+ store/retrieve Byte type test failed"
+ );
+ }
+
+ public static void test6() {
+ mc.set( "foo", new StringBuffer( "hello" ) );
+ StringBuffer o = (StringBuffer)mc.get( "foo" );
+ assertion(
+ o.toString().equals( "hello" ),
+ "+ store/retrieve StringBuffer type test failed"
+ );
+ }
+
+ public static void test7() {
+ mc.set( "foo", new Short( (short)100 ) );
+ Short o = (Short)mc.get( "foo" );
+ assertion(
+ o.shortValue() == 100,
+ "+ store/retrieve Short type test failed"
+ );
+ }
+
+ public static void test8() {
+ mc.set( "foo", new Long( Long.MAX_VALUE ) );
+ Long o = (Long)mc.get( "foo" );
+ assertion(
+ o.longValue() == Long.MAX_VALUE,
+ "+ store/retrieve Long type test failed"
+ );
+ }
+
+ public static void test9() {
+ mc.set( "foo", new Double( 1.1 ) );
+ Double o = (Double)mc.get( "foo" );
+ assertion(
+ o.doubleValue() == 1.1,
+ "+ store/retrieve Double type test failed"
+ );
+ }
+
+ public static void test10() {
+ mc.set( "foo", new Float( 1.1f ) );
+ Float o = (Float)mc.get( "foo" );
+ assertion(
+ o.floatValue() == 1.1f,
+ "+ store/retrieve Float type test failed"
+ );
+ }
+
+ public static void test11() {
+ mc.set( "foo", new Integer( 100 ), new Date( System.currentTimeMillis() ));
+ try { Thread.sleep( 1000 ); } catch ( Exception ex ) { }
+ assertion(
+ mc.get( "foo" ) == null,
+ "+ store/retrieve w/ expiration test failed"
+ );
+ }
+
+ public static void test12() {
+ long i = 0;
+ mc.storeCounter("foo", i);
+ mc.incr("foo"); // foo now == 1
+ mc.incr("foo", (long)5); // foo now == 6
+ long j = mc.decr("foo", (long)2); // foo now == 4
+ assertion(
+ j == 4 && j == mc.getCounter( "foo" ),
+ "+ incr/decr test failed"
+ );
+ }
+
+ public static void test13() {
+ Date d1 = new Date();
+ mc.set("foo", d1);
+ Date d2 = (Date) mc.get("foo");
+ assertion(
+ d1.equals( d2 ),
+ "+ store/retrieve Date type test failed"
+ );
+ }
+
+ public static void test14() {
+ assertion(
+ !mc.keyExists( "foobar123" ),
+ "+ store/retrieve test failed"
+ );
+ mc.set( "foobar123", new Integer( 100000) );
+ assertion(
+ mc.keyExists( "foobar123" ),
+ "+ store/retrieve test failed"
+ );
+
+ assertion(
+ !mc.keyExists( "counterTest123" ),
+ "+ store/retrieve test failed"
+ );
+ mc.storeCounter( "counterTest123", 0 );
+ assertion(
+ mc.keyExists( "counterTest123" ),
+ "+ counter store test failed"
+ );
+ }
+
+ public static void test15() {
+
+ Map stats = mc.statsItems();
+ assertion(stats != null, "+ stats test failed");
+
+ stats = mc.statsSlabs();
+ assertion(stats != null, "+ stats test failed");
+ }
+
+ public static void test16() {
+ assertion(
+ !mc.set( "foo", null ),
+ "+ invalid data store [null] test failed"
+ );
+ }
+
+ public static void test17() {
+ mc.set( "foo bar", Boolean.TRUE );
+ Boolean b = (Boolean)mc.get( "foo bar" );
+ assertion(
+ b.booleanValue(),
+ "+ store/retrieve Boolean type test failed"
+ );
+ }
+
+ public static void test18() {
+ long i = 0;
+ mc.addOrIncr( "foo" ); // foo now == 0
+ mc.incr( "foo" ); // foo now == 1
+ mc.incr( "foo", (long)5 ); // foo now == 6
+
+ mc.addOrIncr( "foo" ); // foo now 7
+
+ long j = mc.decr( "foo", (long)3 ); // foo now == 4
+ assertion(
+ j == 4 && j == mc.getCounter( "foo" ),
+ "+ incr/decr test failed"
+ );
+ }
+
+ public static void test19() {
+ int max = 100;
+ String[] keys = new String[ max ];
+ for ( int i=0; i results = mc.getMulti( keys );
+ for ( int i=0; i results = mc.getMulti( keys );
+ for ( int i=0; i results = mc.getMulti( allKeys );
+
+ assertion(
+ allKeys.length == results.size(),
+ "+ getMulti w/ keys that don't exist test failed"
+ );
+ for ( String key : setKeys ) {
+ String val = (String)results.get( key );
+ assertion(
+ key.equals( val ),
+ "+ getMulti w/ keys that don't exist test failed"
+ );
+ }
+ }
+
+ // Sets the config object into the client and grant a
+ // lease to all fragments for 10 mins.
+ public static void setup(RejigConfig config) {
+ boolean b = mc.setConfig( config, null );
+ assertion( b, "+ setting config failed" );
+
+ Date expiry = new Date(System.currentTimeMillis() + 10*60*1000);
+ int fragmentNum = 1;
+ for ( Fragment f : config.getFragmentList() ) {
+ boolean g = mc.grantLease( fragmentNum, expiry, f.getAddress() );
+ assertion( g, "+ granting lease failed: Fragment num: " + fragmentNum );
+ fragmentNum++;
+ }
+ }
+
+ // Revoke all leases granted.
+ private static void cleanup(RejigConfig config) {
+ int fragmentNum = 1;
+ for ( Fragment f : config.getFragmentList() ) {
+ boolean g = mc.revokeLease( fragmentNum, f.getAddress() );
+ assertion( g, "+ revoking lease failed. Fragment num: " + fragmentNum );
+ fragmentNum++;
+ }
+ }
+
+ private static void assertion(boolean condition, String errorMessage) {
+ if (!condition) {
+ throw new AssertionError(errorMessage);
+ }
+ }
+
+ public static void runAlTests( MemcachedClient mc, boolean run14 ) {
+ if (run14) {
+ test14();
+ }
+ for ( int t = 0; t < 2; t++ ) {
+ mc.setCompressEnable( ( t&1 ) == 1 );
+
+ test1();
+ test2();
+ test3();
+ test4();
+ test5();
+ test6();
+ test7();
+ test8();
+ test9();
+ test10();
+ test11();
+ test12();
+ test13();
+ test15();
+ test16();
+ test17();
+ test21();
+ test22();
+ test23();
+ test24();
+
+ for ( int i = 0; i < 3; i++ )
+ test19();
+
+ test20( 8191, 1, 0 );
+ test20( 8192, 1, 0 );
+ test20( 8193, 1, 0 );
+
+ test20( 16384, 100, 0 );
+ test20( 17000, 128, 0 );
+
+ test20( 128*1024, 1023, 0 );
+ test20( 128*1024, 1023, 1 );
+ test20( 128*1024, 1024, 0 );
+ test20( 128*1024, 1024, 1 );
+
+ test20( 128*1024, 1023, 0 );
+ test20( 128*1024, 1023, 1 );
+ test20( 128*1024, 1024, 0 );
+ test20( 128*1024, 1024, 1 );
+
+ test20( 900*1024, 32*1024, 0 );
+ test20( 900*1024, 32*1024, 1 );
+ }
+
+ }
+
+ /**
+ * This runs through some simple tests of the MemcacheClient.
+ */
+ public static void main(String[] args) {
+
+ BasicConfigurator.configure();
+ org.apache.log4j.Logger.getRootLogger().setLevel( Level.WARN );
+
+ String[] serverlist = {
+ "localhost:11210",
+ "localhost:11211",
+ "localhost:11212",
+ "localhost:11213",
+ "localhost:11214",
+ "localhost:11215",
+ "localhost:11216",
+ "localhost:11217",
+ "localhost:11218",
+ "localhost:11219"
+ };
+
+ // initialize the pool options
+ SockIOPool.SockIOPoolOptions options = new SockIOPool.SockIOPoolOptions();
+ options.maxConn = 250;
+ options.maintSleep = 30;
+ options.nagle = false;
+ options.hashingAlg = SockIOPool.FragmentHashingAlgo.NATIVE_HASH;
+
+ // set RejigConfig
+ RejigConfig config = createConfig1(serverlist);
+ MockRejigConfigReader configReader = new MockRejigConfigReader();
+ configReader.setConfig(config);
+
+ // get client instance
+ mc = new MemcachedClient(null, new MockErrorHandler(),
+ "test", configReader, options);
+
+ // run tests.
+ mc.flushAll();
+ System.out.println("Running tests.");
+ setup(config);
+ runAlTests( mc, true );
+
+ // change the config.
+ System.out.println("Re-running with config change.");
+ setConfig2(config, configReader, mc);
+
+ // run tests again
+ runAlTests( mc, false );
+
+ cleanup(configReader.getConfig());
+ mc.flushAll();
+ // shutdown
+ mc.shutDown();
+ }
+
+ private static RejigConfig createConfig1(String[] serverlist) {
+ Integer[] weights = { 1, 1, 1, 1, 10, 5, 1, 1, 1, 3 };
+ RejigConfig.Builder builder = RejigConfig.newBuilder()
+ .setId(1);
+ for (int i = 0; i < weights.length; i++) {
+ int weight = weights[i];
+ for (int j = 0; j < weight; j++) {
+ builder.addFragment(Fragment.newBuilder()
+ .setId(1)
+ .setAddress(serverlist[i])
+ .build()
+ );
+ }
+ }
+ return builder.build();
+ }
+
+ private static void setConfig2(RejigConfig curr, MockRejigConfigReader configReader, MemcachedClient mc) {
+ RejigConfig.Builder builder = curr.toBuilder().setId(2);
+ for (int i = 9; i < 14; i++) {
+ builder.setFragment(i, Fragment.newBuilder()
+ .setId(2)
+ .setAddress("localhost:11215")
+ .build()
+ );
+ mc.revokeLease(i + 1, "localhost:11214");
+ mc.grantLease(i + 1, new Date(System.currentTimeMillis() + 10*60*1000), "localhost:11215");
+ }
+ RejigConfig config = builder.build();
+ configReader.setConfig(config);
+ // Update impacted server.
+ mc.setConfig(config, null, "localhost:11214");
+ }
+
+ /**
+ * Class for testing serializing of objects.
+ *
+ * @author $Author: $
+ * @version $Revision: $ $Date: $
+ */
+ public static final class TestClass implements Serializable {
+
+ private String field1;
+ private String field2;
+ private Integer field3;
+
+ public TestClass( String field1, String field2, Integer field3 ) {
+ this.field1 = field1;
+ this.field2 = field2;
+ this.field3 = field3;
+ }
+
+ public String getField1() { return this.field1; }
+ public String getField2() { return this.field2; }
+ public Integer getField3() { return this.field3; }
+
+ public boolean equals( Object o ) {
+ if ( this == o ) return true;
+ if ( !( o instanceof TestClass ) ) return false;
+
+ TestClass obj = (TestClass)o;
+
+ return ( ( this.field1 == obj.getField1() || ( this.field1 != null && this.field1.equals( obj.getField1() ) ) )
+ && ( this.field2 == obj.getField2() || ( this.field2 != null && this.field2.equals( obj.getField2() ) ) )
+ && ( this.field3 == obj.getField3() || ( this.field3 != null && this.field3.equals( obj.getField3() ) ) ) );
+ }
+ }
+}