Stage6 Emelyanov Pavel

src/main/java/ru/vk/itmo/test/pavelemelyanov/ConsistentHashing.java

@@ -0,0 +1,86 @@
+package ru.vk.itmo.test.pavelemelyanov;
+
+import one.nio.util.Hash;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NavigableMap;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+public class ConsistentHashing {
+    private final NavigableMap<Integer, String> virtualNodeMapping = new TreeMap<>();
+    private final List<String> clusterUrls;
+
+    public ConsistentHashing(List<String> clusterUrls, int numbOfVirtualNodes) {
+        this.clusterUrls = clusterUrls;
+        for (String clusterUrl : clusterUrls) {
+            for (int i = 0; i < numbOfVirtualNodes; i++) {
+                addNode(i, clusterUrl);
+            }
+        }
+    }
+
+    public String getNode(String key) {
+        if (virtualNodeMapping.isEmpty()) {
+            return null;
+        }
+
+        final int hash = getHash(key);
+        SortedMap<Integer, String> tailMap = virtualNodeMapping.tailMap(hash);
+        return (tailMap.isEmpty() ? virtualNodeMapping.firstEntry() : tailMap.firstEntry()).getValue();
+    }
+
+    public List<String> getNodes(String key, int from) {
+        List<String> res = new ArrayList<>();
+
+        if (key != null && from > 0) {
+            if (from < clusterUrls.size()) {
+                int slot = getHash(key);
+                Iterator<String> it = new ClockwiseIterator(slot);
+                while (it.hasNext() && res.size() < from) {
+                    String part = it.next();
+                    res.add(part);
+                }
+            } else {
+                res.addAll(clusterUrls);
+            }
+        }
+
+        return res;
+    }
+
+    private int getHash(String key) {
+        return Hash.murmur3(key);
+    }
+
+    private void addNode(int numOfNode, String node) {
+        int hash = getHash(node + numOfNode);
+        virtualNodeMapping.put(hash, node);
+    }
+
+    private class ClockwiseIterator implements Iterator<String> {
+        private final Iterator<String> head;
+        private final Iterator<String> tail;
+
+        public ClockwiseIterator(int slot) {
+            head = virtualNodeMapping.headMap(slot)
+                    .values()
+                    .iterator();
+            tail = virtualNodeMapping.tailMap(slot)
+                    .values()
+                    .iterator();
+        }
+
+        @Override
+        public boolean hasNext() {
+            return head.hasNext() || tail.hasNext();
+        }
+
+        @Override
+        public String next() {
+            return tail.hasNext() ? tail.next() : head.next();
+        }
+    }
+}

src/main/java/ru/vk/itmo/test/pavelemelyanov/ExecutorServiceConfig.java

@@ -1,16 +1,21 @@
 package ru.vk.itmo.test.pavelemelyanov;
 
+import one.nio.async.CustomThreadFactory;
+
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 public final class ExecutorServiceConfig {
     public static final int CORE_AMOUNT = Runtime.getRuntime().availableProcessors();
-    public static final int CORE_POOL_SIZE = CORE_AMOUNT / 2;
+    public static final int CORE_POOL_SIZE = CORE_AMOUNT;
     public static final int MAX_CORE_POOL_SIZE = CORE_AMOUNT;
-    public static final int KEEP_ALIVE_TIME = 200;
-    public static final int QUEUE_CAPACITY = 64;
+    public static final long KEEP_ALIVE_TIME = 60;
+    public static final TimeUnit UNIT = TimeUnit.SECONDS;
+    public static final int QUEUE_CAPACITY = 1000;
+    public static final CustomThreadFactory threadFactory = new CustomThreadFactory("t", true);
     public static final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);
     public static final RejectedExecutionHandler HANDLER = new ThreadPoolExecutor.AbortPolicy();
 

src/main/java/ru/vk/itmo/test/pavelemelyanov/ExecutorServiceWrapper.java

@@ -0,0 +1,46 @@
+package ru.vk.itmo.test.pavelemelyanov;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class ExecutorServiceWrapper {
+    public static final int TERMINATION_TIMEOUT = 60;
+    private static final Logger LOG = LoggerFactory.getLogger(MyServer.class);
+
+    private final ExecutorService workingPool;
+
+    public ExecutorServiceWrapper() {
+        workingPool = new ThreadPoolExecutor(
+                ExecutorServiceConfig.CORE_POOL_SIZE,
+                ExecutorServiceConfig.MAX_CORE_POOL_SIZE,
+                ExecutorServiceConfig.KEEP_ALIVE_TIME,
+                ExecutorServiceConfig.UNIT,
+                ExecutorServiceConfig.queue,
+                ExecutorServiceConfig.threadFactory,
+                ExecutorServiceConfig.HANDLER
+        );
+    }
+
+    public ExecutorService getExecutorService() {
+        return workingPool;
+    }
+
+    public void shutdownAndAwaitTermination() {
+        workingPool.shutdown();
+        try {
+            if (!workingPool.awaitTermination(TERMINATION_TIMEOUT, TimeUnit.SECONDS)) {
+                workingPool.shutdownNow();
+                if (!workingPool.awaitTermination(TERMINATION_TIMEOUT, TimeUnit.SECONDS)) {
+                    LOG.error("ExecutorService error with stopping");
+                }
+            }
+        } catch (InterruptedException ex) {
+            workingPool.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+    }
+}

src/main/java/ru/vk/itmo/test/pavelemelyanov/FactoryImpl.java

@@ -4,7 +4,7 @@
 import ru.vk.itmo.ServiceConfig;
 import ru.vk.itmo.test.ServiceFactory;
 
-@ServiceFactory(stage = 1)
+@ServiceFactory(stage = 6)
 public class FactoryImpl implements ServiceFactory.Factory {
 
     @Override

src/main/java/ru/vk/itmo/test/pavelemelyanov/HttpUtils.java

@@ -0,0 +1,34 @@
+package ru.vk.itmo.test.pavelemelyanov;
+
+import one.nio.http.Request;
+import one.nio.http.Response;
+
+import java.net.HttpURLConnection;
+import java.util.Map;
+import java.util.Set;
+
+public final class HttpUtils {
+    public static final String NOT_ENOUGH_REPLICAS = "504 Not Enough Replicas";
+    public static final Set<Integer> SUPPORTED_METHODS = Set.of(
+            Request.METHOD_GET,
+            Request.METHOD_PUT,
+            Request.METHOD_DELETE
+    );
+    public static final int REQUEST_TIMEOUT_IN_MILLIS = 3000;
+    public static final Map<Integer, String> HTTP_CODE = Map.of(
+            HttpURLConnection.HTTP_OK, Response.OK,
+            HttpURLConnection.HTTP_ACCEPTED, Response.ACCEPTED,
+            HttpURLConnection.HTTP_CREATED, Response.CREATED,
+            HttpURLConnection.HTTP_NOT_FOUND, Response.NOT_FOUND,
+            HttpURLConnection.HTTP_BAD_REQUEST, Response.BAD_REQUEST,
+            HttpURLConnection.HTTP_INTERNAL_ERROR, Response.INTERNAL_ERROR
+    );
+    public static final int NUMBER_OF_VIRTUAL_NODES = 100;
+    public static final String HTTP_TIMESTAMP_HEADER = "X-Timestamp";
+    public static final String NIO_TIMESTAMP_HEADER = "x-timestamp:";
+    public static final String HTTP_TERMINATION_HEADER = "X-Termination";
+
+    private HttpUtils() {
+
+    }
+}