[onionoo/master] Speed up updating weights files by using 4 updater threads. - tor-commits

27 Jul 2012

commit 752cff0f5d0af833235d48a916f5e9be71a03b0b
Author: Karsten Loesing karsten.loesing@gmx.net
Date:   Thu Jul 26 13:47:19 2012 +0200
Speed up updating weights files by using 4 updater threads.
Using 4 updater threads reduces the time to process 1 consensus from 20 to
    11 seconds here.
---
 src/org/torproject/onionoo/WeightsDataWriter.java |  101 ++++++++++++++++-----
 1 files changed, 78 insertions(+), 23 deletions(-)

diff --git a/src/org/torproject/onionoo/WeightsDataWriter.java b/src/org/torproject/onionoo/WeightsDataWriter.java
index d3a37fc..c4a887b 100644
--- a/src/org/torproject/onionoo/WeightsDataWriter.java
+++ b/src/org/torproject/onionoo/WeightsDataWriter.java
@@ -13,6 +13,7 @@ import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -34,13 +35,6 @@ import org.torproject.descriptor.ServerDescriptor;
public class WeightsDataWriter {
-  private SimpleDateFormat dateTimeFormat = new SimpleDateFormat(
-      "yyyy-MM-dd HH:mm:ss");
-  public WeightsDataWriter() {
-    this.dateTimeFormat.setLenient(false);
-    this.dateTimeFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
-  }
-
   private SortedSet<String> currentFingerprints = new TreeSet<String>();
   public void setCurrentRelays(SortedMap<String, Node> currentRelays) {
     this.currentFingerprints.addAll(currentRelays.keySet());
@@ -97,19 +91,72 @@ public class WeightsDataWriter {
                 freshUntilMillis = consensus.getFreshUntilMillis();
             SortedMap<String, double[]> pathSelectionWeights =
                 this.calculatePathSelectionProbabilities(consensus);
-            for (Map.Entry<String, double[]> e :
-                pathSelectionWeights.entrySet()) {
-              String fingerprint = e.getKey();
-              double[] weights = e.getValue();
-              this.addToHistory(fingerprint, validAfterMillis,
-                  freshUntilMillis, weights);
-            }
+            this.updateWeightsHistory(validAfterMillis, freshUntilMillis,
+                pathSelectionWeights);
           }
         }
       }
     }
   }
+  private static final int HISTORY_UPDATER_WORKERS_NUM = 4;
+  private void updateWeightsHistory(long validAfterMillis,
+      long freshUntilMillis,
+      SortedMap<String, double[]> pathSelectionWeights) {
+    int files = pathSelectionWeights.size();
+    List<HistoryUpdateWorker> historyUpdateWorkers =
+        new ArrayList<HistoryUpdateWorker>();
+    for (int i = 0; i < HISTORY_UPDATER_WORKERS_NUM; i++) {
+      HistoryUpdateWorker historyUpdateWorker =
+          new HistoryUpdateWorker(validAfterMillis, freshUntilMillis,
+          pathSelectionWeights, this);
+      historyUpdateWorkers.add(historyUpdateWorker);
+      historyUpdateWorker.setDaemon(true);
+      historyUpdateWorker.start();
+    }
+    for (HistoryUpdateWorker historyUpdateWorker : historyUpdateWorkers) {
+      try {
+        historyUpdateWorker.join();
+      } catch (InterruptedException e) {
+        /* This is not something that we can take care of.  Just leave the
+         * worker thread alone. */
+      }
+    }
+  }
+
+  private class HistoryUpdateWorker extends Thread {
+    private long validAfterMillis;
+    private long freshUntilMillis;
+    private SortedMap<String, double[]> pathSelectionWeights;
+    private WeightsDataWriter parent;
+    public HistoryUpdateWorker(long validAfterMillis,
+        long freshUntilMillis,
+        SortedMap<String, double[]> pathSelectionWeights,
+        WeightsDataWriter parent) {
+      this.validAfterMillis = validAfterMillis;
+      this.freshUntilMillis = freshUntilMillis;
+      this.pathSelectionWeights = pathSelectionWeights;
+      this.parent = parent;
+    }
+    public void run() {
+      String fingerprint = null;
+      double[] weights = null;
+      do {
+        fingerprint = null;
+        synchronized (pathSelectionWeights) {
+          if (!pathSelectionWeights.isEmpty()) {
+            fingerprint = pathSelectionWeights.firstKey();
+            weights = pathSelectionWeights.remove(fingerprint);
+          }
+        }
+        if (fingerprint != null) {
+          this.parent.addToHistory(fingerprint, this.validAfterMillis,
+              this.freshUntilMillis, weights);
+        }
+      } while (fingerprint != null);
+    }
+  }
+
   private SortedMap<String, double[]> calculatePathSelectionProbabilities(
       RelayNetworkStatusConsensus consensus) {
     double wgg = 1.0, wgd = 1.0, wmg = 1.0, wmm = 1.0, wme = 1.0,
@@ -233,6 +280,10 @@ public class WeightsDataWriter {
     });
     File historyFile = new File("status/weights", fingerprint);
     if (historyFile.exists()) {
+      SimpleDateFormat dateTimeFormat = new SimpleDateFormat(
+          "yyyy-MM-dd HH:mm:ss");
+      dateTimeFormat.setLenient(false);
+      dateTimeFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
       try {
         BufferedReader br = new BufferedReader(new FileReader(
             historyFile));
@@ -245,9 +296,9 @@ public class WeightsDataWriter {
                 + "'.  Skipping this line.");
             continue;
           }
-          long validAfterMillis = this.dateTimeFormat.parse(parts[0]
+          long validAfterMillis = dateTimeFormat.parse(parts[0]
               + " " + parts[1]).getTime();
-          long freshUntilMillis = this.dateTimeFormat.parse(parts[2]
+          long freshUntilMillis = dateTimeFormat.parse(parts[2]
               + " " + parts[3]).getTime();
           long[] interval = new long[] { validAfterMillis,
               freshUntilMillis };
@@ -330,13 +381,16 @@ public class WeightsDataWriter {
       SortedMap<long[], double[]> history) {
     File historyFile = new File("status/weights", fingerprint);
     try {
+      SimpleDateFormat dateTimeFormat = new SimpleDateFormat(
+          "yyyy-MM-dd HH:mm:ss");
+      dateTimeFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
       historyFile.getParentFile().mkdirs();
       BufferedWriter bw = new BufferedWriter(new FileWriter(historyFile));
       for (Map.Entry<long[], double[]> e : history.entrySet()) {
         long[] fresh = e.getKey();
         double[] weights = e.getValue();
-        bw.write(this.dateTimeFormat.format(fresh[0]) + " "
-            + this.dateTimeFormat.format(fresh[1]));
+        bw.write(dateTimeFormat.format(fresh[0]) + " "
+            + dateTimeFormat.format(fresh[1]));
         for (double weight : weights) {
           bw.write(String.format(" %.12f", weight));
         }
@@ -492,12 +546,13 @@ public class WeightsDataWriter {
     double factor = ((double) maxValue) / 999.0;
     int count = lastNonNullIndex - firstNonNullIndex + 1;
     StringBuilder sb = new StringBuilder();
+    SimpleDateFormat dateTimeFormat = new SimpleDateFormat(
+        "yyyy-MM-dd HH:mm:ss");
+    dateTimeFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
     sb.append(""" + graphName + "":{"
-        + ""first":""
-        + this.dateTimeFormat.format(firstDataPointMillis) + "","
-        + ""last":""
-        + this.dateTimeFormat.format(lastDataPointMillis) + "","
-        + ""interval":" + String.valueOf(dataPointInterval / 1000L)
+        + ""first":"" + dateTimeFormat.format(firstDataPointMillis)
+        + "","last":"" + dateTimeFormat.format(lastDataPointMillis)
+        + "","interval":" + String.valueOf(dataPointInterval / 1000L)
         + ","factor":" + String.format(Locale.US, "%.9f", factor)
         + ","count":" + String.valueOf(count) + ","values":[");
     int dataPointsWritten = 0, previousNonNullIndex = -2;

    