Mercurial > hg > Database > Alice
view src/alice/datasegment/LocalDataSegmentManager.java @ 225:bc061ee5f31f
bitonic sort work but data is not sorted
author | sugi |
---|---|
date | Thu, 28 Mar 2013 18:10:24 +0900 |
parents | 409d7679cf7b |
children | ca1c9c477f54 |
line wrap: on
line source
package alice.datasegment; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; 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"); private ThreadPoolExecutor dataSegmentExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), // initial number of threads Runtime.getRuntime().availableProcessors(), Integer.MAX_VALUE, // keepAliveTime TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()); public LocalDataSegmentManager() { new Thread(replyThread, "LocalDataSegmentManager-replyCommand").start(); } private class RunCommand implements Runnable { DataSegmentKey key; Command cmd; public RunCommand(DataSegmentKey key, Command cmd) { this.key = key; this.cmd = cmd; } @Override public void run() { key.runCommand(cmd); } } public void submitCommand(DataSegmentKey key, Command cmd) { dataSegmentExecutor.execute(new RunCommand(key, cmd)); } public DataSegmentKey getDataSegmentKey(String key) { DataSegmentKey dsKey = dataSegments.get(key); if (dsKey != null) return dsKey; if (key == null) return null; DataSegmentKey newDataSegmentKey = new DataSegmentKey(); DataSegmentKey dataSegmentKey = dataSegments.putIfAbsent(key, newDataSegmentKey); if (dataSegmentKey == null) { dataSegmentKey = newDataSegmentKey; } return dataSegmentKey; } @Override public void put(String key, Value val) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.PUT, null, key, val, 0, 0, null, null, reverseKey); submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) logger.debug(cmd.getCommandString()); } public void putObject(String key, Object obj) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.PUT, null, key, obj, 0, 0, null, null, reverseKey); submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) logger.debug(cmd.getCommandString()); } /** * Enqueue update command to the queue of each DataSegment key */ @Override public void update(String key, Value val) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.UPDATE, null, key, val, 0, 0, null, null, reverseKey); submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) logger.debug(cmd.getCommandString()); } public void updateObject(String key, Object val) { DataSegmentKey dataSegmentKey = getDataSegmentKey(key); Command cmd = new Command(CommandType.UPDATE, null, key, val, 0, 0, null, null, reverseKey); submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) 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); submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) 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 submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) 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); submitCommand(dataSegmentKey, cmd); if (logger.isDebugEnabled()) logger.debug(cmd.getCommandString()); } @Override public void finish() { System.exit(0); } @Override public void close() { } }