Decoupling NettyClient from control flow policy

Summary: This diff refactors NettyClient by decoupling flow control mechanism from NettyClient. Through the refactoring process, some performance and correctness bugs have been found due to the better readability of the refactored code. Test Plan: mvn clean verify Tested large jobs and the output was correct Tested large jobs and it did not have any performance degradation for codes using the old mechanism Reviewers: maja.kabiljo, sergey.edunov, avery.ching, dionysis.logothetis Reviewed By: dionysis.logothetis Differential Revision: https://reviews.facebook.net/D56367
apache · Apr 26, 2016 · b90b59d · b90b59d
1 parent b5284cd
commit b90b59d
Show file tree

Hide file tree

Showing 30 changed files with 921 additions and 679 deletions.
diff --git a/findbugs-exclude.xml b/findbugs-exclude.xml
@@ -69,11 +69,7 @@
     <Bug pattern="JLM_JSR166_UTILCONCURRENT_MONITORENTER"/>
   </Match>
   <Match>
-    <Class name="org.apache.giraph.comm.netty.handler.RequestServerHandler"/>
-    <Bug pattern="JLM_JSR166_UTILCONCURRENT_MONITORENTER"/>
-  </Match>
-  <Match>
-    <Class name="org.apache.giraph.comm.netty.handler.ResponseClientHandler"/>
+    <Class name="org.apache.giraph.comm.flow_control.CreditBasedFlowControl"/>
     <Bug pattern="JLM_JSR166_UTILCONCURRENT_MONITORENTER"/>
   </Match>
   <Match>

diff --git a/giraph-core/src/main/java/org/apache/giraph/bsp/CentralizedService.java b/giraph-core/src/main/java/org/apache/giraph/bsp/CentralizedService.java
@@ -21,6 +21,7 @@
 import java.util.List;
 
 import org.apache.giraph.job.JobProgressTracker;
+import org.apache.giraph.master.MasterInfo;
 import org.apache.giraph.worker.WorkerInfo;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
@@ -59,6 +60,13 @@ public interface CentralizedService<I extends WritableComparable,
    */
   List<WorkerInfo> getWorkerInfoList();
 
+  /**
+   * Get master info
+   *
+   * @return Master info
+   */
+  MasterInfo getMasterInfo();
+
   /**
    * Get JobProgressTracker to report progress to
    *

diff --git a/giraph-core/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java b/giraph-core/src/main/java/org/apache/giraph/bsp/CentralizedServiceMaster.java
@@ -24,7 +24,6 @@
 import org.apache.giraph.master.AggregatorToGlobalCommTranslation;
 import org.apache.giraph.master.MasterCompute;
 import org.apache.giraph.master.MasterGlobalCommHandler;
-import org.apache.giraph.master.MasterInfo;
 import org.apache.giraph.worker.WorkerInfo;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
@@ -53,13 +52,6 @@ public interface CentralizedServiceMaster<I extends WritableComparable,
    */
   boolean becomeMaster();
 
-  /**
-   * Get master information
-   *
-   * @return Master information
-   */
-  MasterInfo getMasterInfo();
-
   /**
    * Check all the {@link org.apache.giraph.worker.WorkerInfo} objects to ensure
    * that a minimum number of good workers exists out of the total that have

diff --git a/giraph-core/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java b/giraph-core/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java
@@ -25,7 +25,6 @@
 import org.apache.giraph.graph.GraphTaskManager;
 import org.apache.giraph.graph.VertexEdgeCount;
 import org.apache.giraph.io.superstep_output.SuperstepOutput;
-import org.apache.giraph.master.MasterInfo;
 import org.apache.giraph.metrics.GiraphTimerContext;
 import org.apache.giraph.partition.PartitionOwner;
 import org.apache.giraph.partition.PartitionStats;
@@ -182,13 +181,6 @@ FinishedSuperstepStats finishSuperstep(
   void exchangeVertexPartitions(
       Collection<? extends PartitionOwner> masterSetPartitionOwners);
 
-  /**
-   * Get master info
-   *
-   * @return Master info
-   */
-  MasterInfo getMasterInfo();
-
   /**
    * Get the GraphTaskManager that this service is using.  Vertices need to know
    * this.
@@ -247,13 +239,6 @@ void cleanup(FinishedSuperstepStats finishedSuperstepStats)
    */
   GlobalStats getGlobalStats();
 
-  /**
-   * Get the number of partitions owned by this worker
-   *
-   * @return number of partitions owned
-   */
-  int getNumPartitionsOwned();
-
   /**
    * Get input splits handler used during input
    *

diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/MasterClient.java b/giraph-core/src/main/java/org/apache/giraph/comm/MasterClient.java
@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 
+import org.apache.giraph.comm.flow_control.FlowControl;
 import org.apache.giraph.comm.requests.WritableRequest;
 import org.apache.hadoop.io.Writable;
 
@@ -66,5 +67,12 @@ void sendToOwner(String name, GlobalCommType type, Writable value)
    * Closes all connections.
    */
   void closeConnections();
+
+  /**
+   * Get the reference to the flow control policy used for sending requests
+   *
+   * @return reference to the flow control policy
+   */
+  FlowControl getFlowControl();
 }
 
diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/MasterServer.java b/giraph-core/src/main/java/org/apache/giraph/comm/MasterServer.java
@@ -18,6 +18,8 @@
 
 package org.apache.giraph.comm;
 
+import org.apache.giraph.comm.flow_control.FlowControl;
+
 import java.net.InetSocketAddress;
 
 /**
@@ -42,4 +44,11 @@ public interface MasterServer {
    * Shuts down.
    */
   void close();
+
+  /**
+   * Inform the server about the flow control policy used in sending requests
+   *
+   * @param flowControl reference to flow control policy
+   */
+  void setFlowControl(FlowControl flowControl);
 }
diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/WorkerClient.java b/giraph-core/src/main/java/org/apache/giraph/comm/WorkerClient.java
@@ -18,6 +18,7 @@
 
 package org.apache.giraph.comm;
 
+import org.apache.giraph.comm.flow_control.FlowControl;
 import org.apache.giraph.comm.requests.WritableRequest;
 
 import org.apache.giraph.partition.PartitionOwner;
@@ -96,4 +97,9 @@ public interface WorkerClient<I extends WritableComparable,
    */
   void authenticate() throws IOException;
 /*end[HADOOP_NON_SECURE]*/
+
+  /**
+   * @return the flow control used in sending requests
+   */
+  FlowControl getFlowControl();
 }
diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/WorkerServer.java b/giraph-core/src/main/java/org/apache/giraph/comm/WorkerServer.java
@@ -18,6 +18,7 @@
 
 package org.apache.giraph.comm;
 
+import org.apache.giraph.comm.flow_control.FlowControl;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 
@@ -64,4 +65,11 @@ public interface WorkerServer<I extends WritableComparable,
    * Shuts down.
    */
   void close();
+
+  /**
+   * Inform this server about the flow control used in sending requests
+   *
+   * @param flowControl reference to the flow control policy
+   */
+  void setFlowControl(FlowControl flowControl);
 }