From 09f7f28456e1267ab5c4b22d0181f460eeaef92f Mon Sep 17 00:00:00 2001 From: jianghuazhu <740087514@qq.com> Date: Wed, 17 Jul 2024 20:25:48 +0800 Subject: [PATCH 1/4] HDDS-11083. Avoid duplicate creation of RunningDatanodeState --- .../common/statemachine/StateContext.java | 76 +++++++++------- .../common/states/DatanodeState.java | 5 ++ .../states/datanode/InitDatanodeState.java | 5 ++ .../states/datanode/RunningDatanodeState.java | 89 +++++++------------ .../common/statemachine/TestStateContext.java | 10 +++ 5 files changed, 92 insertions(+), 93 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index 712b3f0f2326..de28bbb39de3 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -152,6 +152,8 @@ public class StateContext { private final String threadNamePrefix; + private RunningDatanodeState runningDatanodeState; + /** * Constructs a StateContext. * @@ -612,9 +614,11 @@ public DatanodeState getTask() { parentDatanodeStateMachine.getConnectionManager(), this); case RUNNING: - return new RunningDatanodeState(this.conf, - parentDatanodeStateMachine.getConnectionManager(), - this); + if (runningDatanodeState == null) { + runningDatanodeState = new RunningDatanodeState(this.conf, + parentDatanodeStateMachine.getConnectionManager(), this); + } + return runningDatanodeState; case SHUTDOWN: return null; default: @@ -655,41 +659,45 @@ public void execute(ExecutorService service, long time, TimeUnit unit) // we called stop DatanodeStateMachine, this sets state to SHUTDOWN, and // there is a chance of getting task as null. if (task != null) { - if (this.isEntering()) { - task.onEnter(); - } - - if (!isThreadPoolAvailable(service)) { - long count = threadPoolNotAvailableCount.incrementAndGet(); - long unavailableTime = Time.monotonicNow() - lastHeartbeatSent.get(); - if (unavailableTime > time && count % getLogWarnInterval(conf) == 0) { - LOG.warn("No available thread in pool for the past {} seconds " + - "and {} times.", unit.toSeconds(unavailableTime), count); + try { + if (this.isEntering()) { + task.onEnter(); } - return; - } - threadPoolNotAvailableCount.set(0); - task.execute(service); - lastHeartbeatSent.set(Time.monotonicNow()); - DatanodeStateMachine.DatanodeStates newState = task.await(time, unit); - if (this.state != newState) { - if (LOG.isDebugEnabled()) { - LOG.debug("Task {} executed, state transited from {} to {}", - task.getClass().getSimpleName(), this.state, newState); + + if (!isThreadPoolAvailable(service)) { + long count = threadPoolNotAvailableCount.incrementAndGet(); + long unavailableTime = Time.monotonicNow() - lastHeartbeatSent.get(); + if (unavailableTime > time && count % getLogWarnInterval(conf) == 0) { + LOG.warn("No available thread in pool for the past {} seconds " + + "and {} times.", unit.toSeconds(unavailableTime), count); + } + return; } - if (isExiting(newState)) { - task.onExit(); + threadPoolNotAvailableCount.set(0); + task.execute(service); + lastHeartbeatSent.set(Time.monotonicNow()); + DatanodeStateMachine.DatanodeStates newState = task.await(time, unit); + if (this.state != newState) { + if (LOG.isDebugEnabled()) { + LOG.debug("Task {} executed, state transited from {} to {}", + task.getClass().getSimpleName(), this.state, newState); + } + if (isExiting(newState)) { + task.onExit(); + } + this.setState(newState); } - this.setState(newState); - } - if (!shutdownGracefully && - this.state == DatanodeStateMachine.DatanodeStates.SHUTDOWN) { - LOG.error("Critical error occurred in StateMachine, setting " + - "shutDownMachine"); - // When some exception occurred, set shutdownStateMachine to true, so - // that we can terminate the datanode. - setShutdownOnError(); + if (!shutdownGracefully && + this.state == DatanodeStateMachine.DatanodeStates.SHUTDOWN) { + LOG.error("Critical error occurred in StateMachine, setting " + + "shutDownMachine"); + // When some exception occurred, set shutdownStateMachine to true, so + // that we can terminate the datanode. + setShutdownOnError(); + } + } finally { + task.clear(); } } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java index 25be207dcd9c..4d9c178005a7 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java @@ -55,4 +55,9 @@ public interface DatanodeState { T await(long time, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException; + /** + * Clean up some resources. + */ + void clear(); + } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java index 0b4fbfe2fa91..80c6c662b99f 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java @@ -178,4 +178,9 @@ public DatanodeStateMachine.DatanodeStates await(long time, ExecutionException, TimeoutException { return result.get(time, timeUnit); } + + @Override + public void clear() { + + } } 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 4e5b64c27e5e..c8609d4c3e0f 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 @@ -31,11 +31,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.EnumMap; -import java.util.HashMap; import java.util.LinkedList; import java.util.List; -import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.CompletionService; import java.util.concurrent.ExecutionException; @@ -54,9 +51,6 @@ public class RunningDatanodeState implements DatanodeState { private final ConfigurationSource conf; private final StateContext context; private CompletionService ecs; - /** Cache the end point task per end point per end point state. */ - private Map>> endpointTasks; public RunningDatanodeState(ConfigurationSource conf, SCMConnectionManager connectionManager, @@ -64,55 +58,6 @@ public RunningDatanodeState(ConfigurationSource conf, this.connectionManager = connectionManager; this.conf = conf; this.context = context; - initEndPointTask(); - } - - /** - * Initialize end point tasks corresponding to each end point, - * each end point state. - */ - private void initEndPointTask() { - endpointTasks = new HashMap<>(); - for (EndpointStateMachine endpoint : connectionManager.getValues()) { - EnumMap> endpointTaskForState = - new EnumMap<>(EndPointStates.class); - - for (EndPointStates state : EndPointStates.values()) { - Callable endPointTask = null; - switch (state) { - case GETVERSION: - endPointTask = new VersionEndpointTask(endpoint, conf, - context.getParent().getContainer()); - break; - case REGISTER: - endPointTask = RegisterEndpointTask.newBuilder() - .setConfig(conf) - .setEndpointStateMachine(endpoint) - .setContext(context) - .setDatanodeDetails(context.getParent().getDatanodeDetails()) - .setOzoneContainer(context.getParent().getContainer()) - .build(); - break; - case HEARTBEAT: - endPointTask = HeartbeatEndpointTask.newBuilder() - .setConfig(conf) - .setEndpointStateMachine(endpoint) - .setDatanodeDetails(context - .getParent() - .getDatanodeDetails()) - .setContext(context) - .build(); - break; - default: - break; - } - - if (endPointTask != null) { - endpointTaskForState.put(state, endPointTask); - } - } - endpointTasks.put(endpoint, endpointTaskForState); - } } /** @@ -173,11 +118,32 @@ public void setExecutorCompletionService(ExecutorCompletionService e) { private Callable getEndPointTask( EndpointStateMachine endpoint) { - if (endpointTasks.containsKey(endpoint)) { - return endpointTasks.get(endpoint).get(endpoint.getState()); - } else { - throw new IllegalArgumentException("Illegal endpoint: " + endpoint); + Callable endpointTask = null; + for (EndpointStateMachine endpointStateMachine : connectionManager.getValues()) { + if (endpointStateMachine.getAddressString().equals(endpoint.getAddressString())) { + if (endpoint.getState().getValue() == EndPointStates.GETVERSION.getValue()) { + endpointTask = new VersionEndpointTask(endpoint, conf, + context.getParent().getContainer()); + } else if (endpoint.getState().getValue() == EndPointStates.REGISTER.getValue()) { + endpointTask = RegisterEndpointTask.newBuilder() + .setConfig(conf) + .setEndpointStateMachine(endpoint) + .setContext(context) + .setDatanodeDetails(context.getParent().getDatanodeDetails()) + .setOzoneContainer(context.getParent().getContainer()) + .build(); + } else if (endpoint.getState().getValue() == EndPointStates.HEARTBEAT.getValue()) { + endpointTask = HeartbeatEndpointTask.newBuilder() + .setConfig(conf) + .setEndpointStateMachine(endpoint) + .setDatanodeDetails(context.getParent().getDatanodeDetails()) + .setContext(context) + .build(); + } + } } + + return endpointTask; } /** @@ -238,4 +204,9 @@ private Callable getEndPointTask( } return computeNextContainerState(results); } + + @Override + public void clear() { + ecs = null; + } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 3ea478cacbb8..951690c8c190 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -542,6 +542,11 @@ public DatanodeStates await(long time, TimeUnit timeUnit) 10000); return DatanodeStates.RUNNING; } + + @Override + public void clear() { + + } }; } }; @@ -625,6 +630,11 @@ public DatanodeStates await(long time, TimeUnit timeUnit) { awaited.incrementAndGet(); return DatanodeStates.INIT; } + + @Override + public void clear() { + + } }; } }; From e8729505be8ceae4b2048bda71211169e4f94356 Mon Sep 17 00:00:00 2001 From: jianghuazhu <740087514@qq.com> Date: Wed, 17 Jul 2024 22:40:17 +0800 Subject: [PATCH 2/4] Add default method --- .../ozone/container/common/states/DatanodeState.java | 3 ++- .../common/states/datanode/InitDatanodeState.java | 5 ----- .../common/statemachine/TestStateContext.java | 10 ---------- 3 files changed, 2 insertions(+), 16 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java index 4d9c178005a7..f057a852bc4e 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/DatanodeState.java @@ -58,6 +58,7 @@ T await(long time, TimeUnit timeUnit) /** * Clean up some resources. */ - void clear(); + @SuppressWarnings("checkstyle:WhitespaceAround") + default void clear(){} } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java index 80c6c662b99f..0b4fbfe2fa91 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/datanode/InitDatanodeState.java @@ -178,9 +178,4 @@ public DatanodeStateMachine.DatanodeStates await(long time, ExecutionException, TimeoutException { return result.get(time, timeUnit); } - - @Override - public void clear() { - - } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 951690c8c190..3ea478cacbb8 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -542,11 +542,6 @@ public DatanodeStates await(long time, TimeUnit timeUnit) 10000); return DatanodeStates.RUNNING; } - - @Override - public void clear() { - - } }; } }; @@ -630,11 +625,6 @@ public DatanodeStates await(long time, TimeUnit timeUnit) { awaited.incrementAndGet(); return DatanodeStates.INIT; } - - @Override - public void clear() { - - } }; } }; From a48499213a35aa883d12b3207c6bb9cd8668b29d Mon Sep 17 00:00:00 2001 From: jianghuazhu <740087514@qq.com> Date: Sat, 20 Jul 2024 23:34:47 +0800 Subject: [PATCH 3/4] Update StateContext. --- .../common/statemachine/StateContext.java | 74 ++++++++++--------- 1 file changed, 38 insertions(+), 36 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index de28bbb39de3..93a45905975a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -658,47 +658,49 @@ public void execute(ExecutorService service, long time, TimeUnit unit) // Adding not null check, in a case where datanode is still starting up, but // we called stop DatanodeStateMachine, this sets state to SHUTDOWN, and // there is a chance of getting task as null. - if (task != null) { - try { - if (this.isEntering()) { - task.onEnter(); - } + if (task == null) { + return; + } - if (!isThreadPoolAvailable(service)) { - long count = threadPoolNotAvailableCount.incrementAndGet(); - long unavailableTime = Time.monotonicNow() - lastHeartbeatSent.get(); - if (unavailableTime > time && count % getLogWarnInterval(conf) == 0) { - LOG.warn("No available thread in pool for the past {} seconds " + - "and {} times.", unit.toSeconds(unavailableTime), count); - } - return; + try { + if (this.isEntering()) { + task.onEnter(); + } + + if (!isThreadPoolAvailable(service)) { + long count = threadPoolNotAvailableCount.incrementAndGet(); + long unavailableTime = Time.monotonicNow() - lastHeartbeatSent.get(); + if (unavailableTime > time && count % getLogWarnInterval(conf) == 0) { + LOG.warn("No available thread in pool for the past {} seconds " + + "and {} times.", unit.toSeconds(unavailableTime), count); } - threadPoolNotAvailableCount.set(0); - task.execute(service); - lastHeartbeatSent.set(Time.monotonicNow()); - DatanodeStateMachine.DatanodeStates newState = task.await(time, unit); - if (this.state != newState) { - if (LOG.isDebugEnabled()) { - LOG.debug("Task {} executed, state transited from {} to {}", - task.getClass().getSimpleName(), this.state, newState); - } - if (isExiting(newState)) { - task.onExit(); - } - this.setState(newState); + return; + } + threadPoolNotAvailableCount.set(0); + task.execute(service); + lastHeartbeatSent.set(Time.monotonicNow()); + DatanodeStateMachine.DatanodeStates newState = task.await(time, unit); + if (this.state != newState) { + if (LOG.isDebugEnabled()) { + LOG.debug("Task {} executed, state transited from {} to {}", + task.getClass().getSimpleName(), this.state, newState); } - - if (!shutdownGracefully && - this.state == DatanodeStateMachine.DatanodeStates.SHUTDOWN) { - LOG.error("Critical error occurred in StateMachine, setting " + - "shutDownMachine"); - // When some exception occurred, set shutdownStateMachine to true, so - // that we can terminate the datanode. - setShutdownOnError(); + if (isExiting(newState)) { + task.onExit(); } - } finally { - task.clear(); + this.setState(newState); } + + if (!shutdownGracefully && + this.state == DatanodeStateMachine.DatanodeStates.SHUTDOWN) { + LOG.error("Critical error occurred in StateMachine, setting " + + "shutDownMachine"); + // When some exception occurred, set shutdownStateMachine to true, so + // that we can terminate the datanode. + setShutdownOnError(); + } + } finally { + task.clear(); } } From d9c541f91596d1ff711b3a821b57a1cc514bcf8f Mon Sep 17 00:00:00 2001 From: jianghuazhu <740087514@qq.com> Date: Sun, 21 Jul 2024 23:53:31 +0800 Subject: [PATCH 4/4] Update RunningDatanodeState. --- .../states/datanode/RunningDatanodeState.java | 51 +++++++++---------- 1 file changed, 24 insertions(+), 27 deletions(-) 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 c8609d4c3e0f..c4d507668d4c 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 @@ -85,7 +85,7 @@ public void onExit() { public void execute(ExecutorService executor) { ecs = new ExecutorCompletionService<>(executor); for (EndpointStateMachine endpoint : connectionManager.getValues()) { - Callable endpointTask = getEndPointTask(endpoint); + Callable endpointTask = buildEndPointTask(endpoint); if (endpointTask != null) { // Just do a timely wait. A slow EndpointStateMachine won't occupy // the thread in executor from DatanodeStateMachine for a long time, @@ -116,34 +116,31 @@ public void setExecutorCompletionService(ExecutorCompletionService e) { this.ecs = e; } - private Callable getEndPointTask( + @SuppressWarnings("checkstyle:Indentation") + private Callable buildEndPointTask( EndpointStateMachine endpoint) { - Callable endpointTask = null; - for (EndpointStateMachine endpointStateMachine : connectionManager.getValues()) { - if (endpointStateMachine.getAddressString().equals(endpoint.getAddressString())) { - if (endpoint.getState().getValue() == EndPointStates.GETVERSION.getValue()) { - endpointTask = new VersionEndpointTask(endpoint, conf, - context.getParent().getContainer()); - } else if (endpoint.getState().getValue() == EndPointStates.REGISTER.getValue()) { - endpointTask = RegisterEndpointTask.newBuilder() - .setConfig(conf) - .setEndpointStateMachine(endpoint) - .setContext(context) - .setDatanodeDetails(context.getParent().getDatanodeDetails()) - .setOzoneContainer(context.getParent().getContainer()) - .build(); - } else if (endpoint.getState().getValue() == EndPointStates.HEARTBEAT.getValue()) { - endpointTask = HeartbeatEndpointTask.newBuilder() - .setConfig(conf) - .setEndpointStateMachine(endpoint) - .setDatanodeDetails(context.getParent().getDatanodeDetails()) - .setContext(context) - .build(); - } - } + switch (endpoint.getState()) { + case GETVERSION: + return new VersionEndpointTask(endpoint, conf, + context.getParent().getContainer()); + case REGISTER: + return RegisterEndpointTask.newBuilder() + .setConfig(conf) + .setEndpointStateMachine(endpoint) + .setContext(context) + .setDatanodeDetails(context.getParent().getDatanodeDetails()) + .setOzoneContainer(context.getParent().getContainer()) + .build(); + case HEARTBEAT: + return HeartbeatEndpointTask.newBuilder() + .setConfig(conf) + .setEndpointStateMachine(endpoint) + .setDatanodeDetails(context.getParent().getDatanodeDetails()) + .setContext(context) + .build(); + default: + return null; } - - return endpointTask; } /**