You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/12/20 08:04:57 UTC
[49/50] [abbrv] ignite git commit: Throttle new events generation
Throttle new events generation
(cherry picked from commit d29ea50)
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f9698f4d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f9698f4d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f9698f4d
Branch: refs/heads/ignite-zk
Commit: f9698f4d60dad816e33c6d6a430fd36312a01047
Parents: 64aa9ea
Author: sboikov <sb...@gridgain.com>
Authored: Wed Dec 13 16:34:47 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Dec 20 11:00:19 2017 +0300
----------------------------------------------------------------------
.../zk/internal/ZookeeperDiscoveryImpl.java | 84 +++++++++++++++++---
1 file changed, 73 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9698f4d/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
index 7032fd8..a72d742 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
@@ -93,6 +93,12 @@ public class ZookeeperDiscoveryImpl {
static final String IGNITE_ZOOKEEPER_DISCOVERY_SPI_ACK_THRESHOLD = "IGNITE_ZOOKEEPER_DISCOVERY_SPI_ACK_THRESHOLD";
/** */
+ private static final String IGNITE_ZOOKEEPER_DISCOVERY_MAX_EVTS = "IGNITE_ZOOKEEPER_DISCOVERY_MAX_EVTS";
+
+ /** */
+ private static final String IGNITE_ZOOKEEPER_DISCOVERY_EVTS_THROTTLE = "IGNITE_ZOOKEEPER_DISCOVERY_EVTS_THROTTLE";
+
+ /** */
final ZookeeperDiscoverySpi spi;
/** */
@@ -1153,23 +1159,20 @@ public class ZookeeperDiscoveryImpl {
TreeMap<Integer, String> alives = new TreeMap<>();
- TreeMap<Long, ZookeeperClusterNode> curTop = new TreeMap<>(rtState.top.nodesByOrder);
-
- boolean newEvts = false;
-
for (String child : aliveNodes) {
Integer internalId = ZkIgnitePaths.aliveInternalId(child);
Object old = alives.put(internalId, child);
assert old == null;
-
- if (!rtState.top.nodesByInternalId.containsKey(internalId)) {
- if (processJoinOnCoordinator(curTop, internalId, child))
- newEvts = true;
- }
}
+ TreeMap<Long, ZookeeperClusterNode> curTop = new TreeMap<>(rtState.top.nodesByOrder);
+
+ int newEvts = 0;
+
+ final int MAX_NEW_EVTS = IgniteSystemProperties.getInteger(IGNITE_ZOOKEEPER_DISCOVERY_MAX_EVTS, 100);
+
List<ZookeeperClusterNode> failedNodes = null;
for (Map.Entry<Integer, ZookeeperClusterNode> e : rtState.top.nodesByInternalId.entrySet()) {
@@ -1183,11 +1186,51 @@ public class ZookeeperDiscoveryImpl {
generateNodeFail(curTop, failedNode);
- newEvts = true;
+ newEvts++;
+
+ if (newEvts == MAX_NEW_EVTS) {
+ saveAndProcessNewEvents();
+
+ if (log.isInfoEnabled()) {
+ log.info("Delay alive nodes change process, max event threshold reached [newEvts=" + newEvts +
+ ", totalEvts=" + rtState.evtsData.evts.size() + ']');
+ }
+
+ throttleNewEventsGeneration();
+
+ rtState.zkClient.getChildrenAsync(zkPaths.aliveNodesDir, rtState.watcher, rtState.watcher);
+
+ return;
+ }
}
}
- if (newEvts)
+ for (Map.Entry<Integer, String> e : alives.entrySet()) {
+ Integer internalId = e.getKey();
+
+ if (!rtState.top.nodesByInternalId.containsKey(internalId)) {
+ if (processJoinOnCoordinator(curTop, internalId, e.getValue())) {
+ newEvts++;
+
+ if (newEvts == MAX_NEW_EVTS) {
+ saveAndProcessNewEvents();
+
+ if (log.isInfoEnabled()) {
+ log.info("Delay alive nodes change process, max event threshold reached [newEvts=" + newEvts +
+ ", totalEvts=" + rtState.evtsData.evts.size() + ']');
+ }
+
+ throttleNewEventsGeneration();
+
+ rtState.zkClient.getChildrenAsync(zkPaths.aliveNodesDir, rtState.watcher, rtState.watcher);
+
+ return;
+ }
+ }
+ }
+ }
+
+ if (newEvts > 0)
saveAndProcessNewEvents();
if (failedNodes != null)
@@ -1195,6 +1238,25 @@ public class ZookeeperDiscoveryImpl {
}
/**
+ *
+ */
+ private void throttleNewEventsGeneration() {
+ long delay = IgniteSystemProperties.getLong(IGNITE_ZOOKEEPER_DISCOVERY_EVTS_THROTTLE, 0);
+
+ if (delay > 0) {
+ if (log.isInfoEnabled())
+ log.info("Sleep delay before generate new events [delay=" + delay + ']');
+
+ try {
+ Thread.sleep(delay);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ /**
* @param nodeId Node ID.
* @param prefixId Path prefix.
* @return Join data.