Merge pull request #5935 from jmacxx/dao_state_store_and_statistics

Improve DAO state store persistence and statistics logging [chimp1984]
bisq-network · Jan 3, 2022 · dde48f7 · dde48f7
2 parents 8d4a785 + 30706f0
commit dde48f7
Show file tree

Hide file tree

Showing 6 changed files with 47 additions and 20 deletions.
diff --git a/core/src/main/java/bisq/core/dao/monitoring/DaoStateMonitoringService.java b/core/src/main/java/bisq/core/dao/monitoring/DaoStateMonitoringService.java
@@ -127,7 +127,7 @@ public interface Listener {
     @Nullable
     private Runnable createSnapshotHandler;
     // Lookup map
-    private Map<Integer, DaoStateBlock> daoStateBlockByHeight = new HashMap<>();
+    private final Map<Integer, DaoStateBlock> daoStateBlockByHeight = new HashMap<>();
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////

diff --git a/core/src/main/java/bisq/core/dao/node/BsqNode.java b/core/src/main/java/bisq/core/dao/node/BsqNode.java
@@ -157,6 +157,7 @@ public void setWarnMessageHandler(@SuppressWarnings("NullableProblems") Consumer
 
     public void shutDown() {
         exportJsonFilesService.shutDown();
+        daoStateSnapshotService.shutDown();
     }
 
 

diff --git a/core/src/main/java/bisq/core/dao/state/DaoStateSnapshotService.java b/core/src/main/java/bisq/core/dao/state/DaoStateSnapshotService.java
@@ -122,6 +122,9 @@ public void addListeners() {
     public void start() {
     }
 
+    public void shutDown() {
+        daoStateStorageService.shutDown();
+    }
 
     ///////////////////////////////////////////////////////////////////////////////////////////
     // DaoStateListener

diff --git a/core/src/main/java/bisq/core/dao/state/storage/DaoStateStorageService.java b/core/src/main/java/bisq/core/dao/state/storage/DaoStateStorageService.java
@@ -33,11 +33,18 @@
 import javax.inject.Inject;
 import javax.inject.Named;
 
+import com.google.common.util.concurrent.MoreExecutors;
+
 import java.io.File;
 import java.io.IOException;
 
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -53,6 +60,8 @@ public class DaoStateStorageService extends StoreService<DaoStateStore> {
     private final BsqBlocksStorageService bsqBlocksStorageService;
     private final File storageDir;
     private final LinkedList<Block> blocks = new LinkedList<>();
+    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
+    private Optional<Future<?>> future = Optional.empty();
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -94,22 +103,36 @@ public void requestPersistence(protobuf.DaoState daoStateAsProto,
             return;
         }
 
-        new Thread(() -> {
-            Thread.currentThread().setName("Write-blocks-and-DaoState");
-            bsqBlocksStorageService.persistBlocks(blocks);
+        if (future.isPresent() && !future.get().isDone()) {
+            UserThread.runAfter(() -> requestPersistence(daoStateAsProto, blocks, daoStateHashChain, completeHandler), 2);
+            return;
+        }
 
-            store.setDaoStateAsProto(daoStateAsProto);
-            store.setDaoStateHashChain(daoStateHashChain);
-            long ts = System.currentTimeMillis();
-            persistenceManager.persistNow(() -> {
-                // After we have written to disk we remove the daoStateAsProto in the store to avoid that it stays in
-                // memory there until the next persist call.
-                log.info("Persist daoState took {} ms", System.currentTimeMillis() - ts);
-                store.releaseMemory();
-                GcUtil.maybeReleaseMemory();
-                UserThread.execute(completeHandler);
-            });
-        }).start();
+        future = Optional.of(executorService.submit(() -> {
+           try {
+               Thread.currentThread().setName("Write-blocks-and-DaoState");
+               bsqBlocksStorageService.persistBlocks(blocks);
+               store.setDaoStateAsProto(daoStateAsProto);
+               store.setDaoStateHashChain(daoStateHashChain);
+               long ts = System.currentTimeMillis();
+               persistenceManager.persistNow(() -> {
+                   // After we have written to disk we remove the daoStateAsProto in the store to avoid that it stays in
+                   // memory there until the next persist call.
+                   log.info("Persist daoState took {} ms", System.currentTimeMillis() - ts);
+                   store.releaseMemory();
+                   GcUtil.maybeReleaseMemory();
+                   UserThread.execute(completeHandler);
+               });
+           } catch (Exception e) {
+               log.error("Exception at persisting BSQ blocks and DaoState", e);
+           }
+        }));
+    }
+
+    public void shutDown() {
+        executorService.shutdown();
+        // noinspection UnstableApiUsage
+        MoreExecutors.shutdownAndAwaitTermination(executorService, 10, TimeUnit.SECONDS);
     }
 
     @Override

diff --git a/p2p/src/main/java/bisq/network/p2p/network/Statistic.java b/p2p/src/main/java/bisq/network/p2p/network/Statistic.java
@@ -72,7 +72,7 @@ public class Statistic {
             totalReceivedBytesPerSec.set(((double) totalReceivedBytes.get()) / passed);
         }, 1);
 
-        // We log statistics every 5 minutes
+        // We log statistics every 60 minutes
         UserThread.runPeriodically(() -> {
             String ls = System.lineSeparator();
             log.info("Accumulated network statistics:" + ls +
@@ -81,14 +81,14 @@ public class Statistic {
                             "Number of sent messages per sec: {};" + ls +
                             "Bytes received: {}" + ls +
                             "Number of received messages/Received messages: {} / {};" + ls +
-                            "Number of received messages per sec: {};" + ls,
+                            "Number of received messages per sec: {}" + ls,
                     Utilities.readableFileSize(totalSentBytes.get()),
                     numTotalSentMessages.get(), totalSentMessages,
                     numTotalSentMessagesPerSec.get(),
                     Utilities.readableFileSize(totalReceivedBytes.get()),
                     numTotalReceivedMessages.get(), totalReceivedMessages,
                     numTotalReceivedMessagesPerSec.get());
-        }, TimeUnit.MINUTES.toSeconds(5));
+        }, TimeUnit.MINUTES.toSeconds(60));
     }
 
     public static LongProperty totalSentBytesProperty() {

diff --git a/p2p/src/main/java/bisq/network/p2p/peers/PeerManager.java b/p2p/src/main/java/bisq/network/p2p/peers/PeerManager.java
@@ -174,7 +174,7 @@ public void onAwakeFromStandby(long missedMs) {
         };
         clockWatcher.addListener(clockWatcherListener);
 
-        printStatisticsTimer = UserThread.runPeriodically(this::printStatistics, TimeUnit.MINUTES.toSeconds(5));
+        printStatisticsTimer = UserThread.runPeriodically(this::printStatistics, TimeUnit.MINUTES.toSeconds(60));
     }
 
     public void shutDown() {