Mercurial > hg > Database > Alice
view src/alice/datasegment/LocalDataSegmentManager.java @ 63:498d1d2524d3
change getDataSegmentKey
author | kazz <kazz@cr.ie.u-ryukyu.ac.jp> |
---|---|
date | Thu, 09 Feb 2012 18:53:11 +0900 |
parents | ebdcab7b9b04 |
children | 7aaadd08288c |
line wrap: on
line source
package alice.datasegment; import java.util.concurrent.ConcurrentHashMap; import org.apache.log4j.Logger; import org.msgpack.type.Value; import alice.codesegment.CodeSegment; public class LocalDataSegmentManager extends DataSegmentManager { private String reverseKey = "local"; private ConcurrentHashMap<String, DataSegmentKey> dataSegments = new ConcurrentHashMap<String, DataSegmentKey>(); private Logger logger = Logger.getLogger("local"); public LocalDataSegmentManager() { new Thread(replyThread, "LocalDataSegmentManager").start(); } public DataSegmentKey getDataSegmentKey(String key) { DataSegmentKey dsKey= dataSegments.get(key); if (dsKey != null) return dsKey; if (key == null) return null; DataSegmentKey newDataSegmentKey = new DataSegmentKey(key); DataSegmentKey dataSegmentKey = dataSegments.putIfAbsent(key, newDataSegmentKey); if (dataSegmentKey == null) { newDataSegmentKey.runKeyThread(); dataSegmentKey = newDataSegmentKey; } return dataSegmentKey; } @Override public void put(String key, Value val, CodeSegment cs) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.PUT, null, key, val, 0, 0, replyQueue, cs, reverseKey); dataSegmentKey.addCommand(cmd); logger.debug(cmd.getCommandString()); } /** * Enqueue update command to the queue of each DataSegment key */ @Override public void update(String key, Value val, CodeSegment cs) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.UPDATE, null, key, val, 0, 0, replyQueue, cs, reverseKey); dataSegmentKey.addCommand(cmd); logger.debug(cmd.getCommandString()); } @Override public void take(Receiver receiver, String key, int index, CodeSegment cs) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); int seq = this.seq.getAndIncrement(); Command cmd = new Command(CommandType.TAKE, receiver, key, null, index, seq, replyQueue, cs, null); seqHash.put(seq, cmd); dataSegmentKey.addCommand(cmd); logger.debug(cmd.getCommandString()); } @Override public void peek(Receiver receiver, String key, int index, CodeSegment cs) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); int seq = this.seq.getAndIncrement(); Command cmd = new Command(CommandType.PEEK, receiver, key, null, index, seq, replyQueue, cs, null); seqHash.put(seq, cmd); // waiting for PUT or UPDATE at unique sequence number dataSegmentKey.addCommand(cmd); logger.debug(cmd.getCommandString()); } @Override public void remove(String key) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.REMOVE, null, key, null, 0, 0, replyQueue, null, null); dataSegmentKey.addCommand(cmd); logger.debug(cmd.getCommandString()); } @Override public void finish() { System.exit(0); } @Override public void close() { } }