Коротких Виктор / ИТМО DWS / Stage 6

src/main/java/ru/vk/itmo/test/viktorkorotkikh/LSMServerImpl.java

@@ -18,11 +18,12 @@
 import ru.vk.itmo.test.viktorkorotkikh.dao.exceptions.LSMDaoOutOfMemoryException;
 import ru.vk.itmo.test.viktorkorotkikh.dao.exceptions.TooManyFlushesException;
 import ru.vk.itmo.test.viktorkorotkikh.http.LSMCustomSession;
+import ru.vk.itmo.test.viktorkorotkikh.http.LSMRangeWriter;
 import ru.vk.itmo.test.viktorkorotkikh.http.LSMServerResponseWithMemorySegment;
+import ru.vk.itmo.test.viktorkorotkikh.util.ClusterResponseMerger;
 import ru.vk.itmo.test.viktorkorotkikh.util.HttpResponseNodeResponse;
 import ru.vk.itmo.test.viktorkorotkikh.util.LSMConstantResponse;
 import ru.vk.itmo.test.viktorkorotkikh.util.LsmServerUtil;
-import ru.vk.itmo.test.viktorkorotkikh.util.NodeResponse;
 import ru.vk.itmo.test.viktorkorotkikh.util.OneNioNodeResponse;
 import ru.vk.itmo.test.viktorkorotkikh.util.ReplicaEmptyResponse;
 import ru.vk.itmo.test.viktorkorotkikh.util.RequestParameters;
@@ -35,18 +36,14 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.time.Instant;
-import java.util.ArrayList;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
 import java.util.SequencedSet;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import static java.net.HttpURLConnection.HTTP_GATEWAY_TIMEOUT;
 import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
@@ -66,13 +63,17 @@ public class LSMServerImpl extends HttpServer {
     private final String selfUrl;
     private final HttpClient clusterClient;
     private final ServiceConfig serviceConfig;
+    private final ExecutorService clusterResponseProcessor;
+    private final ExecutorService localProcessor;
 
     public LSMServerImpl(
             ServiceConfig serviceConfig,
             Dao<MemorySegment, TimestampedEntry<MemorySegment>> dao,
             ExecutorService executorService,
             ConsistentHashingManager consistentHashingManager,
-            HttpClient clusterClient
+            HttpClient clusterClient,
+            ExecutorService clusterResponseProcessor,
+            ExecutorService localProcessor
     ) throws IOException {
         super(createServerConfig(serviceConfig));
         this.dao = dao;
@@ -81,6 +82,8 @@ public LSMServerImpl(
         this.selfUrl = serviceConfig.selfUrl();
         this.clusterClient = clusterClient;
         this.serviceConfig = serviceConfig;
+        this.clusterResponseProcessor = clusterResponseProcessor;
+        this.localProcessor = localProcessor;
     }
 
     private static HttpServerConfig createServerConfig(ServiceConfig serviceConfig) {
@@ -103,6 +106,12 @@ public void handleRequest(Request request, HttpSession session) throws IOExcepti
             executorService.execute(() -> {
                 try {
                     final String path = request.getPath();
+
+                    if (path.startsWith("/v0/entities") && request.getMethod() == METHOD_GET) {
+                        handleEntitiesRangeRequest(request, (LSMCustomSession) session);
+                        return;
+                    }
+
                     if (path.startsWith("/v0/entity")) {
                         handleEntityRequest(request, session);
                     } else {
@@ -186,15 +195,15 @@ public void handleEntityRequest(Request request, HttpSession session) throws IOE
 
         final SequencedSet<String> replicas = consistentHashingManager.getReplicasSet(from, key);
 
-        final Response response = getResponseFromReplicas(
+        getResponseFromReplicas(
                 request,
                 from,
                 replicas,
                 key,
                 id,
-                ack
+                ack,
+                session
         );
-        session.sendResponse(response);
     }
 
     private static RequestParameters getRequestParameters(Request request) {
@@ -220,34 +229,38 @@ private static RequestParameters getRequestParameters(Request request) {
         return new RequestParameters(id, ack, from);
     }
 
-    private Response getResponseFromReplicas(
+    private void getResponseFromReplicas(
             Request request,
-            Integer from,
+            int from,
             SequencedSet<String> replicas,
             byte[] key,
             String id,
-            Integer ack
+            int ack,
+            HttpSession session
     ) {
-        final List<NodeResponse> responses = new ArrayList<>(from);
+        final ClusterResponseMerger clusterResponseMerger = new ClusterResponseMerger(ack, from, request, session);
         final long requestTimestamp = Instant.now().toEpochMilli();
+        int i = 0;
         for (final String replicaUrl : replicas) {
             if (replicaUrl.equals(selfUrl)) {
-                responses.add(processLocal(request, key, id, requestTimestamp));
+                processLocalAsync(request, key, id, requestTimestamp, i, clusterResponseMerger);
             } else {
-                responses.add(processRemote(request, replicaUrl, id, requestTimestamp));
+                processRemote(request, replicaUrl, id, requestTimestamp, i, clusterResponseMerger);
             }
+            i++;
         }
-        return LsmServerUtil.mergeReplicasResponses(request, responses, ack);
     }
 
-    private NodeResponse processRemote(
+    private void processRemote(
             final Request originalRequest,
             final String server,
             final String id,
-            long requestTimestamp
+            long requestTimestamp,
+            int index,
+            ClusterResponseMerger clusterResponseMerge
     ) {
         final HttpRequest request = createClusterRequest(originalRequest, server, id, requestTimestamp);
-        return sendClusterRequest(request);
+        sendClusterRequest(request, index, clusterResponseMerge);
     }
 
     private static HttpRequest createClusterRequest(
@@ -259,7 +272,8 @@ private static HttpRequest createClusterRequest(
         HttpRequest.Builder builder = HttpRequest.newBuilder()
                 .uri(URI.create(server + ENTITY_V0_PATH_WITH_ID_PARAM + id))
                 .header(REPLICA_REQUEST_HEADER, "")
-                .header(LsmServerUtil.TIMESTAMP_HEADER, String.valueOf(requestTimestamp));
+                .header(LsmServerUtil.TIMESTAMP_HEADER, String.valueOf(requestTimestamp))
+                .timeout(Duration.ofMillis(CLUSTER_REQUEST_TIMEOUT_MILLISECONDS));
         switch (originalRequest.getMethod()) {
             case METHOD_GET -> builder.GET();
             case METHOD_PUT -> {
@@ -275,29 +289,47 @@ private static HttpRequest createClusterRequest(
         return builder.build();
     }
 
-    private NodeResponse sendClusterRequest(
-            final HttpRequest request
+    private void sendClusterRequest(
+            final HttpRequest request,
+            final int index,
+            final ClusterResponseMerger clusterResponseMerger
     ) {
-        try {
-            return new HttpResponseNodeResponse(
-                    clusterClient
-                            .sendAsync(request, HttpResponse.BodyHandlers.ofByteArray())
-                            .get(CLUSTER_REQUEST_TIMEOUT_MILLISECONDS, TimeUnit.MILLISECONDS)
-            );
-        } catch (InterruptedException e) {
-            final String clusterUrl = request.uri().toString();
-            Thread.currentThread().interrupt();
-            log.warn("Current thread was interrupted during processing request to cluster {}", clusterUrl);
-            return new ReplicaEmptyResponse(HTTP_UNAVAILABLE);
-        } catch (ExecutionException e) {
-            final String clusterUrl = request.uri().toString();
-            log.error("Unexpected exception occurred while sending request to cluster {}", clusterUrl, e);
-            return new ReplicaEmptyResponse(HTTP_UNAVAILABLE);
-        } catch (TimeoutException e) {
-            final String clusterUrl = request.uri().toString();
-            log.warn("Request timeout to cluster server {}", clusterUrl);
-            return new ReplicaEmptyResponse(HTTP_GATEWAY_TIMEOUT);
-        }
+        clusterClient
+                .sendAsync(request, HttpResponse.BodyHandlers.ofByteArray())
+                .thenAcceptAsync(response ->
+                        clusterResponseMerger.addToMerge(
+                                index,
+                                new HttpResponseNodeResponse(response)
+                        ), clusterResponseProcessor)
+                .exceptionallyAsync(throwable -> {
+                    if (throwable.getCause() instanceof java.net.http.HttpTimeoutException) {
+                        final String clusterUrl = request.uri().toString();
+                        log.warn("Request timeout to cluster server {}", clusterUrl);
+                        clusterResponseMerger.addToMerge(index, new ReplicaEmptyResponse(HTTP_GATEWAY_TIMEOUT));
+                    } else {
+                        final String clusterUrl = request.uri().toString();
+                        log.error(
+                                "Unexpected exception occurred while sending request to cluster {}",
+                                clusterUrl,
+                                throwable
+                        );
+                        clusterResponseMerger.addToMerge(index, new ReplicaEmptyResponse(HTTP_UNAVAILABLE));
+                    }
+                    return null;
+                }, clusterResponseProcessor).state();
+    }
+
+    private void processLocalAsync(
+            Request request,
+            byte[] key,
+            String id,
+            long requestTimestamp,
+            int i,
+            ClusterResponseMerger clusterResponseMerger
+    ) {
+        localProcessor.execute(() ->
+                clusterResponseMerger.addToMerge(i, processLocal(request, key, id, requestTimestamp))
+        );
     }
 
     private OneNioNodeResponse processLocal(
@@ -406,6 +438,21 @@ public Response handleCompact(Request request) throws IOException {
         return LSMConstantResponse.ok(request);
     }
 
+    public void handleEntitiesRangeRequest(Request request, LSMCustomSession session) throws IOException {
+        final String start = request.getParameter("start=");
+        final String end = request.getParameter("end=");
+        if (start == null || start.isEmpty() || (end != null && end.isEmpty())) {
+            log.debug("Bad request: start parameter and end parameter (if it present) should not be empty");
+            session.sendResponse(LSMConstantResponse.badRequest(request));
+            return;
+        }
+
+        final MemorySegment startMemorySegment = fromByteArray(start.getBytes(StandardCharsets.UTF_8));
+        final MemorySegment endMemorySegment = end == null ? null : fromByteArray(end.getBytes(StandardCharsets.UTF_8));
+        Iterator<TimestampedEntry<MemorySegment>> iterator = dao.get(startMemorySegment, endMemorySegment);
+        session.sendRangeResponse(new LSMRangeWriter(iterator, LSMConstantResponse.keepAlive(request)));
+    }
+
     private static MemorySegment fromByteArray(final byte[] data) {
         return MemorySegment.ofArray(data);
     }

src/main/java/ru/vk/itmo/test/viktorkorotkikh/LSMServiceImpl.java

@@ -30,6 +30,14 @@
 public class LSMServiceImpl implements Service {
     private static final long FLUSH_THRESHOLD = 1 << 20; // 1 MB
     private static final int TERMINATION_TIMEOUT_SECONDS = 20;
+    private static final int SERVER_EXECUTOR_SERVICE_THREADS_COUNT = 16;
+    private static final int SERVER_EXECUTOR_SERVICE_QUEUE_SIZE = 1024;
+    private static final int CLUSTER_HTTP_CLIENT_EXECUTOR_SERVICE_THREADS_COUNT = 16;
+    private static final int CLUSTER_HTTP_CLIENT_EXECUTOR_SERVICE_QUEUE_SIZE = 1024;
+    private static final int CLUSTER_RESPONSE_EXECUTOR_SERVICE_THREADS_COUNT = 16;
+    private static final int CLUSTER_RESPONSE_EXECUTOR_SERVICE_QUEUE_SIZE = 1024;
+    private static final int LOCAL_REQUEST_EXECUTOR_SERVICE_THREADS_COUNT = 16;
+    private static final int LOCAL_REQUEST_EXECUTOR_SERVICE_QUEUE_SIZE = 1024;
     private final ServiceConfig serviceConfig;
     private LSMServerImpl httpServer;
     private boolean isRunning;
@@ -38,6 +46,8 @@ public class LSMServiceImpl implements Service {
     private final ConsistentHashingManager consistentHashingManager;
     private HttpClient clusterClient;
     private ExecutorService clusterClientExecutorService;
+    private ExecutorService clusterResponseProcessor;
+    private ExecutorService localProcessor;
 
     public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
         Path baseWorkingDir = Path.of("daoWorkingDir");
@@ -96,14 +106,43 @@ public LSMServiceImpl(ServiceConfig serviceConfig) {
         this.consistentHashingManager = new ConsistentHashingManager(10, serviceConfig.clusterUrls());
     }
 
-    private static LSMServerImpl createServer(
-            ServiceConfig serviceConfig,
-            Dao<MemorySegment, TimestampedEntry<MemorySegment>> dao,
-            ExecutorService executorService,
-            ConsistentHashingManager consistentHashingManager,
-            HttpClient clusterClient
+    private LSMServerImpl createServer(
+            Dao<MemorySegment, TimestampedEntry<MemorySegment>> dao
     ) throws IOException {
-        return new LSMServerImpl(serviceConfig, dao, executorService, consistentHashingManager, clusterClient);
+        executorService = createExecutorService(
+                SERVER_EXECUTOR_SERVICE_THREADS_COUNT,
+                SERVER_EXECUTOR_SERVICE_QUEUE_SIZE,
+                "worker"
+        );
+        clusterClientExecutorService = createExecutorService(
+                CLUSTER_HTTP_CLIENT_EXECUTOR_SERVICE_THREADS_COUNT,
+                CLUSTER_HTTP_CLIENT_EXECUTOR_SERVICE_QUEUE_SIZE,
+                "cluster-request"
+        );
+        clusterResponseProcessor = createExecutorService(
+                CLUSTER_RESPONSE_EXECUTOR_SERVICE_THREADS_COUNT,
+                CLUSTER_RESPONSE_EXECUTOR_SERVICE_QUEUE_SIZE,
+                "cluster-response-processor"
+        );
+        localProcessor = createExecutorService(
+                LOCAL_REQUEST_EXECUTOR_SERVICE_THREADS_COUNT,
+                LOCAL_REQUEST_EXECUTOR_SERVICE_QUEUE_SIZE,
+                "local-processor"
+        );
+
+        clusterClient = HttpClient.newBuilder()
+                .executor(clusterClientExecutorService)
+                .build();
+
+        return new LSMServerImpl(
+                serviceConfig,
+                dao,
+                executorService,
+                consistentHashingManager,
+                clusterClient,
+                clusterResponseProcessor,
+                localProcessor
+        );
     }
 
     private static Dao<MemorySegment, TimestampedEntry<MemorySegment>> createLSMDao(Path workingDir) {
@@ -146,14 +185,7 @@ public synchronized CompletableFuture<Void> start() throws IOException {
         if (isRunning) return CompletableFuture.completedFuture(null);
         dao = createLSMDao(serviceConfig.workingDir());
 
-        executorService = createExecutorService(16, 1024, "worker");
-        clusterClientExecutorService = createExecutorService(16, 1024, "cluster-worker");
-
-        clusterClient = HttpClient.newBuilder()
-                .executor(clusterClientExecutorService)
-                .build();
-
-        httpServer = createServer(serviceConfig, dao, executorService, consistentHashingManager, clusterClient);
+        httpServer = createServer(dao);
         httpServer.start();
 
         isRunning = true;
@@ -167,6 +199,8 @@ public synchronized CompletableFuture<Void> stop() throws IOException {
 
         shutdownHttpClient(clusterClient);
         shutdownExecutorService(clusterClientExecutorService);
+        shutdownExecutorService(clusterResponseProcessor);
+        shutdownExecutorService(localProcessor);
         shutdownExecutorService(executorService);
         executorService = null;
         clusterClient = null;
@@ -211,7 +245,7 @@ private static void shutdownHttpClient(HttpClient httpClient) {
         }
     }
 
-    @ServiceFactory(stage = 4)
+    @ServiceFactory(stage = 6)
     public static class LSMServiceFactoryImpl implements ServiceFactory.Factory {
         @Override
         public Service create(ServiceConfig config) {

src/main/java/ru/vk/itmo/test/viktorkorotkikh/dao/MemTable.java

@@ -4,6 +4,7 @@
 
 import java.lang.foreign.MemorySegment;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.NavigableMap;
 import java.util.NoSuchElementException;
@@ -39,7 +40,12 @@ private Iterator<TimestampedEntry<MemorySegment>> storageIterator(MemorySegment
             return storage.tailMap(from).sequencedValues().iterator();
         }
 
-        return storage.subMap(from, to).sequencedValues().iterator();
+        try {
+            return storage.subMap(from, to).sequencedValues().iterator();
+        } catch (IllegalArgumentException illegalArgumentException) {
+            // we get inconsistent range error when from > to
+            return Collections.emptyIterator();
+        }
     }
 
     public MemTableIterator iterator(MemorySegment from, MemorySegment to, int priorityReduction) {