You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/07/05 03:56:48 UTC
[GitHub] [iotdb] CRZbulabula opened a new pull request, #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
CRZbulabula opened a new pull request, #6590:
URL: https://github.com/apache/iotdb/pull/6590
Update the RegionRouteMap cache in PartitionCache based on the heartbeat sample. Update algorithms are distinguished according to different routing policies:
1. For greedy policy, update RegionRouteMap when there exists some DataNodes fail down.
2. For leader policy, update RegionRouteMap when the leadership of Regions changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] xingtanzjr commented on a diff in pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
xingtanzjr commented on code in PR #6590:
URL: https://github.com/apache/iotdb/pull/6590#discussion_r914580325
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/RegionGroupCache.java:
##########
@@ -18,26 +18,75 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
public class RegionGroupCache implements IRegionGroupCache {
// TODO: This class might be split into SchemaRegionGroupCache and DataRegionGroupCache
- private long timestamp;
+ private static final int maximumWindowSize = 100;
+ // Map<DataNodeId(where a RegionReplica resides), LinkedList<RegionHeartbeatSample>>
+ private final Map<Integer, LinkedList<RegionHeartbeatSample>> slidingWindow;
+ // Indicates the version of the statistics
+ private final AtomicLong versionTimestamp;
+ // The DataNode where the leader resides
private final AtomicInteger leaderDataNodeId;
public RegionGroupCache() {
+ this.slidingWindow = new ConcurrentHashMap<>();
+
+ this.versionTimestamp = new AtomicLong(0);
this.leaderDataNodeId = new AtomicInteger(-1);
}
@Override
- public synchronized void updateLeader(long timestamp, int dataNodeId) {
- if (timestamp > this.timestamp) {
- this.timestamp = timestamp;
- this.leaderDataNodeId.set(dataNodeId);
+ public void cacheHeartbeatSample(RegionHeartbeatSample newHeartbeatSample) {
+ slidingWindow.putIfAbsent(newHeartbeatSample.getBelongedDataNodeId(), new LinkedList<>());
+ synchronized (slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId())) {
+ LinkedList<RegionHeartbeatSample> samples =
+ slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId());
+
+ // Only sequential HeartbeatSamples are accepted.
+ // And un-sequential HeartbeatSamples will be discarded.
+ if (samples.size() == 0
+ || samples.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ samples.add(newHeartbeatSample);
+ }
+
+ while (samples.size() > maximumWindowSize) {
+ samples.removeFirst();
+ }
+ }
+ }
+
+ @Override
+ public boolean updateLoadStatistic() {
+ long updateVersion = Long.MIN_VALUE;
+ int updateLeaderDataNodeId = -1;
+
+ synchronized (slidingWindow) {
+ for (LinkedList<RegionHeartbeatSample> samples : slidingWindow.values()) {
+ if (samples.size() > 0) {
+ RegionHeartbeatSample lastSample = samples.getLast();
+ if (lastSample.getSendTimestamp() > updateVersion && lastSample.isLeader()) {
+ updateVersion = lastSample.getSendTimestamp();
+ updateLeaderDataNodeId = lastSample.getBelongedDataNodeId();
+ }
+ }
+ }
+ }
+
+ if (updateVersion > versionTimestamp.get()) {
+ versionTimestamp.set(updateVersion);
+ leaderDataNodeId.set(updateLeaderDataNodeId);
Review Comment:
Is the check for leaderDataNodeId change necessary here ?
My question is that whether it should return `false` if the leader is not changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] xingtanzjr commented on a diff in pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
xingtanzjr commented on code in PR #6590:
URL: https://github.com/apache/iotdb/pull/6590#discussion_r914580850
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/ConfigNodeHeartbeatCache.java:
##########
@@ -18,33 +18,37 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
+import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.confignode.manager.load.LoadManager;
import java.util.LinkedList;
-public class ConfigNodeHeartbeatCache implements IHeartbeatStatistic {
+public class ConfigNodeHeartbeatCache implements INodeCache {
// Cache heartbeat samples
private static final int maximumWindowSize = 100;
- private final LinkedList<HeartbeatPackage> slidingWindow;
+ private final LinkedList<NodeHeartbeatSample> slidingWindow;
+
+ private final TConfigNodeLocation configNodeLocation;
// For showing cluster
private volatile NodeStatus status;
- public ConfigNodeHeartbeatCache() {
+ public ConfigNodeHeartbeatCache(TConfigNodeLocation configNodeLocation) {
+ this.configNodeLocation = configNodeLocation;
this.slidingWindow = new LinkedList<>();
-
this.status = NodeStatus.Running;
}
@Override
- public void cacheHeartBeat(HeartbeatPackage newHeartbeat) {
+ public void cacheHeartbeatSample(NodeHeartbeatSample newHeartbeatSample) {
synchronized (slidingWindow) {
// Only sequential heartbeats are accepted.
// And un-sequential heartbeats will be discarded.
if (slidingWindow.size() == 0
- || slidingWindow.getLast().getSendTimestamp() < newHeartbeat.getSendTimestamp()) {
- slidingWindow.add(newHeartbeat);
+ || slidingWindow.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ slidingWindow.add(newHeartbeatSample);
}
while (slidingWindow.size() > maximumWindowSize) {
Review Comment:
`if` is enough here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] xingtanzjr commented on a diff in pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
xingtanzjr commented on code in PR #6590:
URL: https://github.com/apache/iotdb/pull/6590#discussion_r914597147
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/RegionGroupCache.java:
##########
@@ -18,26 +18,75 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
public class RegionGroupCache implements IRegionGroupCache {
// TODO: This class might be split into SchemaRegionGroupCache and DataRegionGroupCache
- private long timestamp;
+ private static final int maximumWindowSize = 100;
+ // Map<DataNodeId(where a RegionReplica resides), LinkedList<RegionHeartbeatSample>>
+ private final Map<Integer, LinkedList<RegionHeartbeatSample>> slidingWindow;
+ // Indicates the version of the statistics
+ private final AtomicLong versionTimestamp;
+ // The DataNode where the leader resides
private final AtomicInteger leaderDataNodeId;
public RegionGroupCache() {
+ this.slidingWindow = new ConcurrentHashMap<>();
+
+ this.versionTimestamp = new AtomicLong(0);
this.leaderDataNodeId = new AtomicInteger(-1);
}
@Override
- public synchronized void updateLeader(long timestamp, int dataNodeId) {
- if (timestamp > this.timestamp) {
- this.timestamp = timestamp;
- this.leaderDataNodeId.set(dataNodeId);
+ public void cacheHeartbeatSample(RegionHeartbeatSample newHeartbeatSample) {
+ slidingWindow.putIfAbsent(newHeartbeatSample.getBelongedDataNodeId(), new LinkedList<>());
+ synchronized (slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId())) {
+ LinkedList<RegionHeartbeatSample> samples =
+ slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId());
+
+ // Only sequential HeartbeatSamples are accepted.
+ // And un-sequential HeartbeatSamples will be discarded.
+ if (samples.size() == 0
+ || samples.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ samples.add(newHeartbeatSample);
+ }
+
+ while (samples.size() > maximumWindowSize) {
+ samples.removeFirst();
+ }
+ }
+ }
+
+ @Override
+ public boolean updateLoadStatistic() {
+ long updateVersion = Long.MIN_VALUE;
+ int updateLeaderDataNodeId = -1;
+
+ synchronized (slidingWindow) {
+ for (LinkedList<RegionHeartbeatSample> samples : slidingWindow.values()) {
+ if (samples.size() > 0) {
+ RegionHeartbeatSample lastSample = samples.getLast();
+ if (lastSample.getSendTimestamp() > updateVersion && lastSample.isLeader()) {
+ updateVersion = lastSample.getSendTimestamp();
+ updateLeaderDataNodeId = lastSample.getBelongedDataNodeId();
+ }
+ }
+ }
+ }
+
+ if (updateVersion > versionTimestamp.get()) {
+ versionTimestamp.set(updateVersion);
+ leaderDataNodeId.set(updateLeaderDataNodeId);
Review Comment:
False alarm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] xingtanzjr merged pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
xingtanzjr merged PR #6590:
URL: https://github.com/apache/iotdb/pull/6590
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] xingtanzjr commented on a diff in pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
xingtanzjr commented on code in PR #6590:
URL: https://github.com/apache/iotdb/pull/6590#discussion_r914581113
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/RegionGroupCache.java:
##########
@@ -18,26 +18,75 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
public class RegionGroupCache implements IRegionGroupCache {
// TODO: This class might be split into SchemaRegionGroupCache and DataRegionGroupCache
- private long timestamp;
+ private static final int maximumWindowSize = 100;
+ // Map<DataNodeId(where a RegionReplica resides), LinkedList<RegionHeartbeatSample>>
+ private final Map<Integer, LinkedList<RegionHeartbeatSample>> slidingWindow;
+ // Indicates the version of the statistics
+ private final AtomicLong versionTimestamp;
+ // The DataNode where the leader resides
private final AtomicInteger leaderDataNodeId;
public RegionGroupCache() {
+ this.slidingWindow = new ConcurrentHashMap<>();
+
+ this.versionTimestamp = new AtomicLong(0);
this.leaderDataNodeId = new AtomicInteger(-1);
}
@Override
- public synchronized void updateLeader(long timestamp, int dataNodeId) {
- if (timestamp > this.timestamp) {
- this.timestamp = timestamp;
- this.leaderDataNodeId.set(dataNodeId);
+ public void cacheHeartbeatSample(RegionHeartbeatSample newHeartbeatSample) {
+ slidingWindow.putIfAbsent(newHeartbeatSample.getBelongedDataNodeId(), new LinkedList<>());
+ synchronized (slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId())) {
+ LinkedList<RegionHeartbeatSample> samples =
+ slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId());
+
+ // Only sequential HeartbeatSamples are accepted.
+ // And un-sequential HeartbeatSamples will be discarded.
+ if (samples.size() == 0
+ || samples.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ samples.add(newHeartbeatSample);
+ }
+
+ while (samples.size() > maximumWindowSize) {
Review Comment:
Using `if` is enough`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] xingtanzjr commented on a diff in pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
xingtanzjr commented on code in PR #6590:
URL: https://github.com/apache/iotdb/pull/6590#discussion_r914523990
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/DataNodeHeartbeatCache.java:
##########
@@ -44,13 +44,13 @@ public DataNodeHeartbeatCache() {
}
@Override
- public void cacheHeartBeat(HeartbeatPackage newHeartbeat) {
+ public void cacheHeartbeatSample(NodeHeartbeatSample newHeartbeatSample) {
synchronized (slidingWindow) {
- // Only sequential heartbeats are accepted.
- // And un-sequential heartbeats will be discarded.
+ // Only sequential HeartbeatSamples are accepted.
+ // And un-sequential HeartbeatSamples will be discarded.
if (slidingWindow.size() == 0
- || slidingWindow.getLast().getSendTimestamp() < newHeartbeat.getSendTimestamp()) {
- slidingWindow.add(newHeartbeat);
+ || slidingWindow.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ slidingWindow.add(newHeartbeatSample);
}
while (slidingWindow.size() > maximumWindowSize) {
Review Comment:
Be confident, using `if` is enough here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #6590: [IOTDB-3711] Update RegionRouteMap based on heartbeat sample
Posted by GitBox <gi...@apache.org>.
CRZbulabula commented on code in PR #6590:
URL: https://github.com/apache/iotdb/pull/6590#discussion_r914605052
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/DataNodeHeartbeatCache.java:
##########
@@ -44,13 +44,13 @@ public DataNodeHeartbeatCache() {
}
@Override
- public void cacheHeartBeat(HeartbeatPackage newHeartbeat) {
+ public void cacheHeartbeatSample(NodeHeartbeatSample newHeartbeatSample) {
synchronized (slidingWindow) {
- // Only sequential heartbeats are accepted.
- // And un-sequential heartbeats will be discarded.
+ // Only sequential HeartbeatSamples are accepted.
+ // And un-sequential HeartbeatSamples will be discarded.
if (slidingWindow.size() == 0
- || slidingWindow.getLast().getSendTimestamp() < newHeartbeat.getSendTimestamp()) {
- slidingWindow.add(newHeartbeat);
+ || slidingWindow.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ slidingWindow.add(newHeartbeatSample);
}
while (slidingWindow.size() > maximumWindowSize) {
Review Comment:
fixed~
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/ConfigNodeHeartbeatCache.java:
##########
@@ -18,33 +18,37 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
+import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.confignode.manager.load.LoadManager;
import java.util.LinkedList;
-public class ConfigNodeHeartbeatCache implements IHeartbeatStatistic {
+public class ConfigNodeHeartbeatCache implements INodeCache {
// Cache heartbeat samples
private static final int maximumWindowSize = 100;
- private final LinkedList<HeartbeatPackage> slidingWindow;
+ private final LinkedList<NodeHeartbeatSample> slidingWindow;
+
+ private final TConfigNodeLocation configNodeLocation;
// For showing cluster
private volatile NodeStatus status;
- public ConfigNodeHeartbeatCache() {
+ public ConfigNodeHeartbeatCache(TConfigNodeLocation configNodeLocation) {
+ this.configNodeLocation = configNodeLocation;
this.slidingWindow = new LinkedList<>();
-
this.status = NodeStatus.Running;
}
@Override
- public void cacheHeartBeat(HeartbeatPackage newHeartbeat) {
+ public void cacheHeartbeatSample(NodeHeartbeatSample newHeartbeatSample) {
synchronized (slidingWindow) {
// Only sequential heartbeats are accepted.
// And un-sequential heartbeats will be discarded.
if (slidingWindow.size() == 0
- || slidingWindow.getLast().getSendTimestamp() < newHeartbeat.getSendTimestamp()) {
- slidingWindow.add(newHeartbeat);
+ || slidingWindow.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ slidingWindow.add(newHeartbeatSample);
}
while (slidingWindow.size() > maximumWindowSize) {
Review Comment:
fixed~
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/RegionGroupCache.java:
##########
@@ -18,26 +18,75 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
public class RegionGroupCache implements IRegionGroupCache {
// TODO: This class might be split into SchemaRegionGroupCache and DataRegionGroupCache
- private long timestamp;
+ private static final int maximumWindowSize = 100;
+ // Map<DataNodeId(where a RegionReplica resides), LinkedList<RegionHeartbeatSample>>
+ private final Map<Integer, LinkedList<RegionHeartbeatSample>> slidingWindow;
+ // Indicates the version of the statistics
+ private final AtomicLong versionTimestamp;
+ // The DataNode where the leader resides
private final AtomicInteger leaderDataNodeId;
public RegionGroupCache() {
+ this.slidingWindow = new ConcurrentHashMap<>();
+
+ this.versionTimestamp = new AtomicLong(0);
this.leaderDataNodeId = new AtomicInteger(-1);
}
@Override
- public synchronized void updateLeader(long timestamp, int dataNodeId) {
- if (timestamp > this.timestamp) {
- this.timestamp = timestamp;
- this.leaderDataNodeId.set(dataNodeId);
+ public void cacheHeartbeatSample(RegionHeartbeatSample newHeartbeatSample) {
+ slidingWindow.putIfAbsent(newHeartbeatSample.getBelongedDataNodeId(), new LinkedList<>());
+ synchronized (slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId())) {
+ LinkedList<RegionHeartbeatSample> samples =
+ slidingWindow.get(newHeartbeatSample.getBelongedDataNodeId());
+
+ // Only sequential HeartbeatSamples are accepted.
+ // And un-sequential HeartbeatSamples will be discarded.
+ if (samples.size() == 0
+ || samples.getLast().getSendTimestamp() < newHeartbeatSample.getSendTimestamp()) {
+ samples.add(newHeartbeatSample);
+ }
+
+ while (samples.size() > maximumWindowSize) {
Review Comment:
fixed~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org