p49023857
/
r-memcached
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Message in Memcached Client

This commit is contained in:
lyr90329 2014-08-07 18:53:28 +08:00
parent f0f9797c54
commit 508214066f
34 changed files with 0 additions and 4373 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
MemcachedClient/Test New Branch.txt
View File

0
src/messageBody/memcachedmsg.java → MemcachedClient/src/messageBody/memcachedmsg.java
View File

0
src/messageBody/requestMsg.java → MemcachedClient/src/messageBody/requestMsg.java
View File

79
protobuf/memcached.proto
View File

@ -1,79 +0,0 @@
option java_package = "messageBody";
option java_outer_classname="memcachedmsg";
message Test1
{
optional string key=1;//string delegate String
}
message messageTest//内部类名
{
//optional int32 a = 1; 中1这样的序列号在一个类不能重复
optional int32 a = 1;//optional代表可选 一般写这个 int32代表java的int
optional int64 b = 2;//int64代表java的Long
optional string key=3;//string 代表String
optional bool d = 4;//bool代表boolean
repeated int32 e=5;//代表 List<Integer>
optional Test1 f=6;// 嵌套对象
}
message nm_Connected
{
optional int32 num = 1;
}
message nm_Connected_mem_back
{
optional int32 num = 1;
}
message nm_Connected_web_back
{
optional int32 num = 1;
}
message nm_read
{
optional string key=1;
optional string value=2;
optional int64 time=3;
}
message nm_read_recovery
{
optional string key=1;
optional string value=2;
optional int64 time=3;
}
message nm_write_1
{
optional string key=1;
optional string value=2;
optional int64 time=3;
optional int32 memID=4;
}
message nm_write_1_res
{
optional string key=1;
optional string value=2;
optional int64 time=3;
optional int32 memID=4;
}
message nm_write_2
{
optional string key=1;
optional string value=2;
optional int64 time=3;
optional int32 memID=4;
}
message nm_write_2_res
{
optional string key=1;
optional string value=2;
optional int64 time=3;
optional int32 memID=4;
}

42
protobuf/requestMsg.proto
View File

@ -1,42 +0,0 @@
option java_package = "messageBody";//°üÃû
option java_outer_classname="requestMsg"; //ˈ̟
// Á¬½Ómemcached server
message nr_Connected_mem
{
}
message nr_Connected_mem_back
{
optional int32 memID=1;
}
message nr_Read
{
optional string key=1;
optional int32 clientid=2;
optional int64 time=3;
}
message nr_Read_res
{
optional string key=1;
optional string value=2;
optional int64 time=3;
}
message nr_write
{
optional string key=1;
optional string value=2;
optional int64 time=3;
}
message nr_write_res
{
optional string key=1;
optional string value=2;
optional int64 time=3;
}

27
src/client.xml
View File

@ -1,27 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<clients>
<client>
<id>0000</id>
<host>127.0.0.1</host>
<client_port>30000</client_port>
<memcached>127.0.0.1:20000</memcached>
</client>
<client>
<id>0001</id>
<host>127.0.0.1</host>
<client_port>30001</client_port>
<memcached>127.0.0.1:20001</memcached>
</client>
<client>
<id>0002</id>
<host>127.0.0.1</host>
<client_port>30002</client_port>
<memcached>127.0.0.1:20002</memcached>
</client>
<client>
<id>0003</id>
<host>127.0.0.1</host>
<client_port>30003</client_port>
<memcached>127.0.0.1:20003</memcached>
</client>
</clients>

39
src/client/Client.java
View File

@ -1,39 +0,0 @@
package client;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
public class Client
{
ClientBootstrap bootstrap;
ChannelFuture channelFuture;
String host;
int port;
int id;
public boolean init(String host, int port)
{
bootstrap = new ClientBootstrap(
new NioClientSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
// Set up the event pipeline factory.
bootstrap.setPipelineFactory(new MClientPipelineFactory());
// Start the connection attempt.
channelFuture = bootstrap.connect(new InetSocketAddress(host, port));
return channelFuture.isSuccess();
}
public void stop()
{
channelFuture.awaitUninterruptibly();
if (!channelFuture.isSuccess())
{
channelFuture.getCause().printStackTrace();
}
channelFuture.getChannel().getCloseFuture().awaitUninterruptibly();
bootstrap.releaseExternalResources();
}
}

120
src/client/ClientMgr.java
View File

@ -1,120 +0,0 @@
package client;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import server.ClientConfig;
public class ClientMgr
{
public HashMap<Integer, Client> m_mapLocalClients;
static ClientMgr clientMgr;
public int mClientNumber;
public static int nCopyNode;
public static int protocol;
public static final Integer twoPhaseCommit = 0;
public static final Integer paxos = (nCopyNode-1)-((nCopyNode-1)/2+1);
public static final Integer weak = nCopyNode-2;
public static ClientMgr getInstance()
{
if (clientMgr == null)
{
clientMgr = new ClientMgr();
}
return clientMgr;
}
public Integer getSize() {
return m_mapLocalClients.size()+1;
}
// num ΪhashµÚÒ»¸ö½Úµã
public boolean isCopyNode(Integer hash)
{
if ((mClientNumber-hash + m_mapLocalClients.size()+1)
%(m_mapLocalClients.size()+1)<nCopyNode)
{
return true;
}
return false;
}
@SuppressWarnings("rawtypes")
public void init(int num, HashMap<Integer, ClientConfig> hm)
{
m_mapLocalClients = new HashMap<Integer, Client>();
mClientNumber = num;
Iterator iter = hm.entrySet().iterator();
while (iter.hasNext())
{
Entry entry = (Entry) iter.next();
ClientConfig cc = (ClientConfig)entry.getValue();
if (cc.id != mClientNumber)
{
Client lc = new Client();
lc.host = cc.host;
lc.port = cc.client_port;
lc.id = cc.id;
m_mapLocalClients.put(lc.id, lc);
if(lc.init(lc.host, lc.port))
{
System.out.println("client connected successful");
}
}
}
}
public String getCopyHost(Integer number, Integer index)
{
Client client =m_mapLocalClients.get((number+index)%getSize());
if (client==null) {
return null;
}
return client.host+client.port;
}
public String getCurCoyhost(Integer number)
{
Client client =m_mapLocalClients.get((number+mClientNumber)%m_mapLocalClients.size());
if (client==null) {
return null;
}
return client.host+client.port;
}
@SuppressWarnings("rawtypes")
public Integer getClientNum(String host)
{
Iterator iter = m_mapLocalClients.entrySet().iterator();
while (iter.hasNext())
{
Entry entry = (Entry) iter.next();
Client client = (Client)entry.getValue();
if (host.compareTo("/"+client.host+":"+client.port) == 0)
{
return client.id;
}
}
return -1;
}
@SuppressWarnings("rawtypes")
public void connect(String host)
{
Iterator iter = m_mapLocalClients.entrySet().iterator();
while (iter.hasNext())
{
Entry entry = (Entry) iter.next();
Client client = (Client)entry.getValue();
if (client.host == host)
{
client.init(client.host, client.port);
break;
}
}
}
}

44
src/client/MClientHandler.java
View File

@ -1,44 +0,0 @@
package client;
import messageBody.memcachedmsg.nm_Connected;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import server.NetMsg;
import common.EMSGID;
public class MClientHandler extends SimpleChannelUpstreamHandler
{
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
{
nm_Connected.Builder builder = nm_Connected.newBuilder();
builder.setNum(ClientMgr.getInstance().mClientNumber);
NetMsg sendMsg = NetMsg.newMessage();
sendMsg.setMsgID(EMSGID.nm_connected);
sendMsg.setMessageLite(builder);
e.getChannel().write(sendMsg);
}
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
{
if (!(e.getMessage() instanceof NetMsg))
{
return;//(1)
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
{
if (e.getChannel().getLocalAddress() == null) {
return;
}
}
}

21
src/client/MClientPipelineFactory.java
View File

@ -1,21 +0,0 @@
package client;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import server.MDecoder;
import server.MEncoder;
public class MClientPipelineFactory implements ChannelPipelineFactory
{
public ChannelPipeline getPipeline() throws Exception
{
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", new MDecoder());
pipeline.addLast("encoder", new MEncoder());
return pipeline;
}
}

20
src/common/EMSGID.java
View File

@ -1,20 +0,0 @@
package common;
public enum EMSGID
{
nr_read, //WebApp send read request to R-Memcached
nr_read_res, //R-Memcached respond read request from WebServer
nr_write, //WebApp send write request to R-Memcached
nr_write_copy,
nr_write_res, //R-Memcached respond write request from WebServer
nr_connected_mem, //Init connection channel between WebApp and R-Memcached
nr_connected_mem_back,
nm_connected, //Init connection channel between R-Memcached and R-Memcached
nm_connected_mem_back,
nm_connected_web_back,
nm_read, //R-Memcached send read request to another R-Memcached
nm_read_recovery, //data recovery
nm_write_1, //write phase 1
nm_write_1_res,
nm_write_2 //write phase 2
}

41
src/common/MessageManager.java
View File

@ -1,41 +0,0 @@
package common;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import com.google.protobuf.GeneratedMessage;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.MessageLite;
/**
*
* @author Yanran Lu
*
*/
public class MessageManager { //create messageMap to manage all messages
private static Map<Integer, MessageLite> messageMap = new HashMap<Integer, MessageLite>();
public static void addMessageCla(int id, //add message to map
Class<? extends GeneratedMessage> msgCla)
throws NoSuchMethodException, SecurityException,
IllegalAccessException, IllegalArgumentException,
InvocationTargetException {
if (msgCla == null)
return;
Method method = msgCla.getMethod("getDefaultInstance");
MessageLite lite = (MessageLite) method.invoke(null, null);
messageMap.put(id, lite);
}
public static MessageLite getMessage(int id, byte[] body) //get message from map
throws InvalidProtocolBufferException {
MessageLite list = messageMap.get(id);
if (list == null) {
System.err.printf("msg %d no register", id);
return null;
}
return list.newBuilderForType().mergeFrom(body).build();
}
}

56
src/common/RegisterHandler.java
View File

@ -1,56 +0,0 @@
package common;
import java.lang.reflect.InvocationTargetException;
import messageBody.memcachedmsg.nm_Connected;
import messageBody.memcachedmsg.nm_Connected_mem_back;
import messageBody.memcachedmsg.nm_read;
import messageBody.memcachedmsg.nm_read_recovery;
import messageBody.memcachedmsg.nm_write_1;
import messageBody.memcachedmsg.nm_write_1_res;
import messageBody.memcachedmsg.nm_write_2;
import messageBody.requestMsg.nr_Connected_mem;
import messageBody.requestMsg.nr_Read;
import messageBody.requestMsg.nr_write;
import com.google.protobuf.GeneratedMessage;
public class RegisterHandler {
public static void initHandler() // register all messages created by
// protobuf and will be handled by R-Memcached
{
initHandler(EMSGID.nm_connected.ordinal(), nm_Connected.class);
initHandler(EMSGID.nm_connected_mem_back.ordinal(),
nm_Connected_mem_back.class);
initHandler(EMSGID.nr_connected_mem.ordinal(), nr_Connected_mem.class);
initHandler(EMSGID.nr_read.ordinal(), nr_Read.class);
initHandler(EMSGID.nm_read.ordinal(), nm_read.class);
initHandler(EMSGID.nm_read_recovery.ordinal(), nm_read_recovery.class);
initHandler(EMSGID.nr_write.ordinal(), nr_write.class);
initHandler(EMSGID.nm_write_1.ordinal(), nm_write_1.class);
initHandler(EMSGID.nm_write_1_res.ordinal(), nm_write_1_res.class);
initHandler(EMSGID.nm_write_2.ordinal(), nm_write_2.class);
}
private static void initHandler(int id,
Class<? extends GeneratedMessage> msgCla) {
try {
MessageManager.addMessageCla(id, msgCla);
} catch (NoSuchMethodException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (SecurityException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IllegalAccessException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IllegalArgumentException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InvocationTargetException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

3
src/config.properties
View File

@ -1,3 +0,0 @@
webServerHost = 192.168.3.222
replicasNum = 4
consistencyProtocol =twoPhaseCommit

4
src/log4j.properties
View File

@ -1,4 +0,0 @@
log4j.rootLogger=ERROR, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

117
src/memcached/ByteBufArrayInputStream.java
View File

@ -1,117 +0,0 @@
package cn.edu.buaa.act.memcachedClient;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.List;
public final class ByteBufArrayInputStream extends InputStream implements
LineInputStream {
private ByteBuffer[] bufs;
private int currentBuf = 0;
public ByteBufArrayInputStream(List<ByteBuffer> bufs) throws Exception {
this(bufs.toArray(new ByteBuffer[] {}));
}
public ByteBufArrayInputStream(ByteBuffer[] bufs) throws Exception {
if (bufs == null || bufs.length == 0)
throw new Exception("buffer is empty");
this.bufs = bufs;
for (ByteBuffer b : bufs)
b.flip();
}
public int read() {
do {
if (bufs[currentBuf].hasRemaining())
return bufs[currentBuf].get();
currentBuf++;
} while (currentBuf < bufs.length);
currentBuf--;
return -1;
}
public int read(byte[] buf) {
int len = buf.length;
int bufPos = 0;
do {
if (bufs[currentBuf].hasRemaining()) {
int n = Math.min(bufs[currentBuf].remaining(), len - bufPos);
bufs[currentBuf].get(buf, bufPos, n);
bufPos += n;
}
currentBuf++;
} while (currentBuf < bufs.length && bufPos < len);
currentBuf--;
if (bufPos > 0 || (bufPos == 0 && len == 0))
return bufPos;
else
return -1;
}
public String readLine() throws IOException {
byte[] b = new byte[1];
ByteArrayOutputStream bos = new ByteArrayOutputStream();
boolean eol = false;
while (read(b, 0, 1) != -1) {
if (b[0] == 13) {
eol = true;
} else {
if (eol) {
if (b[0] == 10)
break;
eol = false;
}
}
// cast byte into char array
bos.write(b, 0, 1);
}
if (bos == null || bos.size() <= 0) {
throw new IOException(
"++++ Stream appears to be dead, so closing it down");
}
// else return the string
return bos.toString().trim();
}
public void clearEOL() throws IOException {
byte[] b = new byte[1];
boolean eol = false;
while (read(b, 0, 1) != -1) {
// only stop when we see
// \r (13) followed by \n (10)
if (b[0] == 13) {
eol = true;
continue;
}
if (eol) {
if (b[0] == 10)
break;
eol = false;
}
}
}
public String toString() {
StringBuilder sb = new StringBuilder("ByteBufArrayIS: ");
sb.append(bufs.length).append(" bufs of sizes: \n");
for (int i = 0; i < bufs.length; i++) {
sb.append(" ").append(i)
.append(": ").append(bufs[i]).append("\n");
}
return sb.toString();
}
}

22
src/memcached/ContextObjectInputStream.java
View File

@ -1,22 +0,0 @@
package cn.edu.buaa.act.memcachedClient;
import java.io.*;
public class ContextObjectInputStream extends ObjectInputStream {
ClassLoader mLoader;
public ContextObjectInputStream(InputStream in, ClassLoader loader)
throws IOException, SecurityException {
super(in);
mLoader = loader;
}
protected Class resolveClass(ObjectStreamClass v) throws IOException,
ClassNotFoundException {
if (mLoader == null)
return super.resolveClass(v);
else
return Class.forName(v.getName(), true, mLoader);
}
}

51
src/memcached/ErrorHandler.java
View File

@ -1,51 +0,0 @@
package cn.edu.buaa.act.memcachedClient;
public interface ErrorHandler {
/**
* Called for errors thrown during initialization.
*/
public void handleErrorOnInit( final MemcachedClient client ,
final Throwable error );
/**
* Called for errors thrown during {@link MemcachedClient#get(String)} and related methods.
*/
public void handleErrorOnGet( final MemcachedClient client ,
final Throwable error ,
final String cacheKey );
/**
* Called for errors thrown during {@link MemcachedClient#getMulti(String)} and related methods.
*/
public void handleErrorOnGet( final MemcachedClient client ,
final Throwable error ,
final String[] cacheKeys );
/**
* Called for errors thrown during {@link MemcachedClient#set(String,Object)} and related methods.
*/
public void handleErrorOnSet( final MemcachedClient client ,
final Throwable error ,
final String cacheKey );
/**
* Called for errors thrown during {@link MemcachedClient#delete(String)} and related methods.
*/
public void handleErrorOnDelete( final MemcachedClient client ,
final Throwable error ,
final String cacheKey );
/**
* Called for errors thrown during {@link MemcachedClient#flushAll()} and related methods.
*/
public void handleErrorOnFlush( final MemcachedClient client ,
final Throwable error );
/**
* Called for errors thrown during {@link MemcachedClient#stats()} and related methods.
*/
public void handleErrorOnStats( final MemcachedClient client ,
final Throwable error);
} // interface

28
src/memcached/LineInputStream.java
View File

@ -1,28 +0,0 @@
package cn.edu.buaa.act.memcachedClient;
import java.io.IOException;
public interface LineInputStream {
/**
* Read everything up to the next end-of-line. Does not include the end of
* line, though it is consumed from the input.
*
* @return All next up to the next end of line.
*/
public String readLine() throws IOException;
/**
* Read everything up to and including the end of line.
*/
public void clearEOL() throws IOException;
/**
* Read some bytes.
*
* @param buf
* The buffer into which read.
* @return The number of bytes actually read, or -1 if none could be read.
*/
public int read(byte[] buf) throws IOException;
}

1781
src/memcached/MemcachedClient.java
View File

File diff suppressed because it is too large Load Diff

412
src/memcached/NativeHandler.java
View File

@ -1,412 +0,0 @@
package cn.edu.buaa.act.memcachedClient;
import java.util.Date;
import org.apache.log4j.Logger;
/**
* 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.
*
*/
public class NativeHandler {
// logger
private static Logger log = Logger.getLogger(NativeHandler.class.getName());
/**
* Detemine of object can be natively serialized by this class.
*
* @param value
* Object to test.
* @return true/false
*/
public static boolean isHandled(Object value) {
return (value instanceof Byte || value instanceof Boolean
|| value instanceof Integer || value instanceof Long
|| value instanceof Character || value instanceof String
|| value instanceof StringBuffer || value instanceof Float
|| value instanceof Short || value instanceof Double
|| value instanceof Date || value instanceof StringBuilder || value instanceof byte[]) ? true
: false;
}
/**
* Returns the flag for marking the type of the byte array.
*
* @param value
* Object we are storing.
* @return int marker
*/
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)
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
*
* @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)
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;
}
protected static byte[] encode(Byte value) {
byte[] b = new byte[1];
b[0] = value.byteValue();
return b;
}
protected static byte[] encode(Boolean value) {
byte[] b = new byte[1];
if (value.booleanValue())
b[0] = 1;
else
b[0] = 0;
return b;
}
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;
}
protected static byte[] getBytes(long value) {
byte[] b = new byte[8];
b[0] = (byte) ((value >> 56) & 0xFF);
b[1] = (byte) ((value >> 48) & 0xFF);
b[2] = (byte) ((value >> 40) & 0xFF);
b[3] = (byte) ((value >> 32) & 0xFF);
b[4] = (byte) ((value >> 24) & 0xFF);
b[5] = (byte) ((value >> 16) & 0xFF);
b[6] = (byte) ((value >> 8) & 0xFF);
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);
b[1] = (byte) ((value >> 16) & 0xFF);
b[2] = (byte) ((value >> 8) & 0xFF);
b[3] = (byte) ((value >> 0) & 0xFF);
return b;
}
/**
* Decodes byte array using memcache flag to determine type.
*
* @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
*/
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)
+ ((((long) b[5]) & 0xFF) << 16)
+ ((((long) b[4]) & 0xFF) << 24)
+ ((((long) b[3]) & 0xFF) << 32)
+ ((((long) b[2]) & 0xFF) << 40)
+ ((((long) b[1]) & 0xFF) << 48) + ((((long) b[0]) & 0xFF) << 56));
}
}

27
src/memcached/NestedIOException.java
View File

@ -1,27 +0,0 @@
package cn.edu.buaa.act.memcachedClient;
import java.io.*;
/**
* Bridge class to provide nested Exceptions with IOException which has
* constructors that don't take Throwables.
*
*/
public class NestedIOException extends IOException {
/**
* Create a new <code>NestedIOException</code> instance.
*
* @param cause
* object of type throwable
*/
public NestedIOException(Throwable cause) {
super(cause.getMessage());
super.initCause(cause);
}
public NestedIOException(String message, Throwable cause) {
super(message);
initCause(cause);
}
}

432
src/memcached/SockIO.java
View File

@ -1,432 +0,0 @@
package memcached;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.channels.SocketChannel;
import org.apache.log4j.Logger;
/**
* MemCached client for Java, utility class for Socket IO.
*
* This class is a wrapper around a Socket and its streams.
*
*/
public class SockIO implements LineInputStream {
// logger
private static Logger log = Logger.getLogger(SockIO.class.getName());
// data
private String host;
private Socket sock;
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
* @param timeout
* int ms to block on data for read
* @param connectTimeout
* timeout (in ms) for initial connection
* @param noDelay
* TCP NODELAY option?
* @throws IOException
* if an io error occurrs when creating socket
* @throws UnknownHostException
* if hostname is invalid
*/
public SockIO(String host, int port, int timeout, int connectTimeout,
boolean noDelay) throws IOException, UnknownHostException {
// get a socket channel
sock = getSocket(host, port, connectTimeout);
if (timeout >= 0)
sock.setSoTimeout(timeout);
// testing only
sock.setTcpNoDelay(noDelay);
// wrap streams
in = new DataInputStream(new BufferedInputStream(sock.getInputStream()));
out = new BufferedOutputStream(sock.getOutputStream());
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
* @param noDelay
* TCP NODELAY option?
* @throws IOException
* if an io error occurrs when creating socket
* @throws UnknownHostException
* if hostname is invalid
*/
public SockIO(String host, int timeout, int connectTimeout, boolean noDelay)
throws IOException, UnknownHostException {
String[] ip = host.split(":");
// get socket: default is to use non-blocking connect
sock = getSocket(ip[0], Integer.parseInt(ip[1]), connectTimeout);
if (timeout >= 0)
this.sock.setSoTimeout(timeout);
// testing only
sock.setTcpNoDelay(noDelay);
// wrap streams
in = new DataInputStream(new BufferedInputStream(sock.getInputStream()));
out = new BufferedOutputStream(sock.getOutputStream());
this.host = host;
}
/**
* Method which gets a connection from SocketChannel.
*
* @param host
* host to establish connection to
* @param port
* port on that host
* @param timeout
* connection timeout in ms
*
* @return connected socket
* @throws IOException
* if errors connecting or if connection times out
*/
protected static Socket getSocket(String host, int port, int timeout)
throws IOException {
SocketChannel sock = SocketChannel.open();
sock.socket().connect(new InetSocketAddress(host, port), timeout);
return sock.socket();
}
/**
* 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
*
* @return String representation of host (hostname:port)
*/
public String getHost() {
return this.host;
}
/**
* closes socket and all streams connected to it
*
* @throws IOException
* if fails to close streams or socket
*/
public void trueClose() throws IOException {
trueClose(true);
}
/**
* closes socket and all streams connected to it
*
* @throws IOException
* if fails to close streams or socket
*/
public void trueClose(boolean addToDeadPool) throws IOException {
if (log.isDebugEnabled())
log.debug("++++ Closing socket for real: " + toString());
boolean err = false;
StringBuilder errMsg = new StringBuilder();
if (in != null) {
try {
in.close();
} catch (IOException ioe) {
log.error("++++ error closing input stream for socket: "
+ toString() + " for host: " + getHost());
log.error(ioe.getMessage(), ioe);
errMsg.append("++++ error closing input stream for socket: "
+ toString() + " for host: " + getHost() + "\n");
errMsg.append(ioe.getMessage());
err = true;
}
}
if (out != null) {
try {
out.close();
} catch (IOException ioe) {
log.error("++++ error closing output stream for socket: "
+ toString() + " for host: " + getHost());
log.error(ioe.getMessage(), ioe);
errMsg.append("++++ error closing output stream for socket: "
+ toString() + " for host: " + getHost() + "\n");
errMsg.append(ioe.getMessage());
err = true;
}
}
if (sock != null) {
try {
sock.close();
} catch (IOException ioe) {
log.error("++++ error closing socket: " + toString()
+ " for host: " + getHost());
log.error(ioe.getMessage(), ioe);
errMsg.append("++++ error closing socket: " + toString()
+ " for host: " + getHost() + "\n");
errMsg.append(ioe.getMessage());
err = true;
}
}
// check in to pool
if (addToDeadPool && sock != null) {
// check in to pool pool.checkIn( this, false );
System.out.println("check in to pool ");
}
in = null;
out = null;
sock = null;
if (err)
throw new IOException(errMsg.toString());
}
/**
* sets closed flag and checks in to connection pool but does not close
* connections
*/
void close() {
// check in to pool
if (log.isDebugEnabled())
log.debug("++++ marking socket (" + this.toString()
+ ") as closed and available to return to avail pool");
// pool.checkIn( this );
}
/**
* checks if the connection is open
*
* @return true if connected
*/
boolean isConnected() {
return (sock != null && sock.isConnected());
}
/*
* checks to see that the connection is still working
*
* @return true if still alive
*/
boolean isAlive() {
if (!isConnected())
return false;
// try to talk to the server w/ a dumb query to ask its version
try {
this.write("version\r\n".getBytes());
this.flush();
this.readLine();
} catch (IOException ex) {
return false;
}
return true;
}
/**
* reads a line intentionally not using the deprecated readLine method from
* DataInputStream
*
* @return String that was read in
* @throws IOException
* if io problems during read
*/
public String readLine() throws IOException {
if (sock == null || !sock.isConnected()) {
log.error("++++ attempting to read from closed socket");
throw new IOException("++++ attempting to read from closed socket");
}
byte[] b = new byte[1];
ByteArrayOutputStream bos = new ByteArrayOutputStream();
boolean eol = false;
while (in.read(b, 0, 1) != -1) {
if (b[0] == 13) {
eol = true;
} else {
if (eol) {
if (b[0] == 10)
break;
eol = false;
}
}
// cast byte into char array
bos.write(b, 0, 1);
}
if (bos == null || bos.size() <= 0) {
throw new IOException(
"++++ Stream appears to be dead, so closing it down");
}
// else return the string
return bos.toString().trim();
}
/**
* reads up to end of line and returns nothing
*
* @throws IOException
* if io problems during read
*/
public void clearEOL() throws IOException {
if (sock == null || !sock.isConnected()) {
log.error("++++ attempting to read from closed socket");
throw new IOException("++++ attempting to read from closed socket");
}
byte[] b = new byte[1];
boolean eol = false;
while (in.read(b, 0, 1) != -1) {
// only stop when we see
// \r (13) followed by \n (10)
if (b[0] == 13) {
eol = true;
continue;
}
if (eol) {
if (b[0] == 10)
break;
eol = false;
}
}
}
/**
* reads length bytes into the passed in byte array from dtream
*
* @param b
* byte array
* @throws IOException
* if io problems during read
*/
public int read(byte[] b) throws IOException {
if (sock == null || !sock.isConnected()) {
log.error("++++ attempting to read from closed socket");
throw new IOException("++++ attempting to read from closed socket");
}
int count = 0;
while (count < b.length) {
int cnt = in.read(b, count, (b.length - count));
count += cnt;
}
return count;
}
/**
* flushes output stream
*
* @throws IOException
* if io problems during read
*/
void flush() throws IOException {
if (sock == null || !sock.isConnected()) {
log.error("++++ attempting to write to closed socket");
throw new IOException("++++ attempting to write to closed socket");
}
out.flush();
}
/**
* writes a byte array to the output stream
*
* @param b
* byte array to write
* @throws IOException
* if an io error happens
*/
void write(byte[] b) throws IOException {
if (sock == null || !sock.isConnected()) {
log.error("++++ attempting to write to closed socket");
throw new IOException("++++ attempting to write to closed socket");
}
out.write(b);
}
/**
* use the sockets hashcode for this object 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
*
* @return string
*/
public String toString() {
return (sock == null) ? "" : sock.toString();
}
/**
* Hack to reap any leaking children.
*/
protected void finalize() throws Throwable {
try {
if (sock != null) {
log.error("++++ closing potentially leaked socket in finalize");
sock.close();
sock = null;
}
} catch (Throwable t) {
log.error(t.getMessage(), t);
} finally {
super.finalize();
}
}
}

9
src/server/ClientConfig.java
View File

@ -1,9 +0,0 @@
package server;
public class ClientConfig
{
public int id;
public String host;
public int client_port;
public String memcached;
}

19
src/server/LockKey.java
View File

@ -1,19 +0,0 @@
package server;
public class LockKey {
public Integer memNumber = 0;
public Integer ncount = 0;
public Integer state = unLock;
public long time;
LockKey(Integer num, Integer count, long t, Integer s) {
memNumber = num;
ncount = count;
time = t;
state = s;
}
public final static Integer unLock = 0;
public final static Integer badLock = 1;
public final static Integer waitLock = 2;
}

33
src/server/MDecoder.java
View File

@ -1,33 +0,0 @@
package server;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import common.EMSGID;
public class MDecoder extends FrameDecoder
{
@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
if (buffer.readableBytes() < 4)
{
return null;//(1)
}
int dataLength = buffer.getInt(buffer.readerIndex());
if (buffer.readableBytes() < dataLength + 4)
{
return null;//(2)
}
buffer.skipBytes(4);//(3)
int id = buffer.readInt();
byte[] decoded = new byte[dataLength-4];
buffer.readBytes(decoded);
NetMsg msg = new NetMsg(decoded, id);//(4)
msg.setMsgID(EMSGID.values()[id]);
return msg;
}
}

27
src/server/MEncoder.java
View File

@ -1,27 +0,0 @@
package server;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
public class MEncoder extends OneToOneEncoder
{
@Override
protected Object encode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception
{
if (!(msg instanceof NetMsg))
{
return msg;//(1)
}
NetMsg res = (NetMsg)msg;
byte[] data = res.getBytes();
int dataLength = data.length+4;
ChannelBuffer buf = ChannelBuffers.dynamicBuffer();//(2)
buf.writeInt(dataLength);
buf.writeInt(res.msgID.ordinal());
buf.writeBytes(data);
return buf;//(3)
}
}

35
src/server/MServerHandler.java
View File

@ -1,35 +0,0 @@
package server;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
public class MServerHandler extends SimpleChannelUpstreamHandler
{
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
{
if (!(e.getMessage() instanceof NetMsg))
{
return;
}
memSession.getInstance().addSession(e);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
{
Channel channel = e.getChannel();
channel.close();
}
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)throws Exception
{
}
}

18
src/server/MServerPipelineFactory.java
View File

@ -1,18 +0,0 @@
package server;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
public class MServerPipelineFactory implements ChannelPipelineFactory
{
public ChannelPipeline getPipeline() throws Exception
{
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", new MDecoder());
pipeline.addLast("encoder", new MEncoder());
pipeline.addLast("handler", new MServerHandler());
return pipeline;
}
}

143
src/server/MemcachedMain.java
View File

@ -1,143 +0,0 @@
package server;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.HashMap;
import java.util.Properties;
import java.util.Scanner;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;
import client.Client;
import client.ClientMgr;
import common.RegisterHandler;
public class MemcachedMain {
HashMap<Integer, ClientConfig> m_mapMemcachedClient;
String webServerHost;
String protocolName;
public boolean initConfig() {
m_mapMemcachedClient = new HashMap<Integer, ClientConfig>();
File f = new File(System.getProperty("user.dir"));
String path = f.getPath() + File.separator + "bin" + File.separator;
readClientsXML(path + "client.xml");
try {
Properties properties = new Properties();
properties.load(new FileInputStream(path+"config.properties"));
webServerHost = properties.getProperty("webServerHost").toString();
ClientMgr.nCopyNode = Integer.parseInt(properties.getProperty("replicasNum"));
protocolName = properties.getProperty("consistencyProtocol").toString();
if(protocolName.equals("twoPhaseCommit")){
ClientMgr.protocol = ClientMgr.twoPhaseCommit;
}else if(protocolName.equals("paxos")){
ClientMgr.protocol = ClientMgr.paxos;
}else if(protocolName.equals("weak")){
ClientMgr.protocol = ClientMgr.weak;
}else{
System.err.print("consistency protocol input error");
return false;
}
} catch (FileNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return true;
}
public int getMemcachedNumber() {
System.out.print("Please in put R-Memcached ID:");
@SuppressWarnings("resource")
Scanner scanner = new Scanner(System.in);
return Integer.decode(scanner.next());
}
public void start() {
initConfig();
int num = getMemcachedNumber();
RegisterHandler.initHandler();
memSession.getInstance().start(m_mapMemcachedClient.get(num).memcached);
ClientMgr clientMgr = ClientMgr.getInstance();
Server server = Server.getInstance();
server.init(m_mapMemcachedClient.get(num).client_port);
clientMgr.init(num, m_mapMemcachedClient);
Client webClient = new Client();
webClient.init(webServerHost, 8888);
}
public static void main(String[] args) {
MemcachedMain entrance = new MemcachedMain();
entrance.start();
}
// ÁÈ¡memcached clientÅäÖÃ
public boolean readClientsXML(String str) {
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
try {
factory.setIgnoringElementContentWhitespace(true);
DocumentBuilder db = factory.newDocumentBuilder();
Document xmldoc = db.parse(new File(str));
Element elmtInfo = xmldoc.getDocumentElement();
NodeList nodes = elmtInfo.getChildNodes();
for (int i = 0; i < nodes.getLength(); i++) {
Node result = nodes.item(i);
if (result.getNodeType() == Node.ELEMENT_NODE
&& result.getNodeName().equals("client")) {
NodeList ns = result.getChildNodes();
ClientConfig localClient = new ClientConfig();
int m = 0;
for (int j = 0; j < ns.getLength(); j++) {
Node record = ns.item(j);
if (record.getNodeType() == Node.ELEMENT_NODE) {
if (record.getNodeName().equals("id")) {
m++;
localClient.id = Integer.decode(record
.getTextContent());
} else if (record.getNodeName().equals("host")) {
m++;
localClient.host = record.getTextContent();
} else if (record.getNodeName().equals(
"client_port")) {
m++;
localClient.client_port = Integer.decode(record
.getTextContent());
} else if (record.getNodeName().equals("memcached")) {
m++;
localClient.memcached = record.getTextContent();
}
}
}
if (m == 4) {
m_mapMemcachedClient.put(localClient.id, localClient);
}
}
}
} catch (ParserConfigurationException e) {
e.printStackTrace();
} catch (SAXException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
}

51
src/server/NetMsg.java
View File

@ -1,51 +0,0 @@
package server;
import com.google.protobuf.GeneratedMessage;
import com.google.protobuf.MessageLite;
import common.EMSGID;
import common.MessageManager;
public class NetMsg // package different messages
{
EMSGID msgID;
MessageLite messageLite;
private NetMsg(){};
public static NetMsg newMessage()
{
return new NetMsg();
}
NetMsg(byte[] decoded, int id) throws Exception
{
messageLite = MessageManager.getMessage(id, decoded);
}
public byte[] getBytes() //get data in messageLite
{
return messageLite.toByteArray();
}
public EMSGID getMsgID() //get message catagory
{
return msgID;
}
public void setMsgID(EMSGID id) { //according EMSGID.java set the message ID
this.msgID = id;
}
@SuppressWarnings("unchecked")
public <T extends MessageLite> T getMessageLite() //get messageLite
{
return (T)messageLite;
}
@SuppressWarnings("rawtypes")
public void setMessageLite(GeneratedMessage.Builder builder)
{
this.messageLite = builder.build();
}
}

45
src/server/Server.java
View File

@ -1,45 +0,0 @@
package server;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
public class Server
{
static Server server;
ServerBootstrap bootstrap;
public void init(int port)
{
bootstrap = new ServerBootstrap(
new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
// Set up the default event pipeline.
bootstrap.setPipelineFactory(new MServerPipelineFactory());
bootstrap.setOption("child.tcpNoDelay", true);
bootstrap.setOption("child.keepAlive", true);
bootstrap.setOption("reuseAddress", true);
// Bind and start to accept incoming connections.
bootstrap.bind(new InetSocketAddress(port));
System.out.println("server init complete");
}
public void stop()
{
bootstrap.releaseExternalResources();
}
public static Server getInstance()
{
if (server == null)
{
server = new Server();
}
return server;
}
}

507
src/server/memSession.java
View File

@ -1,507 +0,0 @@
package server;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentLinkedQueue;
import messageBody.memcachedmsg.nm_Connected;
import messageBody.memcachedmsg.nm_Connected_mem_back;
import messageBody.memcachedmsg.nm_read;
import messageBody.memcachedmsg.nm_read_recovery;
import messageBody.memcachedmsg.nm_write_1;
import messageBody.memcachedmsg.nm_write_1_res;
import messageBody.memcachedmsg.nm_write_2;
import messageBody.requestMsg.nr_Connected_mem_back;
import messageBody.requestMsg.nr_Read;
import messageBody.requestMsg.nr_Read_res;
import messageBody.requestMsg.nr_write;
import messageBody.requestMsg.nr_write_res;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.util.internal.ConcurrentHashMap;
import memcached.MemcachedClient;
import client.ClientMgr;
import common.EMSGID;
public class memSession implements Runnable {
ConcurrentLinkedQueue<MessageEvent> recvQueue = new ConcurrentLinkedQueue<MessageEvent>();
ConcurrentHashMap<Integer, Channel> ClientChannelMap = new ConcurrentHashMap<Integer, Channel>();
ConcurrentHashMap<String, LockKey> LockKeyMap = new ConcurrentHashMap<String, LockKey>();
public MemcachedClient client;
static memSession session = null;
Channel webServeChannel = null;
public static memSession getInstance() {
if (session == null) {
session = new memSession();
}
return session;
}
public void start(String host) {
client = new MemcachedClient(host);
new Thread(session).start();
System.out.println("session start");
}
// Ôö¼ÓclientÁ¬½Ó
public void addClientChannel(Integer num, Channel ch) {
ClientChannelMap.put(num, ch);
}
public Channel getClientChannel(Integer id) {
return ClientChannelMap.get(id);
}
// ɾµôclientÁ¬½Ó
@SuppressWarnings("rawtypes")
public void removeClientChannel(Channel ch) {
Iterator iter = ClientChannelMap.entrySet().iterator();
while (iter.hasNext()) {
Entry entry = (Entry) iter.next();
if ((Channel) entry.getValue() == ch) {
ClientChannelMap.remove((Integer) entry.getKey());
break;
}
}
}
// ////////////////////////////////////////////////////////
public void run() {
long curTime = System.currentTimeMillis();
while (true) {
MessageEvent event = recvQueue.poll();
while (event != null) {
handle(event);
event = recvQueue.poll();
if (System.currentTimeMillis() - curTime > 20000) {
curTime = System.currentTimeMillis();
HandleBadLock(curTime);
}
}
try {
Thread.sleep((long) 0.00001);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
@SuppressWarnings("rawtypes")
public void HandleBadLock(long curTime) {
Iterator it = LockKeyMap.keySet().iterator();
while(it.hasNext()){
String key = (String) it.next();
LockKey value = LockKeyMap.get(key);
if (value.state == LockKey.waitLock) {
if (curTime - value.time > 20000) {
value.state = LockKey.badLock;
LockKeyMap.remove(key);
LockKeyMap.put(key, value);
}
}
}
}
public int getLockState(String key) {
LockKey lock = LockKeyMap.get(key);
if (lock == null) {
return LockKey.unLock;
}
return lock.state;
}
public void setLockState(String key, Integer state) {
LockKey lock = LockKeyMap.get(key);
if (lock != null) {
lock.state = state;
LockKeyMap.put(key, lock);
} else {
System.out.println("set Lock state error");
return;
}
}
public boolean lockKey(String key, LockKey lock) {
LockKey lockKey = LockKeyMap.put(key, lock);
if (lockKey != null && (lockKey.state == 0 || lockKey.state == 1)) {
return true;
}
return lockKey == null;
}
public int desLockKeyCount(String key) {
LockKey lock = LockKeyMap.get(key);
if (lock != null) {
lock.ncount--;
LockKeyMap.put(key, lock);
return lock.ncount;
}
return 0;
}
public boolean removeLock(String key) {
return LockKeyMap.remove(key) != null;
}
public void handle(MessageEvent e) {
NetMsg msg = (NetMsg) e.getMessage();
switch (msg.getMsgID()) {
case nm_connected: {
nm_Connected msgLite = msg.getMessageLite();
addClientChannel(msgLite.getNum(), e.getChannel());
nm_Connected_mem_back.Builder builder = nm_Connected_mem_back
.newBuilder();
builder.setNum(ClientMgr.getInstance().mClientNumber);
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nm_connected_mem_back);
e.getChannel().write(send);
}
break;
case nm_connected_mem_back: {
nm_Connected_mem_back msgLite = msg.getMessageLite();
addClientChannel(msgLite.getNum(), e.getChannel());
}
break;
case nr_connected_mem: {
webServeChannel = e.getChannel();
nr_Connected_mem_back.Builder builder = nr_Connected_mem_back
.newBuilder();
builder.setMemID(ClientMgr.getInstance().mClientNumber);
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nr_connected_mem_back);
e.getChannel().write(send);
}
break;
case nm_connected_web_back: {
webServeChannel = e.getChannel();
}
break;
case nr_read: {
nr_Read msgLite = msg.getMessageLite();
Integer state = getLockState(msgLite.getKey());
if (state == LockKey.waitLock) {
nr_Read_res.Builder builder = nr_Read_res.newBuilder();
builder.setKey(msgLite.getKey());
builder.setValue("");
builder.setTime(msgLite.getTime());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nr_read_res);
webServeChannel.write(send);
return;
} else if (state == LockKey.unLock) {
String value = (String) client.get(msgLite.getKey());
if (value != null) {
nr_Read_res.Builder builder = nr_Read_res.newBuilder();
builder.setKey(msgLite.getKey());
builder.setValue(value);
builder.setTime(msgLite.getTime());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nr_read_res);
webServeChannel.write(send);
return;
}
}
// Can't get data in local Memcached server, try to ask for data from another cache node
nm_read.Builder builder = nm_read.newBuilder();
builder.setKey(msgLite.getKey());
builder.setTime(msgLite.getTime());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nm_read);
if (sendOtherCopyMsg(gethashMem(msgLite.getKey()), send) == false) {
nr_Read_res.Builder builder1 = nr_Read_res.newBuilder();
builder1.setKey(msgLite.getKey());
builder.setValue("");
builder1.setTime(msgLite.getTime());
NetMsg send1 = NetMsg.newMessage();
send1.setMessageLite(builder1);
send1.setMsgID(EMSGID.nr_read_res);
webServeChannel.write(send);
}
}
break;
case nm_read: {
nm_read msgLite = msg.getMessageLite();
Integer state = getLockState(msgLite.getKey());
if (state == LockKey.unLock) {
String value = (String) client.get(msgLite.getKey());
if (value != null) {
nr_Read_res.Builder builder = nr_Read_res.newBuilder();
builder.setKey(msgLite.getKey());
builder.setTime(msgLite.getTime());
builder.setValue(value);
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nr_read_res);
webServeChannel.write(send);
nm_read_recovery.Builder builder1 = nm_read_recovery.newBuilder();
builder1.setKey(msgLite.getKey());
builder1.setTime(msgLite.getTime());
builder1.setValue(value);
NetMsg send1 = NetMsg.newMessage();
send.setMessageLite(builder1);
send.setMsgID(EMSGID.nm_read_recovery);
e.getChannel().write(send1);
return;
}
}
nr_Read_res.Builder builder = nr_Read_res.newBuilder();
builder.setKey(msgLite.getKey());
builder.setValue("");
builder.setTime(msgLite.getTime());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nr_read_res);
webServeChannel.write(send);
}
break;
case nm_read_recovery: {
nm_read_recovery msgLite = msg.getMessageLite();
Integer state = getLockState(msgLite.getKey());
if(state == LockKey.waitLock) {
System.out.println("recovery fail because of waitlock.");
return;
}else if (state == LockKey.badLock) {
removeLock(msgLite.getKey());
}
boolean res = client.set(msgLite.getKey(), msgLite.getValue());
if (!res) {
setLockState(msgLite.getKey(), LockKey.badLock);
System.err.println("read recovery fail");
}
}
break;
case nr_write: {
nr_write msgLite = msg.getMessageLite();
Integer state = getLockState(msgLite.getKey());
if (state == LockKey.waitLock) {
System.out.println("write conflict, please request again.");
nr_write_res.Builder builder2 = nr_write_res.newBuilder();
builder2.setKey(msgLite.getKey());
builder2.setValue("");
builder2.setTime(msgLite.getTime());
NetMsg send2 = NetMsg.newMessage();
send2.setMessageLite(builder2);
send2.setMsgID(EMSGID.nr_write_res);
webServeChannel.write(send2);
return;
}
// else if (state == LockKey.badLock) {
// removeLock(msgLite.getKey());
// }
LockKey lockKey = new LockKey(
ClientMgr.getInstance().mClientNumber, ClientMgr.nCopyNode-1,
System.currentTimeMillis(), LockKey.waitLock);
if (lockKey(msgLite.getKey(), lockKey) == false) {
System.out.println("nr_write lock fail");
nr_write_res.Builder builder2 = nr_write_res.newBuilder();
builder2.setKey(msgLite.getKey());
builder2.setValue("");
builder2.setTime(msgLite.getTime());
NetMsg send2 = NetMsg.newMessage();
send2.setMessageLite(builder2);
send2.setMsgID(EMSGID.nr_write_res);
webServeChannel.write(send2);
System.out.println("write lock conflict, please request again.");
return;
}
nm_write_1.Builder builder = nm_write_1.newBuilder();
builder.setKey(msgLite.getKey());
builder.setValue(msgLite.getValue());
builder.setMemID(ClientMgr.getInstance().mClientNumber);
builder.setTime(msgLite.getTime());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nm_write_1);
int count = sendOtherAllCopyMsg(gethashMem(msgLite.getKey()), send);
if (count != ClientMgr.nCopyNode-1) {
System.out.println("write failed in send invalidation msg to remote node.");
nr_write_res.Builder builder2 = nr_write_res.newBuilder();
builder2.setKey(msgLite.getKey());
builder2.setValue("");
builder2.setTime(msgLite.getTime());
NetMsg send2 = NetMsg.newMessage();
send2.setMessageLite(builder2);
send2.setMsgID(EMSGID.nr_write_res);
webServeChannel.write(send2);
}
}
break;
case nm_write_1: {
nm_write_1 msgLite = msg.getMessageLite();
Integer state = getLockState(msgLite.getKey());
if (state == LockKey.waitLock) {
removeLock(msgLite.getKey());
}
else if (state == LockKey.badLock) {
removeLock(msgLite.getKey());
}
// System.out.println("ready to write key: "+msgLite.getKey());
LockKey lockKey = new LockKey(
ClientMgr.getInstance().mClientNumber, 0,
System.currentTimeMillis(), LockKey.waitLock);
if (lockKey(msgLite.getKey(), lockKey) == false) {
System.out.println("nm_write_1 Lock fail");
}
nm_write_1_res.Builder builder = nm_write_1_res.newBuilder();
builder.setKey(msgLite.getKey());
builder.setValue(msgLite.getValue());
builder.setTime(msgLite.getTime());
builder.setMemID(msgLite.getMemID());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nm_write_1_res);
getClientChannel(msgLite.getMemID()).write(send);
}
break;
case nm_write_1_res: {
nm_write_1_res msgLite = msg.getMessageLite();
if (desLockKeyCount(msgLite.getKey()) == ClientMgr.protocol) {
boolean res = client.set(msgLite.getKey(),
msgLite.getValue());
if (res) {
removeLock(msgLite.getKey());
// System.out.println("key:"+msgLite.getKey()+", value:"+msgLite.getValue());
nr_write_res.Builder builder2 = nr_write_res.newBuilder();
builder2.setKey(msgLite.getKey());
builder2.setValue(msgLite.getValue());
builder2.setTime(msgLite.getTime());
NetMsg send2 = NetMsg.newMessage();
send2.setMessageLite(builder2);
send2.setMsgID(EMSGID.nr_write_res);
webServeChannel.write(send2);
nm_write_2.Builder builder = nm_write_2.newBuilder();
builder.setKey(msgLite.getKey());
builder.setValue(msgLite.getValue());
builder.setMemID(msgLite.getMemID());
builder.setTime(msgLite.getTime());
NetMsg send = NetMsg.newMessage();
send.setMessageLite(builder);
send.setMsgID(EMSGID.nm_write_2);
sendOtherAllCopyMsg(gethashMem(msgLite.getKey()),send);
} else {
setLockState(msgLite.getKey(), LockKey.badLock);
System.err.println("write to memcached server error");
nr_write_res.Builder builder2 = nr_write_res.newBuilder();
builder2.setKey(msgLite.getKey());
builder2.setValue("");
builder2.setTime(msgLite.getTime());
NetMsg send2 = NetMsg.newMessage();
send2.setMessageLite(builder2);
send2.setMsgID(EMSGID.nr_write_res);
webServeChannel.write(send2);
}
}
}
break;
case nm_write_2: {
nm_write_2 msgLite = msg.getMessageLite();
boolean res = client.set(msgLite.getKey(), msgLite.getValue());
if (res) {
removeLock(msgLite.getKey());
// System.out.println("key:"+msgLite.getKey()+", value:"+msgLite.getValue());
} else {
setLockState(msgLite.getKey(), LockKey.badLock);
System.err.println("write in write_2 fail");
}
}
break;
default:
System.err.println(msg.getMsgID().toString());
break;
}
}
public void addSession(MessageEvent e) {
recvQueue.offer(e);
}
public int gethashMem(String key) {
return Math.abs(key.hashCode() % ClientMgr.getInstance().getSize());
}
public boolean sendOtherCopyMsg(Integer hash, NetMsg msg) {
for (int i = 0; i < ClientMgr.nCopyNode; i++) {
Integer index = (hash + i + ClientMgr.getInstance().getSize())
% ClientMgr.getInstance().getSize();
if (index == ClientMgr.getInstance().mClientNumber)
continue;
Channel eChannel = getClientChannel(index);
if (eChannel != null) {
eChannel.write(msg);
return true;
}
}
return false;
}
public int sendOtherAllCopyMsg(int hash, NetMsg msg) {
int count = 0;
for (int i = 0; i < ClientMgr.nCopyNode; i++) {
int index = (hash + i + ClientMgr.getInstance().getSize())
% ClientMgr.getInstance().getSize();
if (index == ClientMgr.getInstance().mClientNumber)
continue;
Channel eChannel = getClientChannel(index);
if (eChannel != null) {
eChannel.write(msg);
count++;
}
}
return count;
}
}

120
src/test/MainTest.java
View File

@ -1,120 +0,0 @@
package test;
import java.util.Hashtable;
import memcached.MemcachedClient;
import org.apache.log4j.PropertyConfigurator;
/**
* Test the communication between R-Memcached and Memcached server in one cache node.
* @author Yanran Lu
*/
public class MainTest {
// store results from threads
private static Hashtable<Integer, StringBuilder> threadInfo = new Hashtable<Integer, StringBuilder>();
public static void main(String[] args) {
PropertyConfigurator.configure(MainTest.class.getClass()
.getResource("/").getPath()
+ "log4j.properties");
// String[] serverlist = { "127.0.0.1:20000", "127.0.0.1:20001" ,
// "127.0.0.1:20002" };
int threads = Integer.parseInt(args[0]); // the number of threads
int runs = Integer.parseInt(args[1]); // how many request send out by a thread
int Nums = Integer.parseInt(args[2]); // the total size of data
int size = Integer.parseInt(args[3]); // the size of a data
// get object to store
byte[] obj = new byte[size];
for (int i = 0; i < size; i++) {
obj[i] = '1';
}
String value = new String(obj);
String[] keys = new String[size];
for (int i = 0; i < Nums; i++) {
keys[i] = "key" + i;
}
for (int i = 0; i < threads; i++) {
bench b = new bench(runs, Nums, i, value, keys);
b.start();
}
int i = 0;
while (i < threads) {
if (threadInfo.containsKey(new Integer(i))) {
System.out.println(threadInfo.get(new Integer(i)));
i++;
} else {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
System.exit(1);
}
/**
* Test code per thread.
*/
private static class bench extends Thread {
private int runs;
private int threadNum;
private String object;
private String[] keys;
private int size;
private int nums;
public bench(int runs, int nums, int threadNum, String object,
String[] keys) {
this.runs = runs;
this.threadNum = threadNum;
this.object = object;
this.keys = keys;
this.size = object.length();
this.nums = nums;
}
public void run() {
StringBuilder result = new StringBuilder();
// get client instance
MemcachedClient mc = new MemcachedClient("127.0.0.1:20000");
mc.setCompressEnable(false);
mc.setCompressThreshold(0);
try {
Thread.sleep(0);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
// time deletes
long start = System.nanoTime();
randReadWrite(mc, 0.8);
long elapse = (System.nanoTime() - start) / 1000000;
float avg = (float) elapse / runs;
result.append("\nthread " + threadNum + ": runs: " + runs
+ " read or write of obj " + (size / 1024)
+ "KB -- avg time per req " + avg + " ms (total: " + elapse
+ " ms)");
threadInfo.put(new Integer(threadNum), result);
}
public void randReadWrite(MemcachedClient mc, double scale) {
for (int i = 0; i < runs; i++) {
if (Math.random() < scale) {
mc.get(keys[i % nums]);
} else {
mc.set(keys[i % nums], object);
}
}
}
}
}