From 3008272e309a947b15b88b9c8b8e6e7a4867dd98 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Mon, 7 Oct 2024 11:46:53 +0530 Subject: [PATCH 1/6] HDDS-11386. Multithreading bug in ContainerBalancerTask --- .../scm/container/balancer/ContainerBalancerTask.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java index 19a2f3c2e621..26bf5a034c59 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java @@ -55,6 +55,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -117,7 +118,7 @@ public class ContainerBalancerTask implements Runnable { private IterationResult iterationResult; private int nextIterationIndex; private boolean delayStart; - private List iterationsStatistic; + private CopyOnWriteArrayList iterationsStatistic; /** * Constructs ContainerBalancerTask with the specified arguments. @@ -166,7 +167,7 @@ public ContainerBalancerTask(StorageContainerManager scm, findTargetStrategy = new FindTargetGreedyByUsageInfo(containerManager, placementPolicyValidateProxy, nodeManager); } - this.iterationsStatistic = new ArrayList<>(); + this.iterationsStatistic = new CopyOnWriteArrayList<>(); } /** @@ -342,7 +343,7 @@ private void saveIterationStatistic(Integer iterationNumber, IterationResult iR) iterationsStatistic.add(iterationStatistic); } - public List getCurrentIterationsStatistic() { + public CopyOnWriteArrayList getCurrentIterationsStatistic() { int lastIterationNumber = iterationsStatistic.stream() .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) @@ -380,7 +381,7 @@ public List getCurrentIterationsStatis ) ) ); - List resultList = new ArrayList<>(iterationsStatistic); + CopyOnWriteArrayList resultList = new CopyOnWriteArrayList<>(iterationsStatistic); resultList.add(currentIterationStatistic); return resultList; } From 523e391a4dce81b51c67535c3f7a717ca4defcbf Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Mon, 7 Oct 2024 11:55:22 +0530 Subject: [PATCH 2/6] Fixed checkstyle error --- .../hdds/scm/container/balancer/ContainerBalancerTask.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java index 26bf5a034c59..52c5cd591a2f 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java @@ -381,7 +381,8 @@ public CopyOnWriteArrayList getCurrent ) ) ); - CopyOnWriteArrayList resultList = new CopyOnWriteArrayList<>(iterationsStatistic); + CopyOnWriteArrayList resultList = + new CopyOnWriteArrayList<>(iterationsStatistic); resultList.add(currentIterationStatistic); return resultList; } From c88f8ea491cf51a2c4e4aea82d4a1ff0f0269bb9 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Mon, 21 Oct 2024 13:47:54 +0530 Subject: [PATCH 3/6] HDDS-11386. Multithreading bug in ContainerBalancerTask --- .../balancer/ContainerBalancerTask.java | 90 +++++++++---------- 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java index 52c5cd591a2f..6d3f303a605c 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java @@ -55,7 +55,6 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -118,7 +117,7 @@ public class ContainerBalancerTask implements Runnable { private IterationResult iterationResult; private int nextIterationIndex; private boolean delayStart; - private CopyOnWriteArrayList iterationsStatistic; + private final List iterationsStatistic; /** * Constructs ContainerBalancerTask with the specified arguments. @@ -167,7 +166,7 @@ public ContainerBalancerTask(StorageContainerManager scm, findTargetStrategy = new FindTargetGreedyByUsageInfo(containerManager, placementPolicyValidateProxy, nodeManager); } - this.iterationsStatistic = new CopyOnWriteArrayList<>(); + this.iterationsStatistic = new ArrayList<>(); } /** @@ -343,48 +342,49 @@ private void saveIterationStatistic(Integer iterationNumber, IterationResult iR) iterationsStatistic.add(iterationStatistic); } - public CopyOnWriteArrayList getCurrentIterationsStatistic() { - - int lastIterationNumber = iterationsStatistic.stream() - .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) - .max() - .orElse(0); - - ContainerBalancerTaskIterationStatusInfo currentIterationStatistic = new ContainerBalancerTaskIterationStatusInfo( - lastIterationNumber + 1, - null, - getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, - sizeActuallyMovedInLatestIteration / OzoneConsts.GB, - metrics.getNumContainerMovesScheduledInLatestIteration(), - metrics.getNumContainerMovesCompletedInLatestIteration(), - metrics.getNumContainerMovesFailedInLatestIteration(), - metrics.getNumContainerMovesTimeoutInLatestIteration(), - findTargetStrategy.getSizeEnteringNodes() - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect(Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ), - findSourceStrategy.getSizeLeavingNodes() - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect( - Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ) - ); - CopyOnWriteArrayList resultList = - new CopyOnWriteArrayList<>(iterationsStatistic); - resultList.add(currentIterationStatistic); - return resultList; + public List getCurrentIterationsStatistic() { + + synchronized (iterationsStatistic) { + int lastIterationNumber = iterationsStatistic.stream() + .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) + .max() + .orElse(0); + + ContainerBalancerTaskIterationStatusInfo currentIterationStatistic = new ContainerBalancerTaskIterationStatusInfo( + lastIterationNumber + 1, + null, + getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, + sizeActuallyMovedInLatestIteration / OzoneConsts.GB, + metrics.getNumContainerMovesScheduledInLatestIteration(), + metrics.getNumContainerMovesCompletedInLatestIteration(), + metrics.getNumContainerMovesFailedInLatestIteration(), + metrics.getNumContainerMovesTimeoutInLatestIteration(), + new ConcurrentHashMap<>(findTargetStrategy.getSizeEnteringNodes()) + .entrySet() + .stream() + .filter(Objects::nonNull) + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect(Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) + ), + new ConcurrentHashMap<>(findSourceStrategy.getSizeLeavingNodes()) + .entrySet() + .stream() + .filter(Objects::nonNull) + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) + ) + ); + List resultList = new ArrayList<>(iterationsStatistic); + resultList.add(currentIterationStatistic); + return resultList; + } } /** From 36b7112b55c9c6bcdf266c176bc819f28a385ca8 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Fri, 15 Nov 2024 13:01:58 +0530 Subject: [PATCH 4/6] Added locks and ConcurrentLinkedQueue to iterationsStatistic --- .../balancer/ContainerBalancerTask.java | 84 +++++++++++-------- 1 file changed, 47 insertions(+), 37 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java index 6d3f303a605c..613a1529c4af 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java @@ -55,9 +55,11 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL; @@ -117,7 +119,8 @@ public class ContainerBalancerTask implements Runnable { private IterationResult iterationResult; private int nextIterationIndex; private boolean delayStart; - private final List iterationsStatistic; + private final ConcurrentLinkedQueue iterationsStatistic; + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); /** * Constructs ContainerBalancerTask with the specified arguments. @@ -166,7 +169,7 @@ public ContainerBalancerTask(StorageContainerManager scm, findTargetStrategy = new FindTargetGreedyByUsageInfo(containerManager, placementPolicyValidateProxy, nodeManager); } - this.iterationsStatistic = new ArrayList<>(); + this.iterationsStatistic = new ConcurrentLinkedQueue<>(); } /** @@ -307,44 +310,49 @@ private void balance() { } private void saveIterationStatistic(Integer iterationNumber, IterationResult iR) { - ContainerBalancerTaskIterationStatusInfo iterationStatistic = new ContainerBalancerTaskIterationStatusInfo( - iterationNumber, - iR.name(), - getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, - metrics.getDataSizeMovedGBInLatestIteration(), - metrics.getNumContainerMovesScheduledInLatestIteration(), - metrics.getNumContainerMovesCompletedInLatestIteration(), - metrics.getNumContainerMovesFailedInLatestIteration(), - metrics.getNumContainerMovesTimeoutInLatestIteration(), - findTargetStrategy.getSizeEnteringNodes() - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect( - Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ), - findSourceStrategy.getSizeLeavingNodes() - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect( - Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ) - ); - iterationsStatistic.add(iterationStatistic); + lock.writeLock().lock(); + try { + ContainerBalancerTaskIterationStatusInfo iterationStatistic = new ContainerBalancerTaskIterationStatusInfo( + iterationNumber, + iR.name(), + getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, + metrics.getDataSizeMovedGBInLatestIteration(), + metrics.getNumContainerMovesScheduledInLatestIteration(), + metrics.getNumContainerMovesCompletedInLatestIteration(), + metrics.getNumContainerMovesFailedInLatestIteration(), + metrics.getNumContainerMovesTimeoutInLatestIteration(), + new ConcurrentHashMap<>(findTargetStrategy.getSizeEnteringNodes()) + .entrySet() + .stream() + .filter(Objects::nonNull) + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) + ), + new ConcurrentHashMap<>(findSourceStrategy.getSizeLeavingNodes()) + .entrySet() + .stream() + .filter(Objects::nonNull) + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) + ) + ); + iterationsStatistic.offer(iterationStatistic); + } finally { + lock.writeLock().unlock(); + } } public List getCurrentIterationsStatistic() { - - synchronized (iterationsStatistic) { + lock.readLock().lock(); + try { int lastIterationNumber = iterationsStatistic.stream() .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) .max() @@ -384,6 +392,8 @@ public List getCurrentIterationsStatis List resultList = new ArrayList<>(iterationsStatistic); resultList.add(currentIterationStatistic); return resultList; + } finally { + lock.readLock().unlock(); } } From 8b3ff5e1cc6c34d3c0550d608002c42ca700625b Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Wed, 20 Nov 2024 11:33:10 +0530 Subject: [PATCH 5/6] Replaced Map with ConcurrentHashMap --- .../balancer/AbstractFindTargetGreedy.java | 9 +- .../balancer/ContainerBalancerTask.java | 151 ++++++++---------- ...tainerBalancerTaskIterationStatusInfo.java | 14 +- .../container/balancer/FindSourceGreedy.java | 9 +- .../balancer/FindSourceStrategy.java | 4 +- .../balancer/FindTargetStrategy.java | 4 +- 6 files changed, 84 insertions(+), 107 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java index dd2d1c578940..9e288efe6444 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java @@ -32,11 +32,10 @@ import org.slf4j.Logger; import java.util.Collection; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; /** @@ -46,7 +45,7 @@ public abstract class AbstractFindTargetGreedy implements FindTargetStrategy { private Logger logger; private ContainerManager containerManager; private PlacementPolicyValidateProxy placementPolicyValidateProxy; - private Map sizeEnteringNode; + private ConcurrentHashMap sizeEnteringNode; private NodeManager nodeManager; private ContainerBalancerConfiguration config; private Double upperLimit; @@ -56,7 +55,7 @@ protected AbstractFindTargetGreedy( ContainerManager containerManager, PlacementPolicyValidateProxy placementPolicyValidateProxy, NodeManager nodeManager) { - sizeEnteringNode = new HashMap<>(); + sizeEnteringNode = new ConcurrentHashMap<>(); this.containerManager = containerManager; this.placementPolicyValidateProxy = placementPolicyValidateProxy; this.nodeManager = nodeManager; @@ -280,7 +279,7 @@ NodeManager getNodeManager() { } @Override - public Map getSizeEnteringNodes() { + public ConcurrentHashMap getSizeEnteringNodes() { return sizeEnteringNode; } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java index 613a1529c4af..25fce513e1dd 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java @@ -51,7 +51,6 @@ import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -59,8 +58,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.locks.ReentrantReadWriteLock; -import java.util.stream.Collectors; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL_DEFAULT; @@ -120,7 +117,6 @@ public class ContainerBalancerTask implements Runnable { private int nextIterationIndex; private boolean delayStart; private final ConcurrentLinkedQueue iterationsStatistic; - private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); /** * Constructs ContainerBalancerTask with the specified arguments. @@ -310,91 +306,74 @@ private void balance() { } private void saveIterationStatistic(Integer iterationNumber, IterationResult iR) { - lock.writeLock().lock(); - try { - ContainerBalancerTaskIterationStatusInfo iterationStatistic = new ContainerBalancerTaskIterationStatusInfo( - iterationNumber, - iR.name(), - getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, - metrics.getDataSizeMovedGBInLatestIteration(), - metrics.getNumContainerMovesScheduledInLatestIteration(), - metrics.getNumContainerMovesCompletedInLatestIteration(), - metrics.getNumContainerMovesFailedInLatestIteration(), - metrics.getNumContainerMovesTimeoutInLatestIteration(), - new ConcurrentHashMap<>(findTargetStrategy.getSizeEnteringNodes()) - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect( - Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ), - new ConcurrentHashMap<>(findSourceStrategy.getSizeLeavingNodes()) - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect( - Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ) - ); - iterationsStatistic.offer(iterationStatistic); - } finally { - lock.writeLock().unlock(); - } + ContainerBalancerTaskIterationStatusInfo iterationStatistic = new ContainerBalancerTaskIterationStatusInfo( + iterationNumber, + iR.name(), + getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, + metrics.getDataSizeMovedGBInLatestIteration(), + metrics.getNumContainerMovesScheduledInLatestIteration(), + metrics.getNumContainerMovesCompletedInLatestIteration(), + metrics.getNumContainerMovesFailedInLatestIteration(), + metrics.getNumContainerMovesTimeoutInLatestIteration(), + findTargetStrategy.getSizeEnteringNodes() + .entrySet() + .stream() + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + ConcurrentHashMap::new, + (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), + ConcurrentHashMap::putAll + ), + findSourceStrategy.getSizeLeavingNodes() + .entrySet() + .stream() + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + ConcurrentHashMap::new, + (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), + ConcurrentHashMap::putAll + ) + ); + iterationsStatistic.offer(iterationStatistic); } public List getCurrentIterationsStatistic() { - lock.readLock().lock(); - try { - int lastIterationNumber = iterationsStatistic.stream() - .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) - .max() - .orElse(0); - - ContainerBalancerTaskIterationStatusInfo currentIterationStatistic = new ContainerBalancerTaskIterationStatusInfo( - lastIterationNumber + 1, - null, - getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, - sizeActuallyMovedInLatestIteration / OzoneConsts.GB, - metrics.getNumContainerMovesScheduledInLatestIteration(), - metrics.getNumContainerMovesCompletedInLatestIteration(), - metrics.getNumContainerMovesFailedInLatestIteration(), - metrics.getNumContainerMovesTimeoutInLatestIteration(), - new ConcurrentHashMap<>(findTargetStrategy.getSizeEnteringNodes()) - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect(Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ), - new ConcurrentHashMap<>(findSourceStrategy.getSizeLeavingNodes()) - .entrySet() - .stream() - .filter(Objects::nonNull) - .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) - .collect( - Collectors.toMap( - entry -> entry.getKey().getUuid(), - entry -> entry.getValue() / OzoneConsts.GB - ) - ) - ); - List resultList = new ArrayList<>(iterationsStatistic); - resultList.add(currentIterationStatistic); - return resultList; - } finally { - lock.readLock().unlock(); - } + int lastIterationNumber = iterationsStatistic.stream() + .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) + .max() + .orElse(0); + + ContainerBalancerTaskIterationStatusInfo currentIterationStatistic = new ContainerBalancerTaskIterationStatusInfo( + lastIterationNumber + 1, + null, + getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB, + sizeActuallyMovedInLatestIteration / OzoneConsts.GB, + metrics.getNumContainerMovesScheduledInLatestIteration(), + metrics.getNumContainerMovesCompletedInLatestIteration(), + metrics.getNumContainerMovesFailedInLatestIteration(), + metrics.getNumContainerMovesTimeoutInLatestIteration(), + findTargetStrategy.getSizeEnteringNodes() + .entrySet() + .stream() + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + ConcurrentHashMap::new, + (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), + ConcurrentHashMap::putAll + ), + findSourceStrategy.getSizeLeavingNodes() + .entrySet() + .stream() + .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) + .collect( + ConcurrentHashMap::new, + (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), + ConcurrentHashMap::putAll + ) + ); + List resultList = new ArrayList<>(iterationsStatistic); + resultList.add(currentIterationStatistic); + return resultList; } /** diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java index 1d597b0ca273..a4be9d412690 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java @@ -18,8 +18,8 @@ package org.apache.hadoop.hdds.scm.container.balancer; -import java.util.Map; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; /** * Information about balancer task iteration. @@ -33,8 +33,8 @@ public class ContainerBalancerTaskIterationStatusInfo { private final long containerMovesCompleted; private final long containerMovesFailed; private final long containerMovesTimeout; - private final Map sizeEnteringNodesGB; - private final Map sizeLeavingNodesGB; + private final ConcurrentHashMap sizeEnteringNodesGB; + private final ConcurrentHashMap sizeLeavingNodesGB; @SuppressWarnings("checkstyle:ParameterNumber") public ContainerBalancerTaskIterationStatusInfo( @@ -46,8 +46,8 @@ public ContainerBalancerTaskIterationStatusInfo( long containerMovesCompleted, long containerMovesFailed, long containerMovesTimeout, - Map sizeEnteringNodesGB, - Map sizeLeavingNodesGB) { + ConcurrentHashMap sizeEnteringNodesGB, + ConcurrentHashMap sizeLeavingNodesGB) { this.iterationNumber = iterationNumber; this.iterationResult = iterationResult; this.sizeScheduledForMoveGB = sizeScheduledForMoveGB; @@ -92,11 +92,11 @@ public long getContainerMovesTimeout() { return containerMovesTimeout; } - public Map getSizeEnteringNodesGB() { + public ConcurrentHashMap getSizeEnteringNodesGB() { return sizeEnteringNodesGB; } - public Map getSizeLeavingNodesGB() { + public ConcurrentHashMap getSizeLeavingNodesGB() { return sizeLeavingNodesGB; } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java index 435cc9859a94..42dda0696ed7 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java @@ -26,11 +26,10 @@ import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.PriorityQueue; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; /** * The selection criteria for selecting source datanodes , the containers of @@ -39,14 +38,14 @@ public class FindSourceGreedy implements FindSourceStrategy { private static final Logger LOG = LoggerFactory.getLogger(FindSourceGreedy.class); - private Map sizeLeavingNode; + private ConcurrentHashMap sizeLeavingNode; private PriorityQueue potentialSources; private NodeManager nodeManager; private ContainerBalancerConfiguration config; private Double lowerLimit; FindSourceGreedy(NodeManager nodeManager) { - sizeLeavingNode = new HashMap<>(); + sizeLeavingNode = new ConcurrentHashMap<>(); potentialSources = new PriorityQueue<>((a, b) -> { double currentUsageOfA = a.calculateUtilization( -sizeLeavingNode.get(a.getDatanodeDetails())); @@ -203,7 +202,7 @@ public void reInitialize(List potentialDataNodes, } @Override - public Map getSizeLeavingNodes() { + public ConcurrentHashMap getSizeLeavingNodes() { return sizeLeavingNode; } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java index 9e429aaa21d9..a7851df557ee 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java @@ -24,7 +24,7 @@ import jakarta.annotation.Nonnull; import java.util.Collection; import java.util.List; -import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; /** * This interface can be used to implement strategies to get a @@ -87,5 +87,5 @@ void reInitialize(List potentialDataNodes, */ void resetPotentialSources(@Nonnull Collection sources); - Map getSizeLeavingNodes(); + ConcurrentHashMap getSizeLeavingNodes(); } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java index 389ea6e5192f..2932a9707bcf 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java @@ -25,7 +25,7 @@ import jakarta.annotation.Nonnull; import java.util.Collection; import java.util.List; -import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; /** * This interface can be used to implement strategies to find a target for a @@ -70,5 +70,5 @@ void reInitialize(List potentialDataNodes, */ void resetPotentialTargets(@Nonnull Collection targets); - Map getSizeEnteringNodes(); + ConcurrentHashMap getSizeEnteringNodes(); } From 4e3ea5bdff5de4586823b5e41a97d149cd9b68a1 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Fri, 22 Nov 2024 16:46:44 +0530 Subject: [PATCH 6/6] Use ConcurrentHashMap instead of HashMap and calc last iteration based on resultList --- .../balancer/AbstractFindTargetGreedy.java | 5 ++- .../balancer/ContainerBalancerTask.java | 37 +++++++++++-------- ...tainerBalancerTaskIterationStatusInfo.java | 14 +++---- .../container/balancer/FindSourceGreedy.java | 5 ++- .../balancer/FindSourceStrategy.java | 4 +- .../balancer/FindTargetStrategy.java | 4 +- 6 files changed, 39 insertions(+), 30 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java index 9e288efe6444..88657047a07a 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java @@ -33,6 +33,7 @@ import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -45,7 +46,7 @@ public abstract class AbstractFindTargetGreedy implements FindTargetStrategy { private Logger logger; private ContainerManager containerManager; private PlacementPolicyValidateProxy placementPolicyValidateProxy; - private ConcurrentHashMap sizeEnteringNode; + private Map sizeEnteringNode; private NodeManager nodeManager; private ContainerBalancerConfiguration config; private Double upperLimit; @@ -279,7 +280,7 @@ NodeManager getNodeManager() { } @Override - public ConcurrentHashMap getSizeEnteringNodes() { + public Map getSizeEnteringNodes() { return sizeEnteringNode; } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java index 25fce513e1dd..93d496519b26 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java @@ -51,6 +51,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Queue; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -58,6 +59,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL_DEFAULT; @@ -116,7 +118,7 @@ public class ContainerBalancerTask implements Runnable { private IterationResult iterationResult; private int nextIterationIndex; private boolean delayStart; - private final ConcurrentLinkedQueue iterationsStatistic; + private Queue iterationsStatistic; /** * Constructs ContainerBalancerTask with the specified arguments. @@ -320,25 +322,29 @@ private void saveIterationStatistic(Integer iterationNumber, IterationResult iR) .stream() .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) .collect( - ConcurrentHashMap::new, - (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), - ConcurrentHashMap::putAll + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) ), findSourceStrategy.getSizeLeavingNodes() .entrySet() .stream() .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) .collect( - ConcurrentHashMap::new, - (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), - ConcurrentHashMap::putAll + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) ) ); iterationsStatistic.offer(iterationStatistic); } public List getCurrentIterationsStatistic() { - int lastIterationNumber = iterationsStatistic.stream() + List resultList = new ArrayList<>(iterationsStatistic); + + int lastIterationNumber = resultList.stream() .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber) .max() .orElse(0); @@ -357,21 +363,22 @@ public List getCurrentIterationsStatis .stream() .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) .collect( - ConcurrentHashMap::new, - (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), - ConcurrentHashMap::putAll + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) ), findSourceStrategy.getSizeLeavingNodes() .entrySet() .stream() .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0) .collect( - ConcurrentHashMap::new, - (map, entry) -> map.put(entry.getKey().getUuid(), entry.getValue() / OzoneConsts.GB), - ConcurrentHashMap::putAll + Collectors.toMap( + entry -> entry.getKey().getUuid(), + entry -> entry.getValue() / OzoneConsts.GB + ) ) ); - List resultList = new ArrayList<>(iterationsStatistic); resultList.add(currentIterationStatistic); return resultList; } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java index a4be9d412690..1d597b0ca273 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTaskIterationStatusInfo.java @@ -18,8 +18,8 @@ package org.apache.hadoop.hdds.scm.container.balancer; +import java.util.Map; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; /** * Information about balancer task iteration. @@ -33,8 +33,8 @@ public class ContainerBalancerTaskIterationStatusInfo { private final long containerMovesCompleted; private final long containerMovesFailed; private final long containerMovesTimeout; - private final ConcurrentHashMap sizeEnteringNodesGB; - private final ConcurrentHashMap sizeLeavingNodesGB; + private final Map sizeEnteringNodesGB; + private final Map sizeLeavingNodesGB; @SuppressWarnings("checkstyle:ParameterNumber") public ContainerBalancerTaskIterationStatusInfo( @@ -46,8 +46,8 @@ public ContainerBalancerTaskIterationStatusInfo( long containerMovesCompleted, long containerMovesFailed, long containerMovesTimeout, - ConcurrentHashMap sizeEnteringNodesGB, - ConcurrentHashMap sizeLeavingNodesGB) { + Map sizeEnteringNodesGB, + Map sizeLeavingNodesGB) { this.iterationNumber = iterationNumber; this.iterationResult = iterationResult; this.sizeScheduledForMoveGB = sizeScheduledForMoveGB; @@ -92,11 +92,11 @@ public long getContainerMovesTimeout() { return containerMovesTimeout; } - public ConcurrentHashMap getSizeEnteringNodesGB() { + public Map getSizeEnteringNodesGB() { return sizeEnteringNodesGB; } - public ConcurrentHashMap getSizeLeavingNodesGB() { + public Map getSizeLeavingNodesGB() { return sizeLeavingNodesGB; } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java index 42dda0696ed7..57cc8b32b949 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java @@ -27,6 +27,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.PriorityQueue; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -38,7 +39,7 @@ public class FindSourceGreedy implements FindSourceStrategy { private static final Logger LOG = LoggerFactory.getLogger(FindSourceGreedy.class); - private ConcurrentHashMap sizeLeavingNode; + private Map sizeLeavingNode; private PriorityQueue potentialSources; private NodeManager nodeManager; private ContainerBalancerConfiguration config; @@ -202,7 +203,7 @@ public void reInitialize(List potentialDataNodes, } @Override - public ConcurrentHashMap getSizeLeavingNodes() { + public Map getSizeLeavingNodes() { return sizeLeavingNode; } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java index a7851df557ee..9e429aaa21d9 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java @@ -24,7 +24,7 @@ import jakarta.annotation.Nonnull; import java.util.Collection; import java.util.List; -import java.util.concurrent.ConcurrentHashMap; +import java.util.Map; /** * This interface can be used to implement strategies to get a @@ -87,5 +87,5 @@ void reInitialize(List potentialDataNodes, */ void resetPotentialSources(@Nonnull Collection sources); - ConcurrentHashMap getSizeLeavingNodes(); + Map getSizeLeavingNodes(); } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java index 2932a9707bcf..389ea6e5192f 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetStrategy.java @@ -25,7 +25,7 @@ import jakarta.annotation.Nonnull; import java.util.Collection; import java.util.List; -import java.util.concurrent.ConcurrentHashMap; +import java.util.Map; /** * This interface can be used to implement strategies to find a target for a @@ -70,5 +70,5 @@ void reInitialize(List potentialDataNodes, */ void resetPotentialTargets(@Nonnull Collection targets); - ConcurrentHashMap getSizeEnteringNodes(); + Map getSizeEnteringNodes(); }