This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8efd9ef0ac4 Replenish annotations to load balancers (#13675)
8efd9ef0ac4 is described below
commit 8efd9ef0ac4f08642472f0eec126819b867e8f61
Author: Yongzao <[email protected]>
AuthorDate: Thu Oct 3 23:34:26 2024 +0800
Replenish annotations to load balancers (#13675)
* annotations
* spotless
* test revise
---
.../manager/load/balancer/RegionBalancer.java | 4 +-
.../manager/load/balancer/RouteBalancer.java | 4 +-
.../region/GreedyRegionGroupAllocator.java | 57 +++--
...PartiteGraphPlacementRegionGroupAllocator.java} | 162 ++++++-------
...r.java => CostFlowSelectionLeaderBalancer.java} | 263 +++++++++++----------
.../router/leader/CFDLeaderBalancerTest.java | 6 +-
.../leader/LeaderBalancerComparisonTest.java | 4 +-
7 files changed, 250 insertions(+), 250 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
index 6ba5b34c3dc..1b553eda448 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
@@ -33,7 +33,7 @@ import org.apache.iotdb.confignode.manager.load.LoadManager;
import
org.apache.iotdb.confignode.manager.load.balancer.region.GreedyCopySetRegionGroupAllocator;
import
org.apache.iotdb.confignode.manager.load.balancer.region.GreedyRegionGroupAllocator;
import
org.apache.iotdb.confignode.manager.load.balancer.region.IRegionGroupAllocator;
-import
org.apache.iotdb.confignode.manager.load.balancer.region.PartiteGraphReplicationRegionGroupAllocator;
+import
org.apache.iotdb.confignode.manager.load.balancer.region.PartiteGraphPlacementRegionGroupAllocator;
import org.apache.iotdb.confignode.manager.node.NodeManager;
import org.apache.iotdb.confignode.manager.partition.PartitionManager;
import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
@@ -59,7 +59,7 @@ public class RegionBalancer {
this.regionGroupAllocator = new GreedyRegionGroupAllocator();
break;
case PGR:
- this.regionGroupAllocator = new
PartiteGraphReplicationRegionGroupAllocator();
+ this.regionGroupAllocator = new
PartiteGraphPlacementRegionGroupAllocator();
break;
case GCR:
default:
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
index a5d4e3d0166..77d534c237d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java
@@ -34,8 +34,8 @@ import org.apache.iotdb.confignode.manager.IManager;
import org.apache.iotdb.confignode.manager.ProcedureManager;
import org.apache.iotdb.confignode.manager.load.LoadManager;
import
org.apache.iotdb.confignode.manager.load.balancer.router.leader.AbstractLeaderBalancer;
+import
org.apache.iotdb.confignode.manager.load.balancer.router.leader.CostFlowSelectionLeaderBalancer;
import
org.apache.iotdb.confignode.manager.load.balancer.router.leader.GreedyLeaderBalancer;
-import
org.apache.iotdb.confignode.manager.load.balancer.router.leader.MinCostFlowLeaderBalancer;
import
org.apache.iotdb.confignode.manager.load.balancer.router.priority.GreedyPriorityBalancer;
import
org.apache.iotdb.confignode.manager.load.balancer.router.priority.IPriorityBalancer;
import
org.apache.iotdb.confignode.manager.load.balancer.router.priority.LeaderPriorityBalancer;
@@ -127,7 +127,7 @@ public class RouteBalancer implements
IClusterStatusSubscriber {
break;
case AbstractLeaderBalancer.CFD_POLICY:
default:
- this.leaderBalancer = new MinCostFlowLeaderBalancer();
+ this.leaderBalancer = new CostFlowSelectionLeaderBalancer();
break;
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java
index d05a8accbec..6535b426928 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/GreedyRegionGroupAllocator.java
@@ -24,23 +24,48 @@ import
org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.tsfile.utils.Pair;
-
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
+import java.util.Random;
import java.util.stream.Collectors;
-import static java.util.Map.Entry.comparingByValue;
-
/** Allocate Region Greedily */
public class GreedyRegionGroupAllocator implements IRegionGroupAllocator {
+ public static final Random RANDOM = new Random();
+
public GreedyRegionGroupAllocator() {
// Empty constructor
}
+ public static class DataNodeEntry implements Comparable<DataNodeEntry> {
+
+ public int dataNodeId;
+ public int regionCount;
+ public double freeDiskSpace;
+ public int randomWeight;
+
+ public DataNodeEntry(int dataNodeId, int regionCount, double
freeDiskSpace) {
+ this.dataNodeId = dataNodeId;
+ this.regionCount = regionCount;
+ this.freeDiskSpace = freeDiskSpace;
+ this.randomWeight = RANDOM.nextInt();
+ }
+
+ @Override
+ public int compareTo(DataNodeEntry other) {
+ if (this.regionCount != other.regionCount) {
+ return this.regionCount - other.regionCount;
+ } else if (this.freeDiskSpace != other.freeDiskSpace) {
+ return (int) (other.freeDiskSpace - this.freeDiskSpace);
+ } else {
+ return this.randomWeight - other.randomWeight;
+ }
+ }
+ }
+
@Override
public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
@@ -73,27 +98,19 @@ public class GreedyRegionGroupAllocator implements
IRegionGroupAllocator {
regionCounter.merge(dataNodeLocation.getDataNodeId(),
1, Integer::sum)));
/* Construct priority map */
- Map<TDataNodeLocation, Pair<Integer, Double>> priorityMap =
- new HashMap<>(availableDataNodeMap.size());
+ List<DataNodeEntry> entryList = new ArrayList<>();
availableDataNodeMap.forEach(
(datanodeId, dataNodeConfiguration) ->
- priorityMap.put(
- dataNodeConfiguration.getLocation(),
- new Pair<>(
+ entryList.add(
+ new DataNodeEntry(
+ datanodeId,
regionCounter.getOrDefault(datanodeId, 0),
freeDiskSpaceMap.getOrDefault(datanodeId, 0d))));
// Sort weightList
- return priorityMap.entrySet().stream()
- .sorted(
- comparingByValue(
- (o1, o2) ->
- !Objects.equals(o1.getLeft(), o2.getLeft())
- // Compare the first key(The number of Regions) by
ascending order
- ? o1.getLeft() - o2.getLeft()
- // Compare the second key(The free disk space) by
descending order
- : (int) (o2.getRight() - o1.getRight())))
- .map(entry -> entry.getKey().deepCopy())
+ return entryList.stream()
+ .sorted()
+ .map(entry -> availableDataNodeMap.get(entry.dataNodeId).getLocation())
.collect(Collectors.toList());
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java
similarity index 59%
rename from
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
rename to
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java
index b24acc1bd46..98cc84f002e 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphReplicationRegionGroupAllocator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/region/PartiteGraphPlacementRegionGroupAllocator.java
@@ -25,18 +25,20 @@ import
org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import
org.apache.iotdb.confignode.manager.load.balancer.region.GreedyRegionGroupAllocator.DataNodeEntry;
+
+import org.apache.tsfile.utils.Pair;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import java.util.TreeMap;
import java.util.stream.Collectors;
-public class PartiteGraphReplicationRegionGroupAllocator implements
IRegionGroupAllocator {
+public class PartiteGraphPlacementRegionGroupAllocator implements
IRegionGroupAllocator {
- private static final Random RANDOM = new Random();
private static final GreedyRegionGroupAllocator GREEDY_ALLOCATOR =
new GreedyRegionGroupAllocator();
@@ -52,12 +54,9 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
private Map<Integer, Integer> fakeToRealIdMap;
private int alphaDataNodeNum;
- // First Key: the sum of overlapped 2-Region combination Regions with
- // other allocated RegionGroups is minimal
- private int optimalEdgeSum;
- // Second Key: the sum of DataRegions in selected DataNodes is minimal
- private int optimalRegionSum;
- private int[] optimalAlphaNodes;
+ // Pair<combinationSum, RegionSum>
+ Pair<Integer, Integer> bestValue;
+ private int[] bestAlphaNodes;
@Override
public TRegionReplicaSet generateOptimalRegionReplicasDistribution(
@@ -67,7 +66,6 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
List<TRegionReplicaSet> databaseAllocatedRegionGroups,
int replicationFactor,
TConsensusGroupId consensusGroupId) {
-
this.regionPerDataNode =
(int)
(consensusGroupId.getType().equals(TConsensusGroupType.DataRegion)
@@ -75,11 +73,12 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
:
ConfigNodeDescriptor.getInstance().getConf().getSchemaRegionPerDataNode());
prepare(replicationFactor, availableDataNodeMap, allocatedRegionGroups);
- // Select a set of optimal alpha nodes
+ // Select alpha nodes set
for (int i = 0; i < subGraphCount; i++) {
- subGraphSearch(i, 0, alphaDataNodeNum, 0, 0, new int[alphaDataNodeNum]);
+ subGraphSearch(i, freeDiskSpaceMap);
}
- if (optimalEdgeSum == Integer.MAX_VALUE) {
+ if (bestValue.left == Integer.MAX_VALUE) {
+ // Use greedy allocator as alternative if no alpha nodes set is found
return GREEDY_ALLOCATOR.generateOptimalRegionReplicasDistribution(
availableDataNodeMap,
freeDiskSpaceMap,
@@ -89,9 +88,10 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
consensusGroupId);
}
- // Select the set of optimal beta nodes
- List<Integer> partiteNodes = partiteGraphSearch(optimalAlphaNodes[0] %
subGraphCount);
- if (partiteNodes.size() < replicationFactor - alphaDataNodeNum) {
+ // Select the beta nodes sets
+ List<Integer> betaDataNodes = partiteGraphSearch(bestAlphaNodes[0] %
subGraphCount);
+ if (betaDataNodes.size() < replicationFactor - alphaDataNodeNum) {
+ // Use greedy allocator as alternative if no beta nodes set is found
return GREEDY_ALLOCATOR.generateOptimalRegionReplicasDistribution(
availableDataNodeMap,
freeDiskSpaceMap,
@@ -101,15 +101,16 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
consensusGroupId);
}
+ // The next placement scheme is alpha \cup beta
TRegionReplicaSet result = new TRegionReplicaSet();
result.setRegionId(consensusGroupId);
for (int i = 0; i < alphaDataNodeNum; i++) {
result.addToDataNodeLocations(
-
availableDataNodeMap.get(fakeToRealIdMap.get(optimalAlphaNodes[i])).getLocation());
+
availableDataNodeMap.get(fakeToRealIdMap.get(bestAlphaNodes[i])).getLocation());
}
for (int i = 0; i < replicationFactor - alphaDataNodeNum; i++) {
result.addToDataNodeLocations(
-
availableDataNodeMap.get(fakeToRealIdMap.get(partiteNodes.get(i))).getLocation());
+
availableDataNodeMap.get(fakeToRealIdMap.get(betaDataNodes.get(i))).getLocation());
}
return result;
}
@@ -118,10 +119,11 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
int replicationFactor,
Map<Integer, TDataNodeConfiguration> availableDataNodeMap,
List<TRegionReplicaSet> allocatedRegionGroups) {
-
this.subGraphCount = replicationFactor / 2 + (replicationFactor % 2 == 0 ?
0 : 1);
this.replicationFactor = replicationFactor;
+ // Initialize the fake index for each DataNode,
+ // since the index of DataNode might not be continuous
this.fakeToRealIdMap = new TreeMap<>();
Map<Integer, Integer> realToFakeIdMap = new TreeMap<>();
this.dataNodeNum = availableDataNodeMap.size();
@@ -156,101 +158,81 @@ public class PartiteGraphReplicationRegionGroupAllocator
implements IRegionGroup
// Reset the optimal result
this.alphaDataNodeNum = replicationFactor / 2 + 1;
- this.optimalEdgeSum = Integer.MAX_VALUE;
- this.optimalRegionSum = Integer.MAX_VALUE;
- this.optimalAlphaNodes = new int[alphaDataNodeNum];
+ this.bestValue = new Pair<>(Integer.MAX_VALUE, Integer.MAX_VALUE);
+ this.bestAlphaNodes = new int[alphaDataNodeNum];
}
- private void subGraphSearch(
- int firstIndex,
- int currentReplica,
- int replicaNum,
- int combinationSum,
- int regionSum,
- int[] currentReplicaSet) {
-
- if (currentReplica == replicaNum) {
- if (combinationSum < optimalEdgeSum
- || (combinationSum == optimalEdgeSum && regionSum <
optimalRegionSum)) {
- // Reset the optimal result when a better one is found
- optimalEdgeSum = combinationSum;
- optimalRegionSum = regionSum;
- optimalAlphaNodes = Arrays.copyOf(currentReplicaSet,
replicationFactor);
- } else if (combinationSum == optimalEdgeSum
- && regionSum == optimalRegionSum
- && RANDOM.nextBoolean()) {
- optimalAlphaNodes = Arrays.copyOf(currentReplicaSet,
replicationFactor);
+ private Pair<Integer, Integer> valuation(int[] nodes) {
+ int edgeSum = 0;
+ int regionSum = 0;
+ for (int iota : nodes) {
+ for (int kappa : nodes) {
+ edgeSum += combinationCounter[iota][kappa];
}
- return;
+ regionSum += regionCounter[iota];
}
+ return new Pair<>(edgeSum, regionSum);
+ }
- for (int i = firstIndex; i < dataNodeNum; i += subGraphCount) {
- if (regionCounter[i] >= regionPerDataNode) {
- // Pruning: skip full DataNodes
- continue;
- }
- int nxtCombinationSum = combinationSum;
- for (int j = 0; j < currentReplica; j++) {
- nxtCombinationSum += combinationCounter[i][currentReplicaSet[j]];
- }
- if (combinationSum > optimalEdgeSum) {
- // Pruning: no needs for further searching when the first key
- // is bigger than the historical optimal result
- return;
+ private void subGraphSearch(int firstIndex, Map<Integer, Double>
freeDiskSpaceMap) {
+ List<DataNodeEntry> entryList = new ArrayList<>();
+ for (int index = firstIndex; index < dataNodeNum; index += subGraphCount) {
+ // Prune: skip filled DataNodes
+ if (regionCounter[index] < regionPerDataNode) {
+ entryList.add(
+ new DataNodeEntry(
+ index, regionCounter[index],
freeDiskSpaceMap.get(fakeToRealIdMap.get(index))));
}
- int nxtRegionSum = regionSum + regionCounter[i];
- if (combinationSum == optimalEdgeSum && regionSum > optimalRegionSum) {
- // Pruning: no needs for further searching when the second key
- // is bigger than the historical optimal result
- return;
+ }
+ if (entryList.size() < alphaDataNodeNum) {
+ // Skip: not enough DataNodes
+ return;
+ }
+ Collections.sort(entryList);
+ int[] alphaNodes = new int[alphaDataNodeNum];
+ for (int i = 0; i < alphaDataNodeNum - 1; i++) {
+ alphaNodes[i] = entryList.get(i).dataNodeId;
+ }
+ for (int i = alphaDataNodeNum - 1; i < entryList.size(); i++) {
+ alphaNodes[alphaDataNodeNum - 1] = entryList.get(i).dataNodeId;
+ Pair<Integer, Integer> currentValue = valuation(alphaNodes);
+ if (currentValue.left < bestValue.left
+ || (currentValue.left.equals(bestValue.left) && currentValue.right <
bestValue.right)) {
+ bestValue = currentValue;
+ System.arraycopy(alphaNodes, 0, bestAlphaNodes, 0, alphaDataNodeNum);
}
- currentReplicaSet[currentReplica] = i;
- subGraphSearch(
- i + subGraphCount,
- currentReplica + 1,
- replicaNum,
- nxtCombinationSum,
- nxtRegionSum,
- currentReplicaSet);
}
}
- private List<Integer> partiteGraphSearch(int selected) {
- List<Integer> partiteNodes = new ArrayList<>();
+ private List<Integer> partiteGraphSearch(int alphaIndex) {
+ List<Integer> betaNodes = new ArrayList<>();
+ int[] tmpNodes = new int[alphaDataNodeNum + 1];
+ System.arraycopy(bestAlphaNodes, 0, tmpNodes, 0, alphaDataNodeNum);
for (int partiteIndex = 0; partiteIndex < subGraphCount; partiteIndex++) {
- if (partiteIndex == selected) {
+ if (partiteIndex == alphaIndex) {
+ // Skip the alphaIndex subgraph
continue;
}
int selectedDataNode = -1;
- int bestScatterWidth = 0;
- int bestRegionSum = Integer.MAX_VALUE;
+ Pair<Integer, Integer> tmpValue = new Pair<>(Integer.MAX_VALUE,
Integer.MAX_VALUE);
for (int i = partiteIndex; i < dataNodeNum; i += subGraphCount) {
if (regionCounter[i] >= regionPerDataNode) {
+ // Pruning: skip filled DataNodes
continue;
}
- int scatterWidth = alphaDataNodeNum;
- for (int k = 0; k < alphaDataNodeNum; k++) {
- scatterWidth -= combinationCounter[i][optimalAlphaNodes[k]];
- }
- if (scatterWidth < bestScatterWidth) {
- continue;
- }
- if (scatterWidth > bestScatterWidth) {
- bestScatterWidth = scatterWidth;
- bestRegionSum = regionCounter[i];
- selectedDataNode = i;
- } else if (regionCounter[i] < bestRegionSum) {
- bestRegionSum = regionCounter[i];
- selectedDataNode = i;
- } else if (regionCounter[i] == bestRegionSum && RANDOM.nextBoolean()) {
+ tmpNodes[alphaDataNodeNum] = i;
+ Pair<Integer, Integer> currentValue = valuation(tmpNodes);
+ if (currentValue.left < tmpValue.left
+ || (currentValue.left.equals(tmpValue.left) && currentValue.right
< tmpValue.right)) {
+ tmpValue = currentValue;
selectedDataNode = i;
}
}
if (selectedDataNode == -1) {
return new ArrayList<>();
}
- partiteNodes.add(selectedDataNode);
+ betaNodes.add(selectedDataNode);
}
- return partiteNodes;
+ return betaNodes;
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/MinCostFlowLeaderBalancer.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CostFlowSelectionLeaderBalancer.java
similarity index 52%
rename from
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/MinCostFlowLeaderBalancer.java
rename to
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CostFlowSelectionLeaderBalancer.java
index 669f2dc9f40..588219d70c0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/MinCostFlowLeaderBalancer.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CostFlowSelectionLeaderBalancer.java
@@ -36,48 +36,48 @@ import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
/** Leader distribution balancer that uses minimum cost flow algorithm */
-public class MinCostFlowLeaderBalancer extends AbstractLeaderBalancer {
+public class CostFlowSelectionLeaderBalancer extends AbstractLeaderBalancer {
private static final int INFINITY = Integer.MAX_VALUE;
- /** Graph nodes */
- // Super source node
- private static final int S_NODE = 0;
-
- // Super terminal node
- private static final int T_NODE = 1;
- // Maximum index of graph nodes
- private int maxNode = T_NODE + 1;
- // Map<RegionGroupId, rNode>
- private final Map<TConsensusGroupId, Integer> rNodeMap;
- // Map<Database, Map<DataNodeId, sDNode>>
- private final Map<String, Map<Integer, Integer>> sDNodeMap;
- // Map<Database, Map<sDNode, DataNodeId>>
- private final Map<String, Map<Integer, Integer>> sDNodeReflect;
- // Map<DataNodeId, tDNode>
- private final Map<Integer, Integer> tDNodeMap;
+ /** Graph vertices */
+ // Super source vertex
+ private static final int S_VERTEX = 0;
+
+ // Super terminal vertex
+ private static final int T_VERTEX = 1;
+ // Maximum index of graph vertices
+ private int maxVertex = T_VERTEX + 1;
+ // Map<RegionGroupId, rVertex>
+ private final Map<TConsensusGroupId, Integer> rVertexMap;
+ // Map<Database, Map<DataNodeId, sDVertex>>
+ private final Map<String, Map<Integer, Integer>> sDVertexMap;
+ // Map<Database, Map<sDVertex, DataNodeId>>
+ private final Map<String, Map<Integer, Integer>> sDVertexReflect;
+ // Map<DataNodeId, tDVertex>
+ private final Map<Integer, Integer> tDVertexMap;
/** Graph edges */
// Maximum index of graph edges
private int maxEdge = 0;
- private final List<MinCostFlowEdge> minCostFlowEdges;
- private int[] nodeHeadEdge;
- private int[] nodeCurrentEdge;
+ private final List<CostFlowEdge> costFlowEdges;
+ private int[] vertexHeadEdge;
+ private int[] vertexCurrentEdge;
- private boolean[] isNodeVisited;
- private int[] nodeMinimumCost;
+ private boolean[] isVertexVisited;
+ private int[] vertexMinimumCost;
private int maximumFlow = 0;
private int minimumCost = 0;
- public MinCostFlowLeaderBalancer() {
+ public CostFlowSelectionLeaderBalancer() {
super();
- this.rNodeMap = new TreeMap<>();
- this.sDNodeMap = new TreeMap<>();
- this.sDNodeReflect = new TreeMap<>();
- this.tDNodeMap = new TreeMap<>();
- this.minCostFlowEdges = new ArrayList<>();
+ this.rVertexMap = new TreeMap<>();
+ this.sDVertexMap = new TreeMap<>();
+ this.sDVertexReflect = new TreeMap<>();
+ this.tDVertexMap = new TreeMap<>();
+ this.costFlowEdges = new ArrayList<>();
}
@Override
@@ -94,7 +94,7 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
dataNodeStatisticsMap,
regionStatisticsMap);
Map<TConsensusGroupId, Integer> result;
- constructMCFGraph();
+ constructFlowNetwork();
dinicAlgorithm();
result = collectLeaderDistribution();
clear();
@@ -104,45 +104,47 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
@Override
protected void clear() {
super.clear();
- this.rNodeMap.clear();
- this.sDNodeMap.clear();
- this.sDNodeReflect.clear();
- this.tDNodeMap.clear();
- this.minCostFlowEdges.clear();
- this.nodeHeadEdge = null;
- this.nodeCurrentEdge = null;
- this.isNodeVisited = null;
- this.nodeMinimumCost = null;
- this.maxNode = T_NODE + 1;
+ this.rVertexMap.clear();
+ this.sDVertexMap.clear();
+ this.sDVertexReflect.clear();
+ this.tDVertexMap.clear();
+ this.costFlowEdges.clear();
+ this.vertexHeadEdge = null;
+ this.vertexCurrentEdge = null;
+ this.isVertexVisited = null;
+ this.vertexMinimumCost = null;
+ this.maxVertex = T_VERTEX + 1;
this.maxEdge = 0;
}
- private void constructMCFGraph() {
+ private void constructFlowNetwork() {
this.maximumFlow = 0;
this.minimumCost = 0;
- /* Indicate nodes in mcf */
+ /* Indicate vertices */
for (Map.Entry<String, List<TConsensusGroupId>> databaseEntry :
databaseRegionGroupMap.entrySet()) {
String database = databaseEntry.getKey();
- sDNodeMap.put(database, new TreeMap<>());
- sDNodeReflect.put(database, new TreeMap<>());
+ sDVertexMap.put(database, new TreeMap<>());
+ sDVertexReflect.put(database, new TreeMap<>());
for (TConsensusGroupId regionGroupId : databaseEntry.getValue()) {
if (regionGroupIntersection.contains(regionGroupId)) {
- rNodeMap.put(regionGroupId, maxNode++);
+ // Map region to region vertices
+ rVertexMap.put(regionGroupId, maxVertex++);
regionLocationMap
.get(regionGroupId)
.forEach(
dataNodeId -> {
if (isDataNodeAvailable(dataNodeId)) {
- if (!sDNodeMap.get(database).containsKey(dataNodeId)) {
- sDNodeMap.get(database).put(dataNodeId, maxNode);
- sDNodeReflect.get(database).put(maxNode, dataNodeId);
- maxNode += 1;
+ // Map DataNode to DataNode vertices
+ if (!sDVertexMap.get(database).containsKey(dataNodeId)) {
+ sDVertexMap.get(database).put(dataNodeId, maxVertex);
+ sDVertexReflect.get(database).put(maxVertex,
dataNodeId);
+ maxVertex += 1;
}
- if (!tDNodeMap.containsKey(dataNodeId)) {
- tDNodeMap.put(dataNodeId, maxNode);
- maxNode += 1;
+ if (!tDVertexMap.containsKey(dataNodeId)) {
+ tDVertexMap.put(dataNodeId, maxVertex);
+ maxVertex += 1;
}
}
});
@@ -151,48 +153,49 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
}
/* Prepare arrays */
- isNodeVisited = new boolean[maxNode];
- nodeMinimumCost = new int[maxNode];
- nodeCurrentEdge = new int[maxNode];
- nodeHeadEdge = new int[maxNode];
- Arrays.fill(nodeHeadEdge, -1);
-
- /* Construct edges: sNode -> rNodes */
- for (int rNode : rNodeMap.values()) {
- // Capacity: 1, Cost: 0, each RegionGroup should elect exactly 1 leader
- addAdjacentEdges(S_NODE, rNode, 1, 0);
+ isVertexVisited = new boolean[maxVertex];
+ vertexMinimumCost = new int[maxVertex];
+ vertexCurrentEdge = new int[maxVertex];
+ vertexHeadEdge = new int[maxVertex];
+ Arrays.fill(vertexHeadEdge, -1);
+
+ /* Construct edges: sVertex -> rVertices */
+ for (int rVertex : rVertexMap.values()) {
+ // Capacity: 1, Cost: 0, select exactly 1 leader for each RegionGroup
+ addAdjacentEdges(S_VERTEX, rVertex, 1, 0);
}
- /* Construct edges: rNodes -> sdNodes */
+ /* Construct edges: rVertices -> sDVertices */
for (Map.Entry<String, List<TConsensusGroupId>> databaseEntry :
databaseRegionGroupMap.entrySet()) {
String database = databaseEntry.getKey();
for (TConsensusGroupId regionGroupId : databaseEntry.getValue()) {
if (regionGroupIntersection.contains(regionGroupId)) {
- int rNode = rNodeMap.get(regionGroupId);
+ int rVertex = rVertexMap.get(regionGroupId);
regionLocationMap
.get(regionGroupId)
.forEach(
dataNodeId -> {
if (isDataNodeAvailable(dataNodeId)
&& isRegionAvailable(regionGroupId, dataNodeId)) {
- int sDNode = sDNodeMap.get(database).get(dataNodeId);
- // Capacity: 1, Cost: 1 if sDNode is the current leader
of the rNode, 0
- // otherwise.
- // Therefore, the RegionGroup will keep the leader as
constant as possible.
+ int sDVertex = sDVertexMap.get(database).get(dataNodeId);
+ // Capacity: 1, Cost: 1 if the DataNode is the current
leader of the
+ // RegionGroup;
+ // 0 otherwise. Thus, the RegionGroup will keep the
leader as constant as
+ // possible.
int cost =
Objects.equals(
regionLeaderMap.getOrDefault(regionGroupId,
-1), dataNodeId)
? 0
: 1;
- addAdjacentEdges(rNode, sDNode, 1, cost);
+ addAdjacentEdges(rVertex, sDVertex, 1, cost);
}
});
}
}
}
- /* Construct edges: sDNodes -> tDNodes */
+ /* Construct edges: sDVertices -> tDVertices */
for (Map.Entry<String, List<TConsensusGroupId>> databaseEntry :
databaseRegionGroupMap.entrySet()) {
String database = databaseEntry.getKey();
@@ -205,21 +208,20 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
.forEach(
dataNodeId -> {
if (isDataNodeAvailable(dataNodeId)) {
- int sDNode = sDNodeMap.get(database).get(dataNodeId);
- int tDNode = tDNodeMap.get(dataNodeId);
+ int sDVertex = sDVertexMap.get(database).get(dataNodeId);
+ int tDVertex = tDVertexMap.get(dataNodeId);
int leaderCount = leaderCounter.merge(dataNodeId, 1,
Integer::sum);
- // Capacity: 1, Cost: x^2 for the x-th edge at the
current sDNode.
+ // Capacity: 1, Cost: 2*x-1 for the x-th edge at the
current sDVertex.
// Thus, the leader distribution will be as balance as
possible within each
- // Database
- // based on the Jensen's-Inequality.
- addAdjacentEdges(sDNode, tDNode, 1, leaderCount *
leaderCount);
+ // Database according to the Jensen's-Inequality.
+ addAdjacentEdges(sDVertex, tDVertex, 1, 2 * leaderCount
- 1);
}
});
}
}
}
- /* Construct edges: tDNodes -> tNode */
+ /* Construct edges: tDVertices -> tVertex */
// Map<DataNodeId, possible maximum leader>
// Count the possible maximum number of leader in each DataNode
Map<Integer, Integer> maxLeaderCounter = new TreeMap<>();
@@ -227,89 +229,88 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
(regionGroupId, dataNodeIds) ->
dataNodeIds.forEach(
dataNodeId -> {
- if (isDataNodeAvailable(dataNodeId) &&
tDNodeMap.containsKey(dataNodeId)) {
- int tDNode = tDNodeMap.get(dataNodeId);
+ if (isDataNodeAvailable(dataNodeId) &&
tDVertexMap.containsKey(dataNodeId)) {
+ int tDVertex = tDVertexMap.get(dataNodeId);
int leaderCount = maxLeaderCounter.merge(dataNodeId, 1,
Integer::sum);
- // Cost: x^2 for the x-th edge at the current dNode.
+ // Capacity: 1, Cost: 2*x-1 for the x-th edge at the
current tDVertex.
// Thus, the leader distribution will be as balance as
possible within the
- // cluster
- // Based on the Jensen's-Inequality.
- addAdjacentEdges(tDNode, T_NODE, 1, leaderCount *
leaderCount);
+ // cluster according to the Jensen's-Inequality.
+ addAdjacentEdges(tDVertex, T_VERTEX, 1, 2 * leaderCount -
1);
}
}));
}
- private void addAdjacentEdges(int fromNode, int destNode, int capacity, int
cost) {
- addEdge(fromNode, destNode, capacity, cost);
- addEdge(destNode, fromNode, 0, -cost);
+ private void addAdjacentEdges(int fromVertex, int destVertex, int capacity,
int cost) {
+ addEdge(fromVertex, destVertex, capacity, cost);
+ addEdge(destVertex, fromVertex, 0, -cost);
}
- private void addEdge(int fromNode, int destNode, int capacity, int cost) {
- MinCostFlowEdge edge = new MinCostFlowEdge(destNode, capacity, cost,
nodeHeadEdge[fromNode]);
- minCostFlowEdges.add(edge);
- nodeHeadEdge[fromNode] = maxEdge++;
+ private void addEdge(int fromVertex, int destVertex, int capacity, int cost)
{
+ CostFlowEdge edge = new CostFlowEdge(destVertex, capacity, cost,
vertexHeadEdge[fromVertex]);
+ costFlowEdges.add(edge);
+ vertexHeadEdge[fromVertex] = maxEdge++;
}
/**
- * Check whether there is an augmented path in the MCF graph by Bellman-Ford
algorithm.
+ * Check whether there is an augmented path in the flow network by SPFA
algorithm.
*
* <p>Notice: Never use Dijkstra algorithm to replace this since there might
exist negative
* circles.
*
* @return True if there exist augmented paths, false otherwise.
*/
- private boolean bellmanFordCheck() {
- Arrays.fill(isNodeVisited, false);
- Arrays.fill(nodeMinimumCost, INFINITY);
+ private boolean SPFACheck() {
+ Arrays.fill(isVertexVisited, false);
+ Arrays.fill(vertexMinimumCost, INFINITY);
Queue<Integer> queue = new LinkedList<>();
- nodeMinimumCost[S_NODE] = 0;
- isNodeVisited[S_NODE] = true;
- queue.offer(S_NODE);
+ vertexMinimumCost[S_VERTEX] = 0;
+ isVertexVisited[S_VERTEX] = true;
+ queue.offer(S_VERTEX);
while (!queue.isEmpty()) {
- int currentNode = queue.poll();
- isNodeVisited[currentNode] = false;
- for (int currentEdge = nodeHeadEdge[currentNode];
+ int currentVertex = queue.poll();
+ isVertexVisited[currentVertex] = false;
+ for (int currentEdge = vertexHeadEdge[currentVertex];
currentEdge >= 0;
- currentEdge = minCostFlowEdges.get(currentEdge).nextEdge) {
- MinCostFlowEdge edge = minCostFlowEdges.get(currentEdge);
+ currentEdge = costFlowEdges.get(currentEdge).nextEdge) {
+ CostFlowEdge edge = costFlowEdges.get(currentEdge);
if (edge.capacity > 0
- && nodeMinimumCost[currentNode] + edge.cost <
nodeMinimumCost[edge.destNode]) {
- nodeMinimumCost[edge.destNode] = nodeMinimumCost[currentNode] +
edge.cost;
- if (!isNodeVisited[edge.destNode]) {
- isNodeVisited[edge.destNode] = true;
- queue.offer(edge.destNode);
+ && vertexMinimumCost[currentVertex] + edge.cost <
vertexMinimumCost[edge.destVertex]) {
+ vertexMinimumCost[edge.destVertex] =
vertexMinimumCost[currentVertex] + edge.cost;
+ if (!isVertexVisited[edge.destVertex]) {
+ isVertexVisited[edge.destVertex] = true;
+ queue.offer(edge.destVertex);
}
}
}
}
- return nodeMinimumCost[T_NODE] < INFINITY;
+ return vertexMinimumCost[T_VERTEX] < INFINITY;
}
/** Do augmentation by dfs algorithm */
- private int dfsAugmentation(int currentNode, int inputFlow) {
- if (currentNode == T_NODE || inputFlow == 0) {
+ private int dfsAugmentation(int currentVertex, int inputFlow) {
+ if (currentVertex == T_VERTEX || inputFlow == 0) {
return inputFlow;
}
int currentEdge;
int outputFlow = 0;
- isNodeVisited[currentNode] = true;
- for (currentEdge = nodeCurrentEdge[currentNode];
+ isVertexVisited[currentVertex] = true;
+ for (currentEdge = vertexCurrentEdge[currentVertex];
currentEdge >= 0;
- currentEdge = minCostFlowEdges.get(currentEdge).nextEdge) {
- MinCostFlowEdge edge = minCostFlowEdges.get(currentEdge);
- if (nodeMinimumCost[currentNode] + edge.cost ==
nodeMinimumCost[edge.destNode]
+ currentEdge = costFlowEdges.get(currentEdge).nextEdge) {
+ CostFlowEdge edge = costFlowEdges.get(currentEdge);
+ if (vertexMinimumCost[currentVertex] + edge.cost ==
vertexMinimumCost[edge.destVertex]
&& edge.capacity > 0
- && !isNodeVisited[edge.destNode]) {
+ && !isVertexVisited[edge.destVertex]) {
- int subOutputFlow = dfsAugmentation(edge.destNode, Math.min(inputFlow,
edge.capacity));
+ int subOutputFlow = dfsAugmentation(edge.destVertex,
Math.min(inputFlow, edge.capacity));
minimumCost += subOutputFlow * edge.cost;
edge.capacity -= subOutputFlow;
- minCostFlowEdges.get(currentEdge ^ 1).capacity += subOutputFlow;
+ costFlowEdges.get(currentEdge ^ 1).capacity += subOutputFlow;
inputFlow -= subOutputFlow;
outputFlow += subOutputFlow;
@@ -319,19 +320,19 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
}
}
}
- nodeCurrentEdge[currentNode] = currentEdge;
+ vertexCurrentEdge[currentVertex] = currentEdge;
if (outputFlow > 0) {
- isNodeVisited[currentNode] = false;
+ isVertexVisited[currentVertex] = false;
}
return outputFlow;
}
private void dinicAlgorithm() {
- while (bellmanFordCheck()) {
+ while (SPFACheck()) {
int currentFlow;
- System.arraycopy(nodeHeadEdge, 0, nodeCurrentEdge, 0, maxNode);
- while ((currentFlow = dfsAugmentation(S_NODE, INFINITY)) > 0) {
+ System.arraycopy(vertexHeadEdge, 0, vertexCurrentEdge, 0, maxVertex);
+ while ((currentFlow = dfsAugmentation(S_VERTEX, INFINITY)) > 0) {
maximumFlow += currentFlow;
}
}
@@ -353,13 +354,13 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
return;
}
boolean matchLeader = false;
- for (int currentEdge =
nodeHeadEdge[rNodeMap.get(regionGroupId)];
+ for (int currentEdge =
vertexHeadEdge[rVertexMap.get(regionGroupId)];
currentEdge >= 0;
- currentEdge =
minCostFlowEdges.get(currentEdge).nextEdge) {
- MinCostFlowEdge edge = minCostFlowEdges.get(currentEdge);
- if (edge.destNode != S_NODE && edge.capacity == 0) {
+ currentEdge = costFlowEdges.get(currentEdge).nextEdge) {
+ CostFlowEdge edge = costFlowEdges.get(currentEdge);
+ if (edge.destVertex != S_VERTEX && edge.capacity == 0) {
matchLeader = true;
- result.put(regionGroupId,
sDNodeReflect.get(database).get(edge.destNode));
+ result.put(regionGroupId,
sDVertexReflect.get(database).get(edge.destVertex));
}
}
if (!matchLeader) {
@@ -380,15 +381,15 @@ public class MinCostFlowLeaderBalancer extends
AbstractLeaderBalancer {
return minimumCost;
}
- private static class MinCostFlowEdge {
+ private static class CostFlowEdge {
- private final int destNode;
+ private final int destVertex;
private int capacity;
private final int cost;
private final int nextEdge;
- private MinCostFlowEdge(int destNode, int capacity, int cost, int
nextEdge) {
- this.destNode = destNode;
+ private CostFlowEdge(int destVertex, int capacity, int cost, int nextEdge)
{
+ this.destVertex = destVertex;
this.capacity = capacity;
this.cost = cost;
this.nextEdge = nextEdge;
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java
index f2f51e044ff..218035c99be 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/CFDLeaderBalancerTest.java
@@ -47,7 +47,8 @@ import java.util.stream.Collectors;
public class CFDLeaderBalancerTest {
- private static final MinCostFlowLeaderBalancer BALANCER = new
MinCostFlowLeaderBalancer();
+ private static final CostFlowSelectionLeaderBalancer BALANCER =
+ new CostFlowSelectionLeaderBalancer();
private static final String DATABASE = "root.database";
@@ -262,8 +263,7 @@ public class CFDLeaderBalancerTest {
// The loadCost for each DataNode are the same
int x = regionGroupNum / dataNodeNum;
- // i.e. formula of (1^2 + 2^2 + 3^2 + ...) * 2
- int loadCost = x * (x + 1) * (2 * x + 1) / 3;
+ int loadCost = 2 * x * x;
int dataNodeId = 0;
Random random = new Random();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java
index 885c97989da..49efa2dcc64 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/leader/LeaderBalancerComparisonTest.java
@@ -57,8 +57,8 @@ public class LeaderBalancerComparisonTest {
private static FileWriter WRITER;
private static final GreedyLeaderBalancer GREEDY_LEADER_BALANCER = new
GreedyLeaderBalancer();
- private static final MinCostFlowLeaderBalancer MIN_COST_FLOW_LEADER_BALANCER
=
- new MinCostFlowLeaderBalancer();
+ private static final CostFlowSelectionLeaderBalancer
MIN_COST_FLOW_LEADER_BALANCER =
+ new CostFlowSelectionLeaderBalancer();
private static final Random RANDOM = new Random();
private static final int TEST_MAX_DATA_NODE_NUM = 100;
