[PageRank] Make one-to-many messaging configurable

giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/pagerank/PageRankIteration.java

@@ -80,6 +80,9 @@ public class PageRankIteration<I extends WritableComparable,
   /** Convergence threshold */
   private final float convergenceThreshold;
 
+  /** Whether one-to-many messaging is enabled */
+  private final boolean isOneToManyMessagingEnabled;
+
   /** Sums the errors for each vertex */
   private ReducerHandle<DoubleWritable, DoubleWritable> superstepErrorSum;
   /** Maximum of the errors for each vertex */
@@ -131,6 +134,8 @@ public PageRankIteration(
     dampingFactor = PageRankSettings.getDampingFactor(conf);
     convergenceType = PageRankSettings.getConvergenceType(conf);
     convergenceThreshold = PageRankSettings.getConvergenceThreshold(conf);
+    isOneToManyMessagingEnabled =
+      PageRankSettings.isOneToManyMessagingEnabled(conf);
   }
 
 
@@ -287,6 +292,8 @@ protected MessageCombiner<? super I, DoubleWritable> getMessageCombiner(
 
   @Override
   protected boolean allowOneMessageToManyIdsEncoding() {
-    return edgeValueGetter.allVertexEdgesTheSame();
+    return edgeValueGetter.allVertexEdgesTheSame() &&
+      isOneToManyMessagingEnabled;
   }
+
 }

giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/pagerank/PageRankSettings.java

@@ -55,6 +55,19 @@ public class PageRankSettings {
       new BooleanConfOption("giraph.pagerank.weighted", true,
           "Whether to run weighted or unweighted pagerank");
 
+  /**
+   * One-to-many messaging is used as an optimization to reduce the size of the
+   * messages.
+   *
+   * However, because of the way the {@link SendWorkerOneMessageToManyRequest}
+   * is implemented it may result in large {@link ByteArrayVertexIdMessages}
+   * instances getting created. Disabling this option avoids this at the cost
+   * of sending larger messages.
+   */
+  private static final BooleanConfOption ONE_TO_MANY_MESSAGING =
+    new BooleanConfOption("giraph.pagerank.msg.onetomany", true,
+      "Enable one-to-many message optimization");
+
   /** Don't construct */
   protected PageRankSettings() { }
 
@@ -117,4 +130,13 @@ public static float getConvergenceThreshold(Configuration conf) {
   public static boolean isWeighted(Configuration conf) {
     return WEIGHTED_PAGERANK.get(conf);
   }
+
+  /**
+   * Checks whether one-to-many messaging is enabled in the configuration.
+   * @param conf Configuration
+   * @return Whether one-to-many messaging is enabled.
+   */
+  public static boolean isOneToManyMessagingEnabled(Configuration conf) {
+    return ONE_TO_MANY_MESSAGING.get(conf);
+  }
 }