You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by GitBox <gi...@apache.org> on 2020/06/26 20:46:42 UTC
[GitHub] [helix] zhangmeng916 opened a new pull request #1124: add stale message
zhangmeng916 opened a new pull request #1124:
URL: https://github.com/apache/helix/pull/1124
### Issues
- [X] My PR addresses the following Helix issues and references them in the PR description:
(#1067 )
### Description
- [X] Here are some details about my PR, including screenshots of any UI changes:
This PR adds the stale message map in Instance Message cache and updates it in every current state computation stage. For those stale message, we do not put it in pending messages, so that it won't block the next state transition on that partition.
### Tests
- [ ] The following tests are written for this issue:
(List the names of added unit/integration tests)
- [ ] The following is the result of the "mvn test" command on the appropriate module:
(Copy & paste the result of "mvn test")
### Commits
- [X] My commits all reference appropriate Apache Helix GitHub issues in their subject lines. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
1. Subject is limited to 50 characters (not including Jira issue reference)
1. Subject does not end with a period
1. Subject uses the imperative mood ("add", not "adding")
1. Body wraps at 72 characters
1. Body explains "what" and "why", not "how"
### Code Quality
- [X] My diff has been formatted using helix-style.xml
(helix-style-intellij.xml if IntelliJ IDE is used)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r458303927
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -102,15 +114,22 @@ public void process(ClusterEvent event) throws Exception {
// update all pending messages to CurrentStateOutput.
private void updatePendingMessages(LiveInstance instance, Collection<Message> pendingMessages,
- CurrentStateOutput currentStateOutput, Collection<Message> pendingRelayMessages,
- Map<String, Resource> resourceMap) {
+ Collection<Message> pendingRelayMessages,
+ Map<String, Map<String, Message>> existingStaleMessages,
+ CurrentStateOutput currentStateOutput, Map<String, Resource> resourceMap) {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
// update all pending messages
for (Message message : pendingMessages) {
+ // ignore existing stale messages
+ if (existingStaleMessages.getOrDefault(instanceName, Collections.emptyMap())
Review comment:
We can call currentStateOutput.getStaleMessagesByInstance() to get the message set as the input, right?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -129,7 +148,14 @@ private void updatePendingMessages(LiveInstance instance, Collection<Message> pe
String partitionName = message.getPartitionName();
Partition partition = resource.getPartition(partitionName);
if (partition != null) {
- setMessageState(currentStateOutput, resourceName, partition, instanceName, message);
+ String currentState = currentStateOutput.getCurrentState(resourceName, partition,
+ instanceName);
+ if (_isTaskFrameworkPipeline || !isStaleMessage(message, currentState)) {
+ setMessageState(currentStateOutput, resourceName, partition, instanceName, message);
+ } else {
+ existingStaleMessages.putIfAbsent(instanceName, new HashMap<>());
Review comment:
As mentioned above, I prefer this to be an explicit set method, addStaleMessage()
##########
File path: helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
##########
@@ -544,6 +544,19 @@ public synchronized void setLiveInstances(List<LiveInstance> liveInstances) {
return _instanceMessagesCache.getMessages(instanceName);
}
+ public Map<String, Map<String, Message>> getStaleMessages() {
Review comment:
So this one will return the map for modifying. I still don't like this implicit map/list ref for modifying purposes.
As I checked the current usage, I think you can replace it with 2 explicit modifying methods,
1. addStaleMessage(instance, message)
2. removeStaleMessage(instance, message) -- or maybe always clean up when the cache is refreshed, so you don't even need this method?
As for supporting test, we can make it protected so only visible to test. Or keeping this method but always return shallow copy of the map only for the tests.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -188,14 +196,21 @@ private void generateMessage(final Resource resource, final BaseControllerDataPr
if (pendingMessage != null && shouldCleanUpPendingMessage(pendingMessage, currentState,
currentStateOutput.getEndTime(resourceName, partition, instanceName))) {
- LogUtil.logInfo(logger, _eventId, String.format(
- "Adding pending message %s on instance %s to clean up. Msg: %s->%s, current state of resource %s:%s is %s",
- pendingMessage.getMsgId(), instanceName, pendingMessage.getFromState(),
- pendingMessage.getToState(), resourceName, partition, currentState));
- if (!pendingMessagesToCleanUp.containsKey(instanceName)) {
- pendingMessagesToCleanUp.put(instanceName, new HashMap<String, Message>());
+ logAndAddToCleanUp(messagesToCleanUp, pendingMessage, instanceName, resourceName,
+ partition, currentState, PENDING_MESSAGE);
+ }
+
+ for (Message staleMessage : staleMessages) {
+ if (staleMessage == null) {
+ logger.warn("Should not contain a stale message as null");
Review comment:
Is it really possible that we have a null element in the return set?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456005033
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
##########
@@ -165,6 +168,19 @@ public void setPendingRelayMessage(String resourceName, Partition partition, Str
setStateMessage(resourceName, partition, instanceName, message, _pendingRelayMessageMap);
}
+ public Map<String, Map<String, Message>> getStaleMessageMap() {
+ return _staleMessageMap;
+ }
+
+ public void setStaleMessageMap(Map<String, Map<String, Message>> staleMessageMap) {
+ _staleMessageMap = staleMessageMap;
Review comment:
I would prefer "_staleMessageMap = new HashMap<>(staleMessageMap)" to avoid implicit reference.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -103,14 +114,20 @@ public void process(ClusterEvent event) throws Exception {
// update all pending messages to CurrentStateOutput.
private void updatePendingMessages(LiveInstance instance, Collection<Message> pendingMessages,
CurrentStateOutput currentStateOutput, Collection<Message> pendingRelayMessages,
- Map<String, Resource> resourceMap) {
+ Map<String, Resource> resourceMap, Map<String, Map<String, Message>> existingStaleMessages) {
Review comment:
nit, could you please move the existingStaleMessages so it adjacent to pendingMessages? That makes the params more organized.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -103,14 +114,20 @@ public void process(ClusterEvent event) throws Exception {
// update all pending messages to CurrentStateOutput.
private void updatePendingMessages(LiveInstance instance, Collection<Message> pendingMessages,
CurrentStateOutput currentStateOutput, Collection<Message> pendingRelayMessages,
- Map<String, Resource> resourceMap) {
+ Map<String, Resource> resourceMap, Map<String, Map<String, Message>> existingStaleMessages) {
Review comment:
Maybe pendingRelayMessages as well : )
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
##########
@@ -194,6 +210,8 @@ public String getCurrentState(String resourceName, Partition partition, String i
return null;
}
+
Review comment:
remove?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
##########
@@ -238,6 +256,16 @@ public Message getPendingMessage(String resourceName, Partition partition, Strin
return getStateMessage(resourceName, partition, instanceName, _pendingMessageMap);
}
+ public Set<Message> getStaleMessages(String instanceName) {
Review comment:
nit, maybe call it getStaleMessagesByInstance would help to differentiate it from getStaleMessageMap
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -56,11 +57,17 @@
*/
public class CurrentStateComputationStage extends AbstractBaseStage {
private static Logger LOG = LoggerFactory.getLogger(CurrentStateComputationStage.class);
+ private boolean _isTaskFrameworkPipeline = false;
@Override
public void process(ClusterEvent event) throws Exception {
_eventId = event.getEventId();
BaseControllerDataProvider cache = event.getAttribute(AttributeName.ControllerDataProvider.name());
+
+ if (cache instanceof WorkflowControllerDataProvider) {
Review comment:
This may potentially increase our future work to split the pipeline. We need a better and more centralized way to tell the pipeline type. Otherwise, once we need more types it would be very messy. For example aggregation pipeline.
Please add a TODO here that we should avoid this hacky style.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -236,6 +267,11 @@ private void updateCurrentStates(LiveInstance instance, Collection<CurrentState>
}
}
+ private void setStaleMessage(CurrentStateOutput currentStateOutput, String instanceName,
Review comment:
Any comment?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
What if we always calculate based on the current states and pending message lists? So don't update the cache, or don't have the cache even. Will that still work?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -103,14 +114,20 @@ public void process(ClusterEvent event) throws Exception {
// update all pending messages to CurrentStateOutput.
private void updatePendingMessages(LiveInstance instance, Collection<Message> pendingMessages,
CurrentStateOutput currentStateOutput, Collection<Message> pendingRelayMessages,
- Map<String, Resource> resourceMap) {
+ Map<String, Resource> resourceMap, Map<String, Map<String, Message>> existingStaleMessages) {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
// update all pending messages
for (Message message : pendingMessages) {
+ // ignore existing stale messages
+ if (existingStaleMessages.containsKey(instanceName) && existingStaleMessages.get(instanceName)
Review comment:
To simplicity,
existingStaleMessages.getOrDefault(instanceName, Collection.EMPTY_MAP).containsKey(message.getMsgId())
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -55,6 +55,8 @@
// <instance -> {<MessageId, Message>}>
private Map<String, Map<String, Message>> _messageCache = Maps.newHashMap();
+ private Map<String, Map<String, Message>> _staleMessageCache = Maps.newHashMap();
Review comment:
Please comment on what are the 2 strings in the map.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
Review comment:
Since currentStateOutput already has existingStaleMessages, we don't need existingStaleMessages in the parameter list, right?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
Is this cache designed for performance concern only?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
##########
@@ -238,6 +256,16 @@ public Message getPendingMessage(String resourceName, Partition partition, Strin
return getStateMessage(resourceName, partition, instanceName, _pendingMessageMap);
}
+ public Set<Message> getStaleMessages(String instanceName) {
+ if (_staleMessageMap.containsKey(instanceName)) {
Review comment:
Is this check necessary? You checked that if staleMessageMap is null or not later, right?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -371,6 +404,11 @@ private boolean shouldCleanUpPendingMessage(Message pendingMsg, String currentSt
}
}
+ private boolean shouldCleanUpStaleMessage(Long currentStateTransitionEndTime) {
Review comment:
nit, I think we can just put this logic to the caller. Since there is just one caller, and the logic is simple. The additional method layers will increase latency a little bit you know.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -77,8 +81,9 @@ protected void processEvent(ClusterEvent event, ResourcesStateMap resourcesState
BaseControllerDataProvider cache = event.getAttribute(AttributeName.ControllerDataProvider.name());
Map<String, Resource> resourceMap =
event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name());
- Map<String, Map<String, Message>> pendingMessagesToCleanUp = new HashMap<>();
CurrentStateOutput currentStateOutput = event.getAttribute(AttributeName.CURRENT_STATE.name());
+
+ Map<String, Map<String, Message>> pendingMessagesToCleanUp = new HashMap<>();
Review comment:
Takes me a while to figure it out. So we shall call it messagesToCleanup now, right?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -335,15 +362,21 @@ private void addGeneratedMessageToMap(final Message message,
*/
private void schedulePendingMessageCleanUp(
final Map<String, Map<String, Message>> pendingMessagesToPurge, ExecutorService workerPool,
- final HelixDataAccessor accessor) {
+ final HelixDataAccessor accessor, Map<String, Map<String, Message>> staleMessageMap) {
workerPool.submit(new Callable<Object>() {
- @Override public Object call() {
+ @Override
+ public Object call() {
for (Map.Entry<String, Map<String, Message>> entry : pendingMessagesToPurge.entrySet()) {
String instanceName = entry.getKey();
for (Message msg : entry.getValue().values()) {
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ staleMessageMap.getOrDefault(msg.getTgtName(), Collections.emptyMap())
Review comment:
So based on what I commented before, if we calculate from scratch in every pipeline, we won't need to maintain this map here, right? Or we will have some usage side this stage that we will need to update the content of the map at runtime?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -77,8 +81,9 @@ protected void processEvent(ClusterEvent event, ResourcesStateMap resourcesState
BaseControllerDataProvider cache = event.getAttribute(AttributeName.ControllerDataProvider.name());
Map<String, Resource> resourceMap =
event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name());
- Map<String, Map<String, Message>> pendingMessagesToCleanUp = new HashMap<>();
CurrentStateOutput currentStateOutput = event.getAttribute(AttributeName.CURRENT_STATE.name());
+
+ Map<String, Map<String, Message>> pendingMessagesToCleanUp = new HashMap<>();
Review comment:
Same for the names of the other parameters that are pointing to this same map.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456157207
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -236,6 +267,11 @@ private void updateCurrentStates(LiveInstance instance, Collection<CurrentState>
}
}
+ private void setStaleMessage(CurrentStateOutput currentStateOutput, String instanceName,
Review comment:
currentStateOutput is already updated with latest stale messages, and accordingly, the cache is updated. So this method is actually redundant, and that's why I used to remove it. But I think from your previous comment, you'd like to an explicit `set` operation?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on pull request #1124:
URL: https://github.com/apache/helix/pull/1124#issuecomment-662618491
This PR is ready to merge. Approved by @jiajunwang
Final commit message:
Add stale message cache to keep processed but not deleted messages. Controller will not need to wait for these messages tp be gone before sending more messages to participants.
This change will improve the rate of P2P messages and help reduce mastership handoff latency.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456156072
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
I got the first point. Should be fine. Thanks for the explanation.
Regarding the leadership switch, the new leader won't know your cached value. So it will treat the previous message ( "first M->S message") as not stale. In general, if you have any logic depends on the cache, then the leadership switch will be troublesome.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r457751659
##########
File path: helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
##########
@@ -544,6 +544,14 @@ public synchronized void setLiveInstances(List<LiveInstance> liveInstances) {
return _instanceMessagesCache.getMessages(instanceName);
}
+ public Map<String, Map<String, Message>> getStaleMessages() {
+ return _instanceMessagesCache.getStaleMessageCache();
+ }
+
+ public void setStaleMessages(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
Actually I found in `InstanceMessageCache`, more are named with `cache`, so I changed in that file.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456162902
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -236,6 +267,11 @@ private void updateCurrentStates(LiveInstance instance, Collection<CurrentState>
}
}
+ private void setStaleMessage(CurrentStateOutput currentStateOutput, String instanceName,
Review comment:
I mean this two lines:
updatePendingMessages(instance, messages.values(), relayMessages.values(), currentStateOutput,
resourceMap);
cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
The second one is redundant. currentStateOutput is already updated in the function. But seems you think it's better to have the second one?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454659906
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -151,6 +153,13 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
+ return _staleMessageCache;
+ }
+
+ public void setStaleMessageCache(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
This is not a good style. Better to make it explicitly updated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456158586
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -236,6 +267,11 @@ private void updateCurrentStates(LiveInstance instance, Collection<CurrentState>
}
}
+ private void setStaleMessage(CurrentStateOutput currentStateOutput, String instanceName,
Review comment:
You mean we can just remove it? The comment here is just saying that this private method is not necessary : )
But regarding the set, yes, I think we need explicit set. Implicit set always introduces bugs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456054406
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
The purpose of the cache is not for performance, but to keep all the historical stale messages. For example, assume a partition goes through M->S->M->S, and the first M->S message is not deleted even during the second M->S transition, if we only look at pending messages only, the first M->S message would be considered as non stale message, which causes confusion. With the cache, the stale messages would not be checked again.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r457691122
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -151,6 +155,25 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
return true;
}
+ // filter stale message cache by message cache to remove deleted messages
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
+ for (String instanceName : _staleMessageCache.keySet()) {
Review comment:
I think you can do this when the _messageCache is refreshed in refresh() method.
1. changing values in a get method is not a good style. We want to be able to call get any time, but if it has a side effect then the callers need to review the detail of the method every time if they want to use it.
2. This is for cleaning up and prevent memory leakage only, so clean up when refresh is done at the beginning of the pipeline is enough, I think.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r457755632
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java
##########
@@ -165,6 +168,19 @@ public void setPendingRelayMessage(String resourceName, Partition partition, Str
setStateMessage(resourceName, partition, instanceName, message, _pendingRelayMessageMap);
}
+ public Map<String, Map<String, Message>> getStaleMessageMap() {
+ return _staleMessageMap;
+ }
+
+ public void setStaleMessageMap(Map<String, Map<String, Message>> staleMessageMap) {
+ _staleMessageMap = staleMessageMap;
Review comment:
The this one should also be changed? cache.setStaleMessages(existingStaleMessages);
Is this a rule we have?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: (WIP)Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454120472
##########
File path: helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
##########
@@ -544,6 +544,13 @@ public synchronized void setLiveInstances(List<LiveInstance> liveInstances) {
return _instanceMessagesCache.getMessages(instanceName);
}
+ public void setStaleMessages(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
Yeah, it's never called. Originally I used it, but then I simplified the code, and it's not used any more. I've removed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454641766
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -371,6 +401,14 @@ private boolean shouldCleanUpPendingMessage(Message pendingMsg, String currentSt
}
}
+ private boolean shouldCleanUpStaleMessage(Message staleMsg, Long currentStateTransitionEndTime) {
+ if (staleMsg == null) {
Review comment:
Normally it shouldn't. But I'm not 100% sure. I saw pendingMessage has similar check. Want to keep it for safe.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang merged pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang merged pull request #1124:
URL: https://github.com/apache/helix/pull/1124
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454647384
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -151,6 +153,13 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
+ return _staleMessageCache;
+ }
+
+ public void setStaleMessageCache(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
So it's used by this way:
Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
currentStateOutput.setStaleMessageMap(existingStaleMessages);
So when the currentStateOutput's stale message gets updated, the instance cache should also be updated, right? as it's just shallow copy. Let me know if you think differently.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -344,6 +367,13 @@ private void schedulePendingMessageCleanUp(
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ if (staleMessageMap != null && staleMessageMap.containsKey(msg.getTgtName())
+ && staleMessageMap.get(msg.getTgtName()).containsKey(msg.getMsgId())) {
Review comment:
Please take a look at the updated one. I believe it's safe.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r458988730
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -148,9 +152,20 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
LOG.info(
"END: InstanceMessagesCache.refresh(), {} of Messages read from ZooKeeper. took {} ms. ",
newMessageKeys.size(), (System.currentTimeMillis() - startTime));
+
+ refreshStaleMessageCache();
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
Review comment:
I think we have this method mainly for supporting the test. Otherwise, we can have the getStaleMessageForInstance() here, right?
My suggestion is that we should not change our business logic in any way because of the test requirement. So I will say, let's move the getStaleMessageForInstance() implementation here. And leave this method as @VisibleForTesting only.
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -148,9 +152,20 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
LOG.info(
"END: InstanceMessagesCache.refresh(), {} of Messages read from ZooKeeper. took {} ms. ",
newMessageKeys.size(), (System.currentTimeMillis() - startTime));
+
+ refreshStaleMessageCache();
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
Review comment:
Or, if you can make the test to check InstanceMessagesCache only, that would be even better. But that is optional.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
##########
@@ -544,6 +544,27 @@ public synchronized void setLiveInstances(List<LiveInstance> liveInstances) {
return _instanceMessagesCache.getMessages(instanceName);
}
+ /**
+ * This function is supposed to be only used by testing purpose for safety. For "get" usage,
+ * please use getStaleMessagesByInstance.
+ */
+ public Map<String, Map<String, Message>> getStaleMessages() {
Review comment:
Let's add a @VisibleForTesting tag 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456072972
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
Then we need to define how the historical stale messages records are removed from the cache, or we will need to reboot the controller periodically, right?
Another concern is that if leadership switch happen, then you will still lose track of the first M->S message. So this is not a complete fix.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r458934261
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -188,14 +196,21 @@ private void generateMessage(final Resource resource, final BaseControllerDataPr
if (pendingMessage != null && shouldCleanUpPendingMessage(pendingMessage, currentState,
currentStateOutput.getEndTime(resourceName, partition, instanceName))) {
- LogUtil.logInfo(logger, _eventId, String.format(
- "Adding pending message %s on instance %s to clean up. Msg: %s->%s, current state of resource %s:%s is %s",
- pendingMessage.getMsgId(), instanceName, pendingMessage.getFromState(),
- pendingMessage.getToState(), resourceName, partition, currentState));
- if (!pendingMessagesToCleanUp.containsKey(instanceName)) {
- pendingMessagesToCleanUp.put(instanceName, new HashMap<String, Message>());
+ logAndAddToCleanUp(messagesToCleanUp, pendingMessage, instanceName, resourceName,
+ partition, currentState, PENDING_MESSAGE);
+ }
+
+ for (Message staleMessage : staleMessages) {
+ if (staleMessage == null) {
+ logger.warn("Should not contain a stale message as null");
Review comment:
Shouldn't be possible. Removed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r457753156
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -335,15 +363,21 @@ private void addGeneratedMessageToMap(final Message message,
*/
private void schedulePendingMessageCleanUp(
final Map<String, Map<String, Message>> pendingMessagesToPurge, ExecutorService workerPool,
- final HelixDataAccessor accessor) {
+ final HelixDataAccessor accessor, Map<String, Map<String, Message>> staleMessageMap) {
workerPool.submit(new Callable<Object>() {
- @Override public Object call() {
+ @Override
+ public Object call() {
for (Map.Entry<String, Map<String, Message>> entry : pendingMessagesToPurge.entrySet()) {
String instanceName = entry.getKey();
for (Message msg : entry.getValue().values()) {
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ staleMessageMap.getOrDefault(msg.getTgtName(), Collections.emptyMap())
Review comment:
yep, I agree. It'll go away later anyway. Shouldn't be a concern to leave it a little longer.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456211231
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
I think about it, I don't think it'll cause any problem even when controller switches. The message won't be processed if the target state is not the ideal state. The only problem is that the code won't automatically remove it for us.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r458314319
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -102,15 +114,22 @@ public void process(ClusterEvent event) throws Exception {
// update all pending messages to CurrentStateOutput.
private void updatePendingMessages(LiveInstance instance, Collection<Message> pendingMessages,
- CurrentStateOutput currentStateOutput, Collection<Message> pendingRelayMessages,
- Map<String, Resource> resourceMap) {
+ Collection<Message> pendingRelayMessages,
+ Map<String, Map<String, Message>> existingStaleMessages,
+ CurrentStateOutput currentStateOutput, Map<String, Resource> resourceMap) {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
// update all pending messages
for (Message message : pendingMessages) {
+ // ignore existing stale messages
+ if (existingStaleMessages.getOrDefault(instanceName, Collections.emptyMap())
Review comment:
CurrentStateOutput currently does not have stale messages in it. It's not necessary because every class has access to cache.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r458998999
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -148,9 +152,20 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
LOG.info(
"END: InstanceMessagesCache.refresh(), {} of Messages read from ZooKeeper. took {} ms. ",
newMessageKeys.size(), (System.currentTimeMillis() - startTime));
+
+ refreshStaleMessageCache();
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
Review comment:
Tried to only use `getStaleMessageCache()` in this class. But it's difficult to change the test to check this class directly. So keep the function in two classes.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454596383
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -344,6 +367,13 @@ private void schedulePendingMessageCleanUp(
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ if (staleMessageMap != null && staleMessageMap.containsKey(msg.getTgtName())
+ && staleMessageMap.get(msg.getTgtName()).containsKey(msg.getMsgId())) {
Review comment:
nit, could be simpler by
if (staleMessageMap != null && staleMessageMap.getOrDefault(msg.getTgtName(), EmptyMap()).remove(msg.getMsgId())) {
if (staleMessageMap.get(msg.getTgtName()).size() == 0) {
staleMessageMap.remove(msg.getTgtName());
}
}
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -371,6 +401,14 @@ private boolean shouldCleanUpPendingMessage(Message pendingMsg, String currentSt
}
}
+ private boolean shouldCleanUpStaleMessage(Message staleMsg, Long currentStateTransitionEndTime) {
+ if (staleMsg == null) {
Review comment:
Can staleMsg be null?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -344,6 +367,13 @@ private void schedulePendingMessageCleanUp(
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ if (staleMessageMap != null && staleMessageMap.containsKey(msg.getTgtName())
Review comment:
I think staleMessageMap won't be null.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -188,14 +192,24 @@ private void generateMessage(final Resource resource, final BaseControllerDataPr
if (pendingMessage != null && shouldCleanUpPendingMessage(pendingMessage, currentState,
currentStateOutput.getEndTime(resourceName, partition, instanceName))) {
- LogUtil.logInfo(logger, _eventId, String.format(
- "Adding pending message %s on instance %s to clean up. Msg: %s->%s, current state of resource %s:%s is %s",
- pendingMessage.getMsgId(), instanceName, pendingMessage.getFromState(),
- pendingMessage.getToState(), resourceName, partition, currentState));
- if (!pendingMessagesToCleanUp.containsKey(instanceName)) {
- pendingMessagesToCleanUp.put(instanceName, new HashMap<String, Message>());
+ String logMsg = String.format(
Review comment:
This also helps to reduce the duplicate code that logs the stale messages.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -188,14 +192,24 @@ private void generateMessage(final Resource resource, final BaseControllerDataPr
if (pendingMessage != null && shouldCleanUpPendingMessage(pendingMessage, currentState,
currentStateOutput.getEndTime(resourceName, partition, instanceName))) {
- LogUtil.logInfo(logger, _eventId, String.format(
- "Adding pending message %s on instance %s to clean up. Msg: %s->%s, current state of resource %s:%s is %s",
- pendingMessage.getMsgId(), instanceName, pendingMessage.getFromState(),
- pendingMessage.getToState(), resourceName, partition, currentState));
- if (!pendingMessagesToCleanUp.containsKey(instanceName)) {
- pendingMessagesToCleanUp.put(instanceName, new HashMap<String, Message>());
+ String logMsg = String.format(
Review comment:
Better to move this logic into the logAndAddToCleanUp() method too.
This will increase the parameters, but the code here would be much simpler.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r457697703
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -335,15 +363,21 @@ private void addGeneratedMessageToMap(final Message message,
*/
private void schedulePendingMessageCleanUp(
final Map<String, Map<String, Message>> pendingMessagesToPurge, ExecutorService workerPool,
- final HelixDataAccessor accessor) {
+ final HelixDataAccessor accessor, Map<String, Map<String, Message>> staleMessageMap) {
workerPool.submit(new Callable<Object>() {
- @Override public Object call() {
+ @Override
+ public Object call() {
for (Map.Entry<String, Map<String, Message>> entry : pendingMessagesToPurge.entrySet()) {
String instanceName = entry.getKey();
for (Message msg : entry.getValue().values()) {
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ staleMessageMap.getOrDefault(msg.getTgtName(), Collections.emptyMap())
Review comment:
Now I have a better understanding of your change. And I think this cleanup is not necessary.
The stale message map in the current state output object is re-calculated from the cache every time. So as long as we clean the cache properly, we don't need to care about the output map. Please correct me if I missed anything.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
##########
@@ -544,6 +544,14 @@ public synchronized void setLiveInstances(List<LiveInstance> liveInstances) {
return _instanceMessagesCache.getMessages(instanceName);
}
+ public Map<String, Map<String, Message>> getStaleMessages() {
+ return _instanceMessagesCache.getStaleMessageCache();
+ }
+
+ public void setStaleMessages(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
nit, cacheStaleMessages so all the updating message cache methods are named in the same way?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +87,20 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages =
Review comment:
nit, can we follow the same style that how the other 2 maps are used here?
Basically passing the existingStaleMessages to updatePendingMessages too instead of set to the currentStateOutput 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: (WIP)Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454118361
##########
File path: helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
##########
@@ -544,6 +544,13 @@ public synchronized void setLiveInstances(List<LiveInstance> liveInstances) {
return _instanceMessagesCache.getMessages(instanceName);
}
+ public void setStaleMessages(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
This is never called? Or there are some other changes that you haven't pushed?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -56,11 +57,17 @@
*/
public class CurrentStateComputationStage extends AbstractBaseStage {
private static Logger LOG = LoggerFactory.getLogger(CurrentStateComputationStage.class);
+ private boolean isTaskFrameworkPipeline = false;
Review comment:
nit, _ isTaskFrameworkPipeline
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456162348
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
I see what you mean. I was talking about the case that message is gone from zk, but still in cache. So if the message hasn't been deleted from ZK yet, then yes, it'll stay there, and may cause confusion. I'm not sure whether this would cause a logic error. But I cannot think of other option unless we mark it in ZK.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454660454
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -344,6 +367,13 @@ private void schedulePendingMessageCleanUp(
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ if (staleMessageMap != null && staleMessageMap.containsKey(msg.getTgtName())
+ && staleMessageMap.get(msg.getTgtName()).containsKey(msg.getMsgId())) {
Review comment:
Both versions are safe. Just too verbose.
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -371,6 +401,14 @@ private boolean shouldCleanUpPendingMessage(Message pendingMsg, String currentSt
}
}
+ private boolean shouldCleanUpStaleMessage(Message staleMsg, Long currentStateTransitionEndTime) {
+ if (staleMsg == null) {
Review comment:
I personally prefer to fail early. If it should not, then we should not check. The concern is that overprotected code will hide potential issues.
For example, if some bugs make it always null, then you will always get false as a result. But it would be better that we get an NPE, then the bug will be identified the first time.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456154161
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -77,15 +84,19 @@ public void process(ClusterEvent event) throws Exception {
String instanceName = instance.getInstanceName();
String instanceSessionId = instance.getEphemeralOwner();
- // update pending messages
- Map<String, Message> messages = cache.getMessages(instanceName);
- Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
- updatePendingMessages(instance, messages.values(), currentStateOutput, relayMessages.values(), resourceMap);
-
// update current states.
Map<String, CurrentState> currentStateMap = cache.getCurrentState(instanceName,
instanceSessionId);
updateCurrentStates(instance, currentStateMap.values(), currentStateOutput, resourceMap);
+
+ Map<String, Map<String, Message>> existingStaleMessages = cache.getStaleMessages();
+ currentStateOutput.setStaleMessageMap(existingStaleMessages);
+ // update pending messages
+ Map<String, Message> messages = cache.getMessages(instanceName);
+ Map<String, Message> relayMessages = cache.getRelayMessages(instanceName);
+ updatePendingMessages(instance, messages.values(), currentStateOutput,
+ relayMessages.values(), resourceMap, existingStaleMessages);
+ cache.setStaleMessages(currentStateOutput.getStaleMessageMap());
Review comment:
The stale message is periodically cleaned up the same way as `pending message`, and the logic is in message generation phase.
If controller switch happens, the in memory data is just lost right? Then all stale messages are automatically cleaned up. Any concern?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456160484
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -335,15 +362,21 @@ private void addGeneratedMessageToMap(final Message message,
*/
private void schedulePendingMessageCleanUp(
final Map<String, Map<String, Message>> pendingMessagesToPurge, ExecutorService workerPool,
- final HelixDataAccessor accessor) {
+ final HelixDataAccessor accessor, Map<String, Map<String, Message>> staleMessageMap) {
workerPool.submit(new Callable<Object>() {
- @Override public Object call() {
+ @Override
+ public Object call() {
for (Map.Entry<String, Map<String, Message>> entry : pendingMessagesToPurge.entrySet()) {
String instanceName = entry.getKey();
for (Message msg : entry.getValue().values()) {
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ staleMessageMap.getOrDefault(msg.getTgtName(), Collections.emptyMap())
Review comment:
We do need for maintaining the history from previous pipelines. This could help better reduce stale messages.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454591410
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -151,6 +153,13 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
+ return _staleMessageCache;
+ }
+
+ public void setStaleMessageCache(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
Not used?
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/InstanceMessagesCache.java
##########
@@ -151,6 +153,13 @@ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liv
return true;
}
+ public Map<String, Map<String, Message>> getStaleMessageCache() {
+ return _staleMessageCache;
+ }
+
+ public void setStaleMessageCache(Map<String, Map<String, Message>> staleMessageMap) {
Review comment:
If no set is called, how the items in the cache being updated?
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -191,6 +213,15 @@ private void updatePendingMessages(LiveInstance instance, Collection<Message> pe
}
}
+ private boolean isStaleMessage(Message message, String currentState) {
+ if (_isTaskFrameworkPipeline || currentState == null || message.getFromState() == null
Review comment:
Introducing _isTaskFrameworkPipeline to isStaleMessage is confusing. I would prefer do this,
"if (!_isTaskFrameworkPipeline && isStaleMessage(...)) {set stale message}"
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -236,6 +267,11 @@ private void updateCurrentStates(LiveInstance instance, Collection<CurrentState>
}
}
+ private void setStaleMessage(CurrentStateOutput currentStateOutput, String instanceName,
Review comment:
Can we just call "currentStateOutput.setStaleMessage(instanceName, message);" in the caller so we don't need this method?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r456631442
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateComputationStage.java
##########
@@ -236,6 +267,11 @@ private void updateCurrentStates(LiveInstance instance, Collection<CurrentState>
}
}
+ private void setStaleMessage(CurrentStateOutput currentStateOutput, String instanceName,
Review comment:
This is related to my other comment. In general, we shall not pass a modifiable map ref directly. Especially if the passed ref will be used to update. And given that you make a shallow copy instead of just pass the ref, then we will need both.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #1124: Add stale message map to improve P2P message rate
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #1124:
URL: https://github.com/apache/helix/pull/1124#discussion_r454630326
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/MessageGenerationPhase.java
##########
@@ -344,6 +367,13 @@ private void schedulePendingMessageCleanUp(
if (accessor.removeProperty(msg.getKey(accessor.keyBuilder(), instanceName))) {
LogUtil.logInfo(logger, _eventId, String
.format("Deleted message %s from instance %s", msg.getMsgId(), instanceName));
+ if (staleMessageMap != null && staleMessageMap.containsKey(msg.getTgtName())
Review comment:
I think it won't be. It's got instantiated in `CurrentStateOutput`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org