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

src/main/java/ru/vk/itmo/test/viktorkorotkikh/util/ClusterResponseMerger.java

@@ -9,14 +9,15 @@
 import java.io.IOException;
 import java.net.HttpURLConnection;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReferenceArray;
 
 public class ClusterResponseMerger {
     private static final Logger log = LoggerFactory.getLogger(ClusterResponseMerger.class);
     private final int ack;
     private final int allowedUnsuccessfulResponses;
     private final Request originalRequest;
     private final HttpSession session;
-    private final NodeResponse[] nodeResponses;
+    private final AtomicReferenceArray<NodeResponse> nodeResponses;
     private final AtomicInteger unsuccessfulResponsesCount;
     private final AtomicInteger successfulResponsesCount;
 
@@ -25,13 +26,14 @@ public ClusterResponseMerger(int ack, int from, Request originalRequest, HttpSes
         this.allowedUnsuccessfulResponses = from - ack;
         this.originalRequest = originalRequest;
         this.session = session;
-        this.nodeResponses = new NodeResponse[from];
+        this.nodeResponses = new AtomicReferenceArray<>(from);
         this.unsuccessfulResponsesCount = new AtomicInteger();
         this.successfulResponsesCount = new AtomicInteger();
     }
 
     public void addToMerge(int index, NodeResponse response) {
-        nodeResponses[index] = response;
+        // we can write in plain semantic because we read in opaque in mergeReplicasResponses method
+        nodeResponses.setPlain(index, response);
         if (isSuccessfulResponse(response.statusCode())) {
             int newSuccessfulResponsesCount = successfulResponsesCount.incrementAndGet();
             if (newSuccessfulResponsesCount == ack) {

src/main/java/ru/vk/itmo/test/viktorkorotkikh/util/LsmServerUtil.java

@@ -7,6 +7,7 @@
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
 import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicReferenceArray;
 
 import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
 import static java.net.HttpURLConnection.HTTP_ENTITY_TOO_LARGE;
@@ -27,7 +28,7 @@ public static String timestampToHeader(long timestamp) {
 
     public static Response mergeReplicasResponses(
             final Request originalRequest,
-            final NodeResponse[] responses,
+            final AtomicReferenceArray<NodeResponse> responses,
             final int ack
     ) {
         switch (originalRequest.getMethod()) {
@@ -44,11 +45,16 @@ public static Response mergeReplicasResponses(
         }
     }
 
-    private static Response mergeGetResponses(Request originalRequest, NodeResponse[] responses, int ack) {
+    private static Response mergeGetResponses(
+            Request originalRequest,
+            AtomicReferenceArray<NodeResponse> responses,
+            int ack
+    ) {
         long maxTimestamp = -1;
         NodeResponse lastValue = null;
         int successfulResponses = 0;
-        for (NodeResponse response : responses) {
+        for (int i = 0; i < responses.length(); i++) {
+            final NodeResponse response = responses.getOpaque(i);
             if (response == null) continue;
             final long valueTimestamp = getTimestamp(response);
             if (valueTimestamp > maxTimestamp) {
@@ -78,7 +84,7 @@ private static Response mergeGetResponses(Request originalRequest, NodeResponse[
 
     private static Response mergePutResponses(
             Request originalRequest,
-            NodeResponse[] responses,
+            AtomicReferenceArray<NodeResponse> responses,
             int ack
     ) {
         if (hasNotEnoughReplicas(responses, ack)) {
@@ -89,7 +95,7 @@ private static Response mergePutResponses(
 
     private static Response mergeDeleteResponses(
             Request originalRequest,
-            NodeResponse[] responses,
+            AtomicReferenceArray<NodeResponse> responses,
             int ack
     ) {
         if (hasNotEnoughReplicas(responses, ack)) {
@@ -98,9 +104,10 @@ private static Response mergeDeleteResponses(
         return LSMConstantResponse.accepted(originalRequest);
     }
 
-    private static boolean hasNotEnoughReplicas(NodeResponse[] responses, int ack) {
+    private static boolean hasNotEnoughReplicas(AtomicReferenceArray<NodeResponse> responses, int ack) {
         int successfulResponses = 0;
-        for (NodeResponse response : responses) {
+        for (int i = 0; i < responses.length(); i++) {
+            final NodeResponse response = responses.getOpaque(i);
             if (response == null) continue;
             if (response.statusCode() >= 200 && response.statusCode() < 300) {
                 successfulResponses++;
@@ -117,8 +124,9 @@ private static long getTimestamp(final NodeResponse response) {
         return Long.parseLong(timestamp);
     }
 
-    private static NodeResponse firstNotNull(NodeResponse[] responses) {
-        for (NodeResponse response : responses) {
+    private static NodeResponse firstNotNull(AtomicReferenceArray<NodeResponse> responses) {
+        for (int i = 0; i < responses.length(); i++) {
+            final NodeResponse response = responses.getOpaque(i);
             if (response != null) return response;
         }
         throw new NoSuchElementException();