DRILL-8132: Improvement of RPC (#2464)

Co-authored-by: feiteng.wtf <feiteng.wtf@cainiao.com>

Author: Leon-WTF

exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlClient.java

@@ -50,7 +50,7 @@ public ControlClient(ControlConnectionConfig config, DrillbitEndpoint remoteEndp
     super(ControlRpcConfig.getMapping(config.getBootstrapContext().getConfig(),
         config.getBootstrapContext().getExecutor()),
         config.getAllocator().getAsByteBufAllocator(),
-        config.getBootstrapContext().getBitLoopGroup(),
+        config.getBootstrapContext().getControlLoopGroup(),
         RpcType.HANDSHAKE,
         BitControlHandshake.class,
         BitControlHandshake.PARSER);

exec/java-exec/src/main/java/org/apache/drill/exec/rpc/control/ControlServer.java

@@ -41,7 +41,7 @@ public ControlServer(ControlConnectionConfig config, ConnectionManagerRegistry c
     super(ControlRpcConfig.getMapping(config.getBootstrapContext().getConfig(),
         config.getBootstrapContext().getExecutor()),
         config.getAllocator().getAsByteBufAllocator(),
-        config.getBootstrapContext().getBitLoopGroup());
+        config.getBootstrapContext().getControlLoopGroup());
     this.config = config;
     this.connectionRegistry = connectionRegistry;
   }

exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataClient.java

@@ -51,7 +51,7 @@ public DataClient(DrillbitEndpoint remoteEndpoint, DataConnectionConfig config,
         DataRpcConfig.getMapping(config.getBootstrapContext().getConfig(),
             config.getBootstrapContext().getExecutor()),
         config.getAllocator().getAsByteBufAllocator(),
-        config.getBootstrapContext().getBitClientLoopGroup(),
+        config.getBootstrapContext().getDataClientLoopGroup(),
         RpcType.HANDSHAKE,
         BitServerHandshake.class,
         BitServerHandshake.PARSER);

exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataServer.java

@@ -41,7 +41,7 @@ public DataServer(DataConnectionConfig config) {
         DataRpcConfig.getMapping(config.getBootstrapContext().getConfig(),
             config.getBootstrapContext().getExecutor()),
         config.getAllocator().getAsByteBufAllocator(),
-        config.getBootstrapContext().getBitLoopGroup());
+        config.getBootstrapContext().getDataServerLoopGroup());
     this.config = config;
   }
 

exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java

@@ -63,8 +63,9 @@ public class BootStrapContext implements AutoCloseable {
   private final DrillConfig config;
   private final CaseInsensitiveMap<OptionDefinition> definitions;
   private final AuthenticatorProvider authProvider;
-  private final EventLoopGroup loop;
-  private final EventLoopGroup loop2;
+  private final EventLoopGroup controlLoopGroup;
+  private final EventLoopGroup dataClientLoopGroup;
+  private final EventLoopGroup dataServerLoopGroup;
   private final MetricRegistry metrics;
   private final BufferAllocator allocator;
   private final ScanResult classpathScan;
@@ -81,8 +82,9 @@ public BootStrapContext(DrillConfig config, CaseInsensitiveMap<OptionDefinition>
     this.hostName = getCanonicalHostName();
     login(config);
     this.authProvider = new AuthenticatorProviderImpl(config, classpathScan);
-    this.loop = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "BitServer-");
-    this.loop2 = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "BitClient-");
+    this.controlLoopGroup = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "Control-");
+    this.dataClientLoopGroup = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "DataClient-");
+    this.dataServerLoopGroup = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "DataServer-");
     // Note that metrics are stored in a static instance
     this.metrics = DrillMetrics.getRegistry();
     this.allocator = RootAllocatorFactory.newRoot(config);
@@ -195,12 +197,16 @@ public CaseInsensitiveMap<OptionDefinition> getDefinitions() {
     return definitions;
   }
 
-  public EventLoopGroup getBitLoopGroup() {
-    return loop;
+  public EventLoopGroup getControlLoopGroup() {
+    return controlLoopGroup;
   }
 
-  public EventLoopGroup getBitClientLoopGroup() {
-    return loop2;
+  public EventLoopGroup getDataClientLoopGroup() {
+    return dataClientLoopGroup;
+  }
+
+  public EventLoopGroup getDataServerLoopGroup() {
+    return dataClientLoopGroup;
   }
 
   public MetricRegistry getMetrics() {
@@ -258,9 +264,9 @@ public void close() {
 
     try {
       AutoCloseables.close(allocator, authProvider);
-      shutdown(loop);
-      shutdown(loop2);
-
+      shutdown(controlLoopGroup);
+      shutdown(dataClientLoopGroup);
+      shutdown(dataServerLoopGroup);
     } catch (final Exception e) {
       logger.error("Error while closing", e);
     }

exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java

@@ -227,7 +227,7 @@ public AliasRegistryProvider getAliasRegistryProvider() {
   public OAuthTokenProvider getoAuthTokenProvider() { return oAuthTokenProvider; }
 
   public EventLoopGroup getBitLoopGroup() {
-    return context.getBitLoopGroup();
+    return context.getControlLoopGroup();
   }
 
   public DataConnectionCreator getDataConnectionsPool() {

exec/java-exec/src/main/resources/drill-module.conf

@@ -91,7 +91,7 @@ drill.exec: {
           count: 7200,
           delay: 500
         },
-        threads: 10
+        threads: 8
         memory: {
           control: {
             reservation: 0,