package voldemort.store.stats;

import com.google.common.collect.ImmutableList;
import voldemort.utils.Time;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

public class StreamStats {

    private final ConcurrentMap<Long, Handle> handles;
    private final AtomicLong handleIdGenerator;
    private final ConcurrentMap<Operation, RequestCounter> networkCounter;
    private final ConcurrentMap<Operation, RequestCounter> diskCounter;

    public StreamStats () {
        this.handles = new ConcurrentHashMap<Long, Handle>();
        this.handleIdGenerator = new AtomicLong(0L);
        this.networkCounter = new ConcurrentHashMap<Operation, RequestCounter>();
        this.diskCounter = new ConcurrentHashMap<Operation, RequestCounter>();

        for(Operation operation: Operation.values()) {
            networkCounter.put(operation, new RequestCounter(300000));
            diskCounter.put(operation, new RequestCounter(30000));
        }
    }

    public Handle makeHandle(Operation operation, List<Integer> partitionIds) {
        Handle handle = new Handle(handleIdGenerator.getAndIncrement(),
                                   operation,
                                   System.currentTimeMillis(),
                                   partitionIds);
        handles.put(handle.getId(), handle);
        return handle;
    }

    public void closeHandle(Handle handle) {
        // Empty for now
    }

    public void clearFinished() {
        for(long handleId: getHandleIds()) {
            if(getHandle(handleId).isFinished())
                handles.remove(handleId);
        }
    }

    protected Handle getHandle(long handleId) {
        if(!handles.containsKey(handleId))
            throw new IllegalArgumentException("No handle with id " + handleId);

        return handles.get(handleId);
    }

    public Collection<Long> getHandleIds() {
        return ImmutableList.copyOf(handles.keySet());
    }

    public Collection<Handle> getHandles() {
        return ImmutableList.copyOf(handles.values());
    }

    public void recordNetworkTime(Handle handle, long timeNs) {
        networkCounter.get(handle.getOperation()).addRequest(timeNs);
    }

    public void recordDiskTime(Handle handle, long timeNs) {
        diskCounter.get(handle.getOperation()).addRequest(timeNs);
    }

    public RequestCounter getNetworkCounter(Operation operation) {
        return networkCounter.get(operation);
    }

    public RequestCounter getDiskCounter(Operation operation) {
        return diskCounter.get(operation);
    }

    public enum Operation {
        FETCH_KEYS,
        FETCH_ENTRIES,
        FETCH_FILE,
        UPDATE,
        SLOP,
        DELETE,
    }

    public static class Handle {

        private final long id;
        private final Operation operation;
        private final long startedMs;
        private final List<Integer> partitionIds;
        private final AtomicLong entriesScanned;
        private final AtomicLong timeNetworkNs;
        private final AtomicLong timeDiskNs;
        private volatile boolean finished;

        private Handle(long id, Operation operation, long startedMs, List<Integer> partitionIds) {
            this.id = id;
            this.operation = operation;
            this.startedMs = startedMs;
            this.partitionIds = partitionIds;
            this.entriesScanned = new AtomicLong(0L);
            this.timeNetworkNs = new AtomicLong(0L);
            this.timeDiskNs = new AtomicLong(0L);
            this.finished = false;
        }

        public long getId() {
            return id;
        }

        public long getStartedMs() {
            return startedMs;
        }

        public Operation getOperation() {
            return operation;
        }

        public long getEntriesScanned() {
            return entriesScanned.get();
        }

        public long incrementEntriesScanned() {
            return entriesScanned.incrementAndGet();
        }

        public void setEntriesScanned(long newVal) {
            entriesScanned.set(newVal);
        }

        public long getEntriesPerSecond() {
            long elapsedSecs = System.currentTimeMillis() - startedMs;
            if(elapsedSecs == 0L)
                return 0L;
            return getEntriesScanned() / elapsedSecs;
        }

        public boolean isFinished() {
            return finished;
        }

        public void setFinished(boolean finished) {
            this.finished = finished;
        }

        public List<Integer> getPartitionIds() {
            return partitionIds;
        }

        public void recordTimeNetwork(long deltaNs) {
            timeNetworkNs.addAndGet(deltaNs);
        }

        public long getTimeNetworkNs() {
            return timeNetworkNs.get();
        }

        public void recordTimeDisk(long deltaNs) {
            timeDiskNs.addAndGet(deltaNs);
        }

        public long getTimeDiskNs() {
            return timeDiskNs.get();
        }

        public double getPercentDisk() {
            long timeDiskMs = getTimeDiskNs() / Time.NS_PER_MS;
            return (timeDiskMs * 100.0) / (System.currentTimeMillis() - startedMs);
        }

        public double getPercentNetwork() {
            long timeNetworkMs = getTimeNetworkNs() / Time.NS_PER_MS;
            return (timeNetworkMs * 100.0) / (System.currentTimeMillis() - startedMs);
        }

        @Override
        public String toString() {
            return "Handle{" +
                   "id=" + id +
                   ", operation=" + operation +
                   ", startedMs=" + startedMs +
                   ", partitionIds=" + partitionIds +
                   ", entriesScanned=" + getEntriesScanned() +
                   ", finished=" + finished +
                   ", entriesPerSecond=" + getEntriesPerSecond() +
                   ", timeDiskNs=" + getTimeDiskNs() +
                   ", timeNetworkNs=" + getTimeNetworkNs() +
                   ", percentDisk=" + getPercentDisk() +
                   ", percentNetwork=" + getPercentNetwork() +
                   '}';
        }
    }
}

@@ -149,20 +149,104 @@
 va.util.
-List
+Collections;%0Aimport java.util.LinkedHashMap;%0Aimport java.util.List;%0Aimport java.util.Map
 ;%0Aimport
@@ -417,32 +417,70 @@
 private 
-final Concurrent
+static final int MAX_ENTRIES = 64;%0A%0A    private final 
 Map%3CLong
@@ -749,44 +749,58 @@
 s = 
-new ConcurrentHashMap%3CLong, Handle%3E(
+Collections.synchronizedMap(new Cache(MAX_ENTRIES)
 );%0A 
@@ -2951,32 +2951,440 @@
  DELETE,%0A    %7D%0A%0A
+    private static class Cache extends LinkedHashMap%3CLong, Handle%3E %7B%0A        private final int maxEntries;%0A%0A        public Cache(int maxEntries) %7B%0A            super();%0A            this.maxEntries = maxEntries;%0A        %7D%0A%0A        @Override%0A        protected boolean removeEldestEntry(Map.Entry%3CLong, Handle%3E eldest) %7B%0A            return eldest.getValue().isFinished() && size() %3E maxEntries;%0A        %7D%0A    %7D%0A%0A
     public stati

Use an LRU cache to keep track of handles