[hadoop-ozone] branch master updated: HDDS-4386: Each EndpointStateMachine uses its own thread pool to talk with SCM/Recon (#1518)

[av...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

avijayan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git


The following commit(s) were added to refs/heads/master by this push:
     new 5e5635f  HDDS-4386: Each EndpointStateMachine uses its own thread pool to talk with SCM/Recon (#1518)
5e5635f is described below

commit 5e5635f2f36e9653ed656a64edc79873426bab8c
Author: GlenGeng <gl...@tencent.com>
AuthorDate: Tue Oct 27 21:10:50 2020 +0800

    HDDS-4386: Each EndpointStateMachine uses its own thread pool to talk with SCM/Recon (#1518)
---
 .../common/statemachine/EndpointStateMachine.java        | 16 ++++++++++++++++
 .../common/states/datanode/RunningDatanodeState.java     |  8 +++++++-
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
index cd1a376..13f953f 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
@@ -16,6 +16,7 @@
  */
 package org.apache.hadoop.ozone.container.common.statemachine;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.ozone.protocol.VersionResponse;
 import org.apache.hadoop.ozone.protocolPB
@@ -27,6 +28,8 @@ import java.io.Closeable;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.time.ZonedDateTime;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
@@ -51,6 +54,7 @@ public class EndpointStateMachine
   private VersionResponse version;
   private ZonedDateTime lastSuccessfulHeartbeat;
   private boolean isPassive;
+  private final ExecutorService executorService;
 
   /**
    * Constructs RPC Endpoints.
@@ -66,6 +70,11 @@ public class EndpointStateMachine
     state = EndPointStates.getInitState();
     lock = new ReentrantLock();
     this.conf = conf;
+    executorService = Executors.newSingleThreadExecutor(
+        new ThreadFactoryBuilder()
+            .setNameFormat("EndpointStateMachine task thread for "
+                + this.address + " - %d ")
+            .build());
   }
 
   /**
@@ -130,6 +139,13 @@ public class EndpointStateMachine
   }
 
   /**
+   * Returns the endpoint specific ExecutorService.
+   */
+  public ExecutorService getExecutorService() {
+    return executorService;
+  }
+
+  /**
    * Closes the connection.
    *
    * @throws IOException
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/RunningDatanodeState.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/RunningDatanodeState.java
index b0cfb4c..7366650 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/RunningDatanodeState.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/RunningDatanodeState.java
@@ -140,7 +140,13 @@ public class RunningDatanodeState implements DatanodeState {
     for (EndpointStateMachine endpoint : connectionManager.getValues()) {
       Callable<EndPointStates> endpointTask = getEndPointTask(endpoint);
       if (endpointTask != null) {
-        ecs.submit(endpointTask);
+        // Just do a timely wait. A slow EndpointStateMachine won't occupy
+        // the thread in executor from DatanodeStateMachine for a long time,
+        // so that it won't affect the communication between datanode and
+        // other EndpointStateMachine.
+        ecs.submit(() -> endpoint.getExecutorService()
+            .submit(endpointTask)
+            .get(context.getHeartbeatFrequency(), TimeUnit.MILLISECONDS));
       } else {
         // This can happen if a task is taking more time than the timeOut
         // specified for the task in await, and when it is completed the task


---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org