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 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();
    }

    public void setReverseKey(String s){
        reverseKey = s;
    }

    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;
    }

    public void removeDataSegmentKey(String key) {
        if (key!=null)
            dataSegments.remove(key);
    }

    @Override
    public void put(String key, ReceiveData rData, boolean quickFlag) {
        Command cmd = new Command(CommandType.PUT, null, key, rData, 0, 0, null, null, reverseKey);
        put1(key, cmd);
    }

    public void put1(String key, Command cmd) {
        DataSegmentKey dataSegmentKey = getDataSegmentKey(key);
        dataSegmentKey.runCommand(cmd);
        if (logger.isDebugEnabled())
            logger.debug(cmd.getCommandString());
    }

    /**
     * Enqueue update command to the queue of each DataSegment key
     */

    @Override
    public void update(String key, ReceiveData rData, boolean quickFlag) {
        Command cmd = new Command(CommandType.UPDATE, null, key, rData, 0, 0, null, null, reverseKey);
        put1(key, cmd);
    }

    @Override
    public void take(Receiver receiver, CodeSegment cs, boolean quickFlag) {

        Command cmd = new Command(CommandType.TAKE, receiver, receiver.key, null, receiver.index, 0, replyQueue, cs, null);
        take1(receiver, cmd);
    }

    public void take1(Receiver receiver, Command cmd) {
        int seq = this.seq.getAndIncrement();
        cmd.setSeq(seq);
        //seqHash.put(seq, cmd);
        DataSegmentKey dataSegmentKey = getDataSegmentKey(receiver.key);
        dataSegmentKey.runCommand(cmd);
        if (logger.isDebugEnabled())
            logger.debug(cmd.getCommandString());
    }

    @Override
    public void peek(Receiver receiver, CodeSegment cs, boolean quickFlag) {
        Command cmd = new Command(CommandType.PEEK, receiver, receiver.key, null, receiver.index, 0, replyQueue, cs, null);
        take1(receiver, cmd);
    }

    @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.runCommand(cmd);
        if (logger.isDebugEnabled())
            logger.debug(cmd.getCommandString());
    }

    @Override public void finish() {
        System.exit(0);
    }

    @Override
    public void close() {

    }

    public void recommand(Receiver receiver, CodeSegment cs) {
        DataSegmentKey dataSegmentKey = getDataSegmentKey(receiver.key);
        int seq = this.seq.getAndIncrement();
        Command cmd = new Command(receiver.type, receiver, receiver.key, null, receiver.index, seq, replyQueue, cs, null);
        dataSegmentKey.runCommand(cmd);
        if (logger.isDebugEnabled())
            logger.debug(cmd.getCommandString());
    }

    @Override
    public void ping(String returnKey) {

    }

    @Override
    public void response(String returnKey) {

    }

    @Override
    public void shutdown() {

    }

    @Override
    public void setSendError(boolean b) {

    }
}