package com.geektime.consistenthash.geektime;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;


@Slf4j
class ConsistentHashTest {

    private static Set<String> requests = new HashSet<>();

    private static final int requestCount = 10000;

    private static final int physicalNodeCount = 10;

    private static Map<Integer, Double> trend = new LinkedHashMap<>();

    static {
        initRequests();
    }

    @Test
    public void nodeRouter() {

        for (int virtualNodeCount = 0; virtualNodeCount < 300; virtualNodeCount++) {


            ConsistentHash consistentHash = new ConsistentHash(physicalNodeCount, virtualNodeCount);

            TreeMap<String, Integer> nodeLoad = new TreeMap<>();
            requests.forEach(ele -> {
                String nodeName = consistentHash.nodeRouter(ele);
                if (nodeLoad.containsKey(nodeName)) {
                    Integer load = nodeLoad.get(nodeName);
                    nodeLoad.replace(nodeName, load + 1);
                } else {
                    nodeLoad.put(nodeName, 1);
                }
            });

            double standardDeviation = getStandardDeviation(nodeLoad, virtualNodeCount);
//            log.info("============= physicalNodeCount = {}, virtualNodeCount = {}, standardDeviation = {}", physicalNodeCount, virtualNodeCount, standardDeviation);

            trend.put(virtualNodeCount, standardDeviation);
        }

        trend.forEach((key, val) -> {
            log.info("{},{}", key, val);
        });
    }

    private static void initRequests() {
        for (int index = 0; index < requestCount; index++) {
            UUID uuid = UUID.randomUUID();
            requests.add(uuid.toString());
        }
//        log.info("---------------------------- requests.size = {}", requests.size());
    }

    private double getStandardDeviation(Map<String, Integer> nodeLoad, Integer virtualNodeCount) {
        int sum = nodeLoad.values().stream().mapToInt(Integer::intValue).sum();
        float average = (float) sum / (float) physicalNodeCount;
        AtomicReference<Double> deviation = new AtomicReference<>(0.0);

        nodeLoad.values().forEach(ele -> {
//            deviation.updateAndGet(v -> (float) (v + Math.pow(ele - average, 2)));
            deviation.updateAndGet(v -> v + (ele - average) * (ele - average));

        });
        return Math.sqrt(deviation.get() / (float) physicalNodeCount);

    }
}

package com.geektime.consistenthash.geektime;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.TreeMap;
import java.util.UUID;
import java.util.stream.Collectors;

@Setter
@Getter
@Slf4j
public class ConsistentHash {

    private TreeMap<Integer, String> CYCLE = new TreeMap<>();
    private TreeMap<Integer, String> PHYSICAL_NODES = new TreeMap<>();
    private TreeMap<Integer, String> VIRTUAL_NODES = new TreeMap<>();

    private int physicalNodeCount = 0;
    private int virtualNodeCount = 0;

    public ConsistentHash(int physicalNodeCount, int virtualNodeCount) {
        setPhysicalNodeCount(physicalNodeCount);
        setVirtualNodeCount(virtualNodeCount);
        initPhysicalNodes();
        initVirtualNodes();
        initCycle();
//        log.info("***************************first CYCLE ele = {}, last CYCLE ele = {}, CYCLE.eles.count = {}", CYCLE.firstEntry().getValue(), CYCLE.lastEntry().getValue(), CYCLE.size());
    }

    private void initCycle() {
        CYCLE.putAll(PHYSICAL_NODES);
        CYCLE.putAll(VIRTUAL_NODES);

    }

    private void initPhysicalNodes() {
        for (int index = 0; index < physicalNodeCount; index++) {
            String nodeName = "node_" + index;
            int hashCode = UUID.randomUUID().hashCode();
            PHYSICAL_NODES.put(hashCode, nodeName);
        }
    }

    private void initVirtualNodes() {
        PHYSICAL_NODES.values().forEach(cNodeName -> {
            for (int index = 0; index < virtualNodeCount; index++) {
                String cVirtualNodeName = cNodeName + ".virtualNode_" + index;
                int hashCode = UUID.randomUUID().hashCode();
                VIRTUAL_NODES.put(hashCode, cVirtualNodeName);
            }
        });
    }

    public String nodeRouter(String requestKey) {
        int hashCode = requestKey.hashCode();
        int l_targetNodeHash = Integer.MIN_VALUE;
        int r_targetNodeHash = Integer.MAX_VALUE;
        List<Integer> collect = CYCLE.keySet()
                .stream()
                .sorted()
                .collect(Collectors.toList());
        for (int cIndex = 0; cIndex < collect.size(); cIndex++) {
            if (collect.size() - 1 == cIndex) {
                l_targetNodeHash = collect.get(cIndex - 2);
                r_targetNodeHash = collect.get(cIndex - 1);
                break;
            }

            if (collect.get(cIndex) > hashCode) {


                if (0 == cIndex) {
                    l_targetNodeHash = collect.get(cIndex);
                    r_targetNodeHash = collect.get(cIndex + 1);
                    break;
                }

                l_targetNodeHash = collect.get(cIndex - 1);
                r_targetNodeHash = collect.get(cIndex);
                break;
            }
        }

        int hash;
        if (Math.abs(l_targetNodeHash - hashCode) < Math.abs(r_targetNodeHash - hashCode)) {
            hash = l_targetNodeHash;
        } else {
            hash = r_targetNodeHash;
        }

        String nodeName = CYCLE.get(hash);
        try {
            if (nodeName.contains(".")) {
                int index = nodeName.indexOf(".");
                return nodeName.substring(0, index);
            }
            return nodeName;
        } catch (IndexOutOfBoundsException e) {
            log.error("============================ node name = {}", nodeName);
            return "";
        } catch (NullPointerException e) {
            log.error("**************************** node name = {}", nodeName);
            return "";
        }
    }


}