Fixing MemoryEstimatorOracle

This diff contains the following fixes:
1) Once the OOC is at OFFLAODING state, it may offload data to disk indefinitely. We should call GC manually and get OOC out of OFFLOADING if anything has changed in computation/communication pattern.
2) Fixes a "if" statement which entirely disabled regression for memory estimation.
3) Interleaving of in resetting and calculating memory estimation potentially can cause data race.
4) The superstep count in memory estimator was not coherent throughout the calculation.

Author: Hassan Eslami

giraph-core/src/main/java/org/apache/giraph/ooc/policy/MemoryEstimatorOracle.java

@@ -78,6 +78,14 @@ public class MemoryEstimatorOracle implements OutOfCoreOracle {
     new FloatConfOption("giraph.garbageEstimator.manualGCPressure", 0.95f,
         "The threshold above which GC is called manually if Full GC has not " +
             "happened in a while");
+  /**
+   * If full GC is not called within this interval, in case of tight memory
+   * pressure, we should call GC manually
+   */
+  public static final LongConfOption MANUAL_GC_INTERVAL =
+      new LongConfOption("giraph.memoryEstimator.manualGCInterval", 20000,
+          "The amount of time passed from the last full GC to call GC " +
+              "manually if memory is tight (in milliseconds)");
   /** Used to detect a high memory pressure situation */
   public static final FloatConfOption GC_MINIMUM_RECLAIM_FRACTION =
     new FloatConfOption("giraph.garbageEstimator.gcReclaimFraction", 0.05f,
@@ -140,6 +148,11 @@ public class MemoryEstimatorOracle implements OutOfCoreOracle {
   private volatile State state = State.STABLE;
   /** Timestamp of the last major GC */
   private volatile long lastMajorGCTime = 0;
+  /**
+   * Lock to avoid interleaving of resetting data structures in memory estimator
+   * with modifying those data structures
+   */
+  private Lock lock = new ReentrantLock();
 
   /**
    * Different states the OOC can be in.
@@ -172,6 +185,7 @@ public MemoryEstimatorOracle(ImmutableClassesGiraphConfiguration conf,
     this.oocThreshold = OOC_THRESHOLD.get(conf);
 
     final long checkMemoryInterval = CHECK_MEMORY_INTERVAL.get(conf);
+    final long manualGCInterval = MANUAL_GC_INTERVAL.get(conf);
 
     ThreadUtils.startThread(new Runnable() {
       @Override
@@ -183,9 +197,9 @@ public void run() {
             updateRates(oldGenUsageEstimate, usage.getMax());
           } else {
             long time = System.currentTimeMillis();
-            if (time - lastMajorGCTime >= 10000) {
+            if (time - lastMajorGCTime >= manualGCInterval) {
               double used = (double) usage.getUsed() / usage.getMax();
-              if (used > manualGCMemoryPressure) {
+              if (used > manualGCMemoryPressure || state == State.OFFLOADING) {
                 if (LOG.isInfoEnabled()) {
                   LOG.info(
                     "High memory pressure with no full GC from the JVM. " +
@@ -231,8 +245,13 @@ public void run() {
   public void startIteration() {
     AbstractEdgeStore.PROGRESS_COUNTER.reset();
     oocBytesInjected.set(0);
-    memoryEstimator.clear();
-    memoryEstimator.setCurrentSuperstep(oocEngine.getSuperstep());
+    lock.lock();
+    try {
+      memoryEstimator.clear();
+      memoryEstimator.setCurrentSuperstep(oocEngine.getSuperstep());
+    } finally {
+      lock.unlock();
+    }
     oocEngine.updateRequestsCreditFraction(1);
     oocEngine.updateActiveThreadsFraction(1);
   }
@@ -331,30 +350,36 @@ public synchronized void gcCompleted(
         }
       }
 
-      // Number of edges loaded so far (if in input superstep)
-      long edgesLoaded = oocEngine.getSuperstep() >= 0 ? 0 :
-        EdgeInputSplitsCallable.getTotalEdgesLoadedMeter().count();
-      // Number of vertices loaded so far (if in input superstep)
-      long verticesLoaded = oocEngine.getSuperstep() >= 0 ? 0 :
-        VertexInputSplitsCallable.getTotalVerticesLoadedMeter().count();
-      // Number of vertices computed (if either in compute or store phase)
-      long verticesComputed = WorkerProgress.get().getVerticesComputed() +
-        WorkerProgress.get().getVerticesStored() +
-        AbstractEdgeStore.PROGRESS_COUNTER.getProgress();
-      // Number of bytes received
-      long receivedBytes =
-        oocEngine.getNetworkMetrics().getBytesReceivedPerSuperstep();
-      // Number of OOC bytes
-      long oocBytes = oocBytesInjected.get();
-
-      memoryEstimator.addRecord(getOldGenUsed().getUsed(), edgesLoaded,
-        verticesLoaded, verticesComputed, receivedBytes, oocBytes);
+      lock.lock();
+      try {
+        // Number of edges loaded so far (if in input superstep)
+        long edgesLoaded = memoryEstimator.getCurrentSuperstep() >= 0 ? 0 :
+            EdgeInputSplitsCallable.getTotalEdgesLoadedMeter().count();
+        // Number of vertices loaded so far (if in input superstep)
+        long verticesLoaded = memoryEstimator.getCurrentSuperstep() >= 0 ? 0 :
+            VertexInputSplitsCallable.getTotalVerticesLoadedMeter().count();
+        // Number of vertices computed (if either in compute or store phase)
+        long verticesComputed = WorkerProgress.get().getVerticesComputed() +
+            WorkerProgress.get().getVerticesStored() +
+            AbstractEdgeStore.PROGRESS_COUNTER.getProgress();
+        // Number of bytes received
+        long receivedBytes =
+            oocEngine.getNetworkMetrics().getBytesReceivedPerSuperstep();
+        // Number of OOC bytes
+        long oocBytes = oocBytesInjected.get();
+
+        memoryEstimator.addRecord(getOldGenUsed().getUsed(), edgesLoaded,
+            verticesLoaded, verticesComputed, receivedBytes, oocBytes);
+      } finally {
+        lock.unlock();
+      }
 
       long garbage = before.getUsed() - after.getUsed();
       long maxMem = after.getMax();
       long memUsed = after.getUsed();
-      boolean isTight = (maxMem - memUsed) < 2 * gcReclaimFraction * maxMem &&
-        garbage < gcReclaimFraction * maxMem;
+      boolean highMemoryUsage =
+          (maxMem - memUsed) < 2 * gcReclaimFraction * maxMem;
+      boolean isTight = highMemoryUsage && garbage < gcReclaimFraction * maxMem;
       boolean predictionExist = memoryEstimator.getUsageEstimate() > 0;
       if (isTight && !predictionExist) {
         if (LOG.isInfoEnabled()) {
@@ -369,6 +394,8 @@ public synchronized void gcCompleted(
         }
         state = State.OFFLOADING;
         updateRates(1, 1);
+      } else if (state == State.OFFLOADING && !highMemoryUsage) {
+        state = State.STABLE;
       }
     }
   }
@@ -483,6 +510,10 @@ public void setCurrentSuperstep(long superstep) {
       this.currentSuperstep = superstep;
     }
 
+    public long getCurrentSuperstep() {
+      return this.currentSuperstep;
+    }
+
     /**
      * Given the current state of computation (i.e. current edges loaded,
      * vertices computed etc) and the current model (i.e. the regression
@@ -787,14 +818,6 @@ private Boolean refineCoefficient(int coefIndex, double lowerBound,
      */
     private static boolean calculateRegression(double[] coefficient,
       List<Integer> validColumnIndices, OLSMultipleLinearRegression mlr) {
-
-      if (coefficient.length != validColumnIndices.size()) {
-        LOG.warn("There are " + coefficient.length +
-          " coefficients, but " + validColumnIndices.size() +
-          " valid columns in the regression");
-        return false;
-      }
-
       double[] beta = mlr.estimateRegressionParameters();
       Arrays.fill(coefficient, 0);
       for (int i = 0; i < validColumnIndices.size(); ++i) {