Mercurial > hg > Database > Alice
view src/main/java/alice/datasegment/DataSegmentKey.java @ 530:4aeebea0c9b5 dispose
can't unzip
author | Nozomi Teruya <e125769@ie.u-ryukyu.ac.jp> |
---|---|
date | Sun, 03 May 2015 10:04:28 +0900 |
parents | bfec2c3ff1b8 |
children | b6049fb123d8 |
line wrap: on
line source
package alice.datasegment; import java.util.ArrayList; import java.util.Iterator; import alice.datasegment.Command; /** * ここがコマンドの中身部分 * * Synchronized DataSegment for each DataSegment key * @author kazz * */ public class DataSegmentKey { private ArrayList<DataSegmentValue> dataList = new ArrayList<DataSegmentValue>(); private ArrayList<Command> waitList = new ArrayList<Command>(); private int tailIndex = 1; public synchronized void runCommand(Command cmd) { switch (cmd.type) { case UPDATE: if (dataList.size() != 0) { dataList.remove(0); } case PUT: int index = tailIndex; tailIndex++; DataSegmentValue dsv = new DataSegmentValue(index, cmd.rData, cmd.reverseKey); dataList.add(dsv); // Process waiting peek and take commands for (Iterator<Command> iter = waitList.iterator(); iter.hasNext(); ) { System.out.println("in DSkey PUTfor"); Command waitCmd = iter.next(); if (waitCmd.index < index) { System.out.println("DSKey cmdFlag:" + cmd.getCompressFlag()); replyValue(waitCmd, dsv, cmd.getCompressFlag()); iter.remove(); if (waitCmd.type == CommandType.TAKE) { // someone is waiting for this put or update command dataList.remove(dsv); break; } } } break; case PEEK: if (cmd.index >= tailIndex) { waitList.add(cmd); break; } boolean waitFlag2 = true; for (DataSegmentValue data : dataList) { if (data.index > cmd.index) { replyValue(cmd, data, cmd.getCompressFlag()); //replyValue(cmd, data); waitFlag2 = false; break; } } if (waitFlag2) waitList.add(cmd); break; case TAKE: if (cmd.index >= tailIndex) { waitList.add(cmd); break; } boolean waitFlag = true; for (Iterator<DataSegmentValue> iter = dataList.iterator(); iter.hasNext(); ) { DataSegmentValue data = iter.next(); if (data.index > cmd.index) { System.out.println("DSKey2 cmdFlag:" + cmd.getCompressFlag()); replyValue(cmd, data, cmd.getCompressFlag()); //replyValue(cmd, data); iter.remove(); waitFlag = false; break; } } if (waitFlag) waitList.add(cmd); break; case REMOVE: // TODO: implements later break; default: } } public void replyValue(Command cmd, DataSegmentValue data){ Command rCmd = new Command(CommandType.REPLY, null, null, data.rData, data.index, cmd.seq, null, null, data.from); if (cmd.cs!=null){ // if cmd has cs-instance, it means Command from local. cmd.cs.ids.reply(cmd.receiver, rCmd); } else { try { if (!cmd.getQuickFlag()) { cmd.connection.sendQueue.put(rCmd); } else { cmd.connection.write(rCmd); } } catch (InterruptedException e) { e.printStackTrace(); } } } public void replyValue(Command cmd, DataSegmentValue data, boolean cFlag){ Command rCmd = new Command(CommandType.REPLY, null, null, data.rData, data.index, cmd.seq, null, null, data.from); rCmd.setCompressFlag(cFlag); if (cmd.cs!=null){ // if cmd has cs-instance, it means Command from local. cmd.cs.ids.reply(cmd.receiver, rCmd); } else { try { if (!cmd.getQuickFlag()) { cmd.connection.sendQueue.put(rCmd); } else { cmd.connection.write(rCmd); } } catch (InterruptedException e) { e.printStackTrace(); } } } }