Merge pull request #6586 from alvasw/implement_network_statistics_ser…

…vice

Decouple NetworkStatistics from UI thread [1/2]

p2p/src/main/java/bisq/network/p2p/network/statistics/ConnectionStatistics.java

@@ -0,0 +1,111 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.network.p2p.network.statistics;
+
+import bisq.common.proto.network.NetworkEnvelope;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+public class ConnectionStatistics {
+
+    public interface Listener {
+        void onNewSentBytes(long numberOfNewBytes);
+
+        void onNewReceivedBytes(long numberOfNewBytes);
+
+        void onAddSentMessage(NetworkEnvelope networkEnvelope);
+
+        void onAddReceivedMessage(NetworkEnvelope networkEnvelope);
+    }
+
+    private final Date creationDate = new Date();
+    private final List<Listener> allListeners = new ArrayList<>();
+    private final Map<String, Integer> receivedMessages = new HashMap<>();
+    private final Map<String, Integer> sentMessages = new HashMap<>();
+
+    private long lastActivityTimestamp = System.currentTimeMillis();
+    private long sentBytes;
+    private long receivedBytes;
+    private int totalSentMessages;
+    private int totalReceivedMessages;
+    @Setter
+    private int roundTripTime;
+
+    public void addListener(Listener listener) {
+        allListeners.add(listener);
+    }
+
+    public void removeListener(Listener listener) {
+        allListeners.remove(listener);
+    }
+
+    void updateLastActivityTimestamp() {
+        lastActivityTimestamp = System.currentTimeMillis();
+    }
+
+    void addSentBytes(int value) {
+        sentBytes += value;
+        allListeners.forEach(listener -> listener.onNewSentBytes(value));
+    }
+
+    void addReceivedBytes(int value) {
+        receivedBytes += value;
+        allListeners.forEach(listener -> listener.onNewReceivedBytes(value));
+    }
+
+    void addSentMessage(NetworkEnvelope networkEnvelope) {
+        String messageClassName = networkEnvelope.getClass().getSimpleName();
+        sentMessages.merge(messageClassName, 1, Integer::sum);
+
+        totalSentMessages++;
+        allListeners.forEach(listener -> listener.onAddSentMessage(networkEnvelope));
+    }
+
+    void addReceivedMessage(NetworkEnvelope networkEnvelope) {
+        String messageClassName = networkEnvelope.getClass().getSimpleName();
+        receivedMessages.merge(messageClassName, 1, Integer::sum);
+
+        totalReceivedMessages++;
+        allListeners.forEach(listener -> listener.onAddReceivedMessage(networkEnvelope));
+    }
+
+    public long getLastActivityAge() {
+        return System.currentTimeMillis() - lastActivityTimestamp;
+    }
+
+    @Override
+    public String toString() {
+        return "ConnectionStatistics{" +
+                "\n     creationDate=" + creationDate +
+                ",\n     lastActivityTimestamp=" + lastActivityTimestamp +
+                ",\n     sentBytes=" + sentBytes +
+                ",\n     receivedBytes=" + receivedBytes +
+                ",\n     receivedMessages=" + receivedMessages +
+                ",\n     sentMessages=" + sentMessages +
+                ",\n     roundTripTime=" + roundTripTime +
+                "\n}";
+    }
+}

p2p/src/main/java/bisq/network/p2p/network/statistics/ConnectionStatsAccumulator.java

@@ -0,0 +1,91 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.network.p2p.network.statistics;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import lombok.Getter;
+
+@Getter
+public class ConnectionStatsAccumulator implements Runnable {
+    private final long startTime = System.currentTimeMillis();
+
+    private final List<ConnectionStatistics> allConnectionStatistics;
+    private final List<NetworkStatisticsService.Listener> allListeners = new CopyOnWriteArrayList<>();
+
+    private long totalSentBytes;
+    private long totalReceivedBytes;
+
+    private int totalSentMessages;
+    private int totalReceivedMessages;
+
+    private double totalSentMessagesPerSec;
+    private double totalReceivedMessagesPerSec;
+
+    private double totalSentBytesPerSec;
+    private double totalReceivedBytesPerSec;
+
+    public ConnectionStatsAccumulator(List<ConnectionStatistics> allConnectionStatistics) {
+        this.allConnectionStatistics = allConnectionStatistics;
+    }
+
+    @Override
+    public void run() {
+        long totalSentBytes = 0;
+        long totalReceivedBytes = 0;
+
+        int totalSentMessages = 0;
+        int totalReceivedMessages = 0;
+
+        for (ConnectionStatistics statistic : allConnectionStatistics) {
+            totalSentBytes += statistic.getSentBytes();
+            totalReceivedBytes += statistic.getReceivedBytes();
+
+            totalSentMessages += statistic.getTotalSentMessages();
+            totalReceivedMessages += statistic.getTotalReceivedMessages();
+        }
+
+        this.totalSentBytes = totalSentBytes;
+        this.totalReceivedBytes = totalReceivedBytes;
+
+        this.totalSentMessages = totalSentMessages;
+        this.totalReceivedMessages = totalReceivedMessages;
+
+        long passed = (System.currentTimeMillis() - startTime) / 1000;
+        totalSentMessagesPerSec = ((double) totalSentMessages / passed);
+        totalReceivedMessagesPerSec = ((double) totalReceivedMessages) / passed;
+
+        totalSentBytesPerSec = ((double) totalSentBytes) / passed;
+        totalReceivedBytesPerSec = ((double) totalReceivedBytes) / passed;
+
+        callListeners();
+    }
+
+    private void callListeners() {
+        allListeners.forEach(listener -> listener.onTotalSentStatsChanged(totalSentBytes, totalSentMessages, totalSentMessagesPerSec));
+    }
+
+    public void addListener(NetworkStatisticsService.Listener listener) {
+        allListeners.add(listener);
+    }
+
+    public void removeListener(NetworkStatisticsService.Listener listener) {
+        allListeners.remove(listener);
+    }
+}

p2p/src/main/java/bisq/network/p2p/network/statistics/NetworkStatisticsService.java

@@ -0,0 +1,98 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.network.p2p.network.statistics;
+
+import bisq.common.util.Utilities;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Getter
+public class NetworkStatisticsService {
+
+    public interface Listener {
+        void onTotalSentStatsChanged(long totalSentBytes, long totalSentMessages, double totalSentMessagesPerSec);
+    }
+
+    private final long startTime = System.currentTimeMillis();
+
+    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+    private final List<ConnectionStatistics> allConnectionStatistics = new CopyOnWriteArrayList<>();
+    private final ConnectionStatsAccumulator connectionStatsAccumulator =
+            new ConnectionStatsAccumulator(allConnectionStatistics);
+
+    private final Map<String, Integer> totalSentMessages = new HashMap<>();
+    private final Map<String, Integer> totalReceivedMessages = new HashMap<>();
+
+    public ConnectionStatistics newConnectionStatistics() {
+        var connectionStatistics = new ConnectionStatistics();
+        allConnectionStatistics.add(connectionStatistics);
+        return connectionStatistics;
+    }
+
+    public void start() {
+        scheduledExecutorService.scheduleAtFixedRate(
+                connectionStatsAccumulator, 1, 1, TimeUnit.SECONDS
+        );
+        scheduledExecutorService.scheduleAtFixedRate(
+                createStatisticsLogger(), 1, 1, TimeUnit.HOURS
+        );
+    }
+
+    public void shutdown() {
+        scheduledExecutorService.shutdownNow();
+    }
+
+    public void addListener(Listener listener) {
+        connectionStatsAccumulator.addListener(listener);
+    }
+
+    public void removeListener(Listener listener) {
+        connectionStatsAccumulator.removeListener(listener);
+    }
+
+    private Runnable createStatisticsLogger() {
+        return () -> {
+            ConnectionStatsAccumulator allStats = connectionStatsAccumulator;
+            String ls = System.lineSeparator();
+
+            log.info("Accumulated network statistics:" + ls +
+                            "Bytes sent: {};" + ls +
+                            "Number of sent messages/Sent messages: {} / {};" + ls +
+                            "Number of sent messages per sec: {};" + ls +
+                            "Bytes received: {}" + ls +
+                            "Number of received messages/Received messages: {} / {};" + ls +
+                            "Number of received messages per sec: {}" + ls,
+                    Utilities.readableFileSize(allStats.getTotalSentBytes()),
+                    allStats.getTotalSentMessages(), totalSentMessages,
+                    allStats.getTotalSentMessagesPerSec(),
+                    Utilities.readableFileSize(allStats.getTotalReceivedBytes()),
+                    allStats.getTotalReceivedMessages(), totalReceivedMessages,
+                    allStats.getTotalReceivedMessagesPerSec());
+        };
+    }
+}