From 8351bec59a819a18757b7c5733c09403f054c90b Mon Sep 17 00:00:00 2001
From: Stepan Usatiuk <stepan@usatiuk.com>
Date: Mon, 31 Mar 2025 16:20:10 +0200
Subject: [PATCH] Revert "Server: parallel op sending"

This reverts commit 0f8002dc2ce5bde6d129e141d91cfea75f8cf4d3.
---
 .../invalidation/InvalidationQueueService.java  |  6 +-----
 .../src/main/resources/application.properties   |  2 +-
 .../com/usatiuk/dhfs/integration/ResyncIT.java  | 17 +++--------------
 3 files changed, 5 insertions(+), 20 deletions(-)

diff --git a/dhfs-parent/server/src/main/java/com/usatiuk/dhfs/repository/invalidation/InvalidationQueueService.java b/dhfs-parent/server/src/main/java/com/usatiuk/dhfs/repository/invalidation/InvalidationQueueService.java
index 87710049..83df5292 100644
--- a/dhfs-parent/server/src/main/java/com/usatiuk/dhfs/repository/invalidation/InvalidationQueueService.java
+++ b/dhfs-parent/server/src/main/java/com/usatiuk/dhfs/repository/invalidation/InvalidationQueueService.java
@@ -1,6 +1,5 @@
 package com.usatiuk.dhfs.repository.invalidation;
 
-import com.usatiuk.dhfs.utils.DataLocker;
 import com.usatiuk.objects.JObjectKey;
 import com.usatiuk.dhfs.PeerId;
 import com.usatiuk.dhfs.repository.PeerManager;
@@ -16,7 +15,6 @@ import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.event.Observes;
 import jakarta.inject.Inject;
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
-import org.apache.commons.lang3.tuple.Pair;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
 import java.util.concurrent.ExecutorService;
@@ -41,8 +39,6 @@ public class InvalidationQueueService {
     @Inject
     PersistentPeerDataService persistentPeerDataService;
 
-    private final DataLocker _pushLocker = new DataLocker();
-
     private ExecutorService _executor;
     private volatile boolean _shutdown = false;
 
@@ -124,7 +120,7 @@ public class InvalidationQueueService {
                             continue;
                         }
 
-                        try (var lock = _pushLocker.lock(Pair.of(e.peer(), e.key()))) {
+                        try {
                             opPusher.doPush(e);
                             success++;
                         } catch (Exception ex) {
diff --git a/dhfs-parent/server/src/main/resources/application.properties b/dhfs-parent/server/src/main/resources/application.properties
index 5c230fd5..b11a23ee 100644
--- a/dhfs-parent/server/src/main/resources/application.properties
+++ b/dhfs-parent/server/src/main/resources/application.properties
@@ -4,7 +4,7 @@ dhfs.objects.peerdiscovery.interval=4s
 dhfs.objects.peerdiscovery.broadcast=true
 dhfs.objects.sync.timeout=30
 dhfs.objects.sync.ping.timeout=5
-dhfs.objects.invalidation.threads=4
+dhfs.objects.invalidation.threads=1
 dhfs.objects.invalidation.delay=1000
 dhfs.objects.reconnect_interval=5s
 dhfs.objects.write_log=false
diff --git a/dhfs-parent/server/src/test/java/com/usatiuk/dhfs/integration/ResyncIT.java b/dhfs-parent/server/src/test/java/com/usatiuk/dhfs/integration/ResyncIT.java
index 864e768f..48fc8c8b 100644
--- a/dhfs-parent/server/src/test/java/com/usatiuk/dhfs/integration/ResyncIT.java
+++ b/dhfs-parent/server/src/test/java/com/usatiuk/dhfs/integration/ResyncIT.java
@@ -123,24 +123,13 @@ public class ResyncIT {
 
         waitingConsumer2.waitUntil(frame -> frame.getUtf8String().contains("Connected"), 60, TimeUnit.SECONDS);
         waitingConsumer1.waitUntil(frame -> frame.getUtf8String().contains("Connected"), 60, TimeUnit.SECONDS);
-        Thread.sleep(20000);
         await().atMost(120, TimeUnit.SECONDS).until(() -> {
             var foundWc2 = container2.execInContainer("/bin/sh", "-c", "find /root/dhfs_default/fuse -type f | wc -l");
-            int val = 0;
-            try {
-                val = Integer.valueOf(foundWc2.getStdout().strip());
-            } catch (NumberFormatException ignored) {
-            }
-            return 400 == val;
+            return 400 == Integer.valueOf(foundWc2.getStdout().strip());
         });
         await().atMost(120, TimeUnit.SECONDS).until(() -> {
-            var foundWc1 = container1.execInContainer("/bin/sh", "-c", "find /root/dhfs_default/fuse -type f | wc -l");
-            int val = 0;
-            try {
-                val = Integer.valueOf(foundWc1.getStdout().strip());
-            } catch (NumberFormatException ignored) {
-            }
-            return 400 == val;
+            var foundWc2 = container1.execInContainer("/bin/sh", "-c", "find /root/dhfs_default/fuse -type f | wc -l");
+            return 400 == Integer.valueOf(foundWc2.getStdout().strip());
         });
     }