http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java index 9402a32..38450df 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataTransport.java @@ -155,22 +155,30 @@ final class BinaryMetadataTransport { * @param metadata Metadata proposed for update. * @return Future to wait for update result on. */ - GridFutureAdapter<MetadataUpdateResult> requestMetadataUpdate(BinaryMetadata metadata) throws IgniteCheckedException { + GridFutureAdapter<MetadataUpdateResult> requestMetadataUpdate(BinaryMetadata metadata) { MetadataUpdateResultFuture resFut = new MetadataUpdateResultFuture(); if (log.isDebugEnabled()) log.debug("Requesting metadata update for " + metadata.typeId() + "; caller thread is blocked on future " + resFut); - synchronized (this) { - unlabeledFutures.add(resFut); + try { + synchronized (this) { + unlabeledFutures.add(resFut); - if (!stopping) - discoMgr.sendCustomEvent(new MetadataUpdateProposedMessage(metadata, ctx.localNodeId())); - else - resFut.onDone(MetadataUpdateResult.createUpdateDisabledResult()); + if (!stopping) + discoMgr.sendCustomEvent(new MetadataUpdateProposedMessage(metadata, ctx.localNodeId())); + else + resFut.onDone(MetadataUpdateResult.createUpdateDisabledResult()); + } + } + catch (Exception e) { + resFut.onDone(MetadataUpdateResult.createUpdateDisabledResult(), e); } + if (ctx.clientDisconnected()) + onDisconnected(); + return resFut; } @@ -237,6 +245,8 @@ final class BinaryMetadataTransport { for (MetadataUpdateResultFuture fut : unlabeledFutures) fut.onDone(res); + unlabeledFutures.clear(); + for (MetadataUpdateResultFuture fut : syncMap.values()) fut.onDone(res);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java index 0416746..df64613 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateAcceptedMessage.java @@ -71,6 +71,11 @@ public class MetadataUpdateAcceptedMessage implements DiscoveryCustomMessage { } /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ @Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, DiscoCache discoCache) { throw new UnsupportedOperationException(); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java index f9bd660..84e32e1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java @@ -134,6 +134,11 @@ public final class MetadataUpdateProposedMessage implements DiscoveryCustomMessa } /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ @Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, DiscoCache discoCache) { throw new UnsupportedOperationException(); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java index 9b3c1ec..5bbbb31 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java @@ -598,6 +598,45 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology { } /** {@inheritDoc} */ + @Override public List<List<ClusterNode>> allOwners() { + lock.readLock().lock(); + + try { + int parts = partitions(); + + List<List<ClusterNode>> res = new ArrayList<>(parts); + + for (int i = 0; i < parts; i++) + res.add(new ArrayList<>()); + + List<ClusterNode> allNodes = discoCache.cacheGroupAffinityNodes(grpId); + + for (int i = 0; i < allNodes.size(); i++) { + ClusterNode node = allNodes.get(i); + + GridDhtPartitionMap nodeParts = node2part.get(node.id()); + + if (nodeParts != null) { + for (Map.Entry<Integer, GridDhtPartitionState> e : nodeParts.map().entrySet()) { + if (e.getValue() == OWNING) { + int part = e.getKey(); + + List<ClusterNode> owners = res.get(part); + + owners.add(node); + } + } + } + } + + return res; + } + finally { + lock.readLock().unlock(); + } + } + + /** {@inheritDoc} */ @Override public List<ClusterNode> moving(int p) { return nodes(p, AffinityTopologyVersion.NONE, MOVING, null); } http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java index ba55543..ea99f5d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java @@ -970,6 +970,10 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap try { ctx.io().send(nodeId, res, ctx.ioPolicy()); } + catch (ClusterTopologyCheckedException e) { + if (log.isDebugEnabled()) + log.debug("Failed to send get response to node, node failed: " + nodeId); + } catch (IgniteCheckedException e) { U.error(log, "Failed to send get response to node (is node still alive?) [nodeId=" + nodeId + ",req=" + req + ", res=" + res + ']', e); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java index 13564c2..7f900cb 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java @@ -237,6 +237,12 @@ public interface GridDhtPartitionTopology { public List<ClusterNode> owners(int p); /** + * @return List indexed by partition number, each list element is collection of all nodes who + * owns corresponding partition. + */ + public List<List<ClusterNode>> allOwners(); + + /** * @param p Partition ID. * @param topVer Topology version. * @return Collection of all nodes who {@code own} this partition. http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java index 528f0a6..538c57e 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java @@ -1217,6 +1217,45 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology { } /** {@inheritDoc} */ + @Override public List<List<ClusterNode>> allOwners() { + lock.readLock().lock(); + + try { + int parts = partitions(); + + List<List<ClusterNode>> res = new ArrayList<>(parts); + + for (int i = 0; i < parts; i++) + res.add(new ArrayList<>()); + + List<ClusterNode> allNodes = discoCache.cacheGroupAffinityNodes(grp.groupId()); + + for (int i = 0; i < allNodes.size(); i++) { + ClusterNode node = allNodes.get(i); + + GridDhtPartitionMap nodeParts = node2part.get(node.id()); + + if (nodeParts != null) { + for (Map.Entry<Integer, GridDhtPartitionState> e : nodeParts.map().entrySet()) { + if (e.getValue() == OWNING) { + int part = e.getKey(); + + List<ClusterNode> owners = res.get(part); + + owners.add(node); + } + } + } + } + + return res; + } + finally { + lock.readLock().unlock(); + } + } + + /** {@inheritDoc} */ @Override public List<ClusterNode> moving(int p) { if (!grp.rebalanceEnabled()) return ownersAndMoving(p, AffinityTopologyVersion.NONE); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java index 8da91a8..cbb4985 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java @@ -1506,12 +1506,16 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte } catch (ClusterTopologyCheckedException ignore) { if (log.isDebugEnabled()) - log.debug("Oldest node left during partition exchange [nodeId=" + oldestNode.id() + + log.debug("Coordinator left during partition exchange [nodeId=" + oldestNode.id() + ", exchId=" + exchId + ']'); } catch (IgniteCheckedException e) { - U.error(log, "Failed to send local partitions to oldest node (will retry after timeout) [oldestNodeId=" + - oldestNode.id() + ", exchId=" + exchId + ']', e); + if (reconnectOnError(e)) + onDone(new IgniteNeedReconnectException(cctx.localNode(), e)); + else { + U.error(log, "Failed to send local partitions to coordinator [crd=" + oldestNode.id() + + ", exchId=" + exchId + ']', e); + } } } @@ -3369,9 +3373,13 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte } if (allReceived) { - awaitSingleMapUpdates(); + cctx.kernalContext().getSystemExecutorService().submit(new Runnable() { + @Override public void run() { + awaitSingleMapUpdates(); - onAllReceived(null); + onAllReceived(null); + } + }); } } else { @@ -3399,7 +3407,13 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte ", newCrd=" + crd0.id() + ']'); } - sendPartitions(crd0); + final ClusterNode newCrd = crd0; + + cctx.kernalContext().getSystemExecutorService().submit(new Runnable() { + @Override public void run() { + sendPartitions(newCrd); + } + }); } } } http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateFinishMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateFinishMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateFinishMessage.java index d7dfa16..bbbd999 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateFinishMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateFinishMessage.java @@ -94,6 +94,11 @@ public class ChangeGlobalStateFinishMessage implements DiscoveryCustomMessage { } /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ @Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, DiscoCache discoCache) { throw new UnsupportedOperationException(); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java index 50fc022..81855fc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ChangeGlobalStateMessage.java @@ -131,6 +131,11 @@ public class ChangeGlobalStateMessage implements DiscoveryCustomMessage { } /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, DiscoCache discoCache) { return mgr.createDiscoCacheOnCacheChange(topVer, discoCache); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterMetricsUpdateMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterMetricsUpdateMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterMetricsUpdateMessage.java new file mode 100644 index 0000000..5c3044b --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterMetricsUpdateMessage.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cluster; + +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.internal.GridDirectMap; +import org.apache.ignite.internal.managers.communication.GridIoMessageFactory; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.jetbrains.annotations.Nullable; + +/** + * + */ +public class ClusterMetricsUpdateMessage implements Message { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + private byte[] nodeMetrics; + + /** */ + @GridDirectMap(keyType = UUID.class, valueType = byte[].class) + private Map<UUID, byte[]> allNodesMetrics; + + /** + * Required by {@link GridIoMessageFactory}. + */ + public ClusterMetricsUpdateMessage() { + // No-op. + } + + /** + * @param nodeMetrics Node metrics. + */ + ClusterMetricsUpdateMessage(byte[] nodeMetrics) { + this.nodeMetrics = nodeMetrics; + } + + /** + * @param allNodesMetrics All nodes metrcis. + */ + ClusterMetricsUpdateMessage(Map<UUID, byte[]> allNodesMetrics) { + this.allNodesMetrics = allNodesMetrics; + } + + /** + * @return Node metrics. + */ + @Nullable byte[] nodeMetrics() { + return nodeMetrics; + } + + /** + * @return All nodes metrics. + */ + @Nullable Map<UUID, byte[]> allNodesMetrics() { + return allNodesMetrics; + } + + /** {@inheritDoc} */ + @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) { + writer.setBuffer(buf); + + if (!writer.isHeaderWritten()) { + if (!writer.writeHeader(directType(), fieldsCount())) + return false; + + writer.onHeaderWritten(); + } + + switch (writer.state()) { + case 0: + if (!writer.writeMap("allNodesMetrics", allNodesMetrics, MessageCollectionItemType.UUID, MessageCollectionItemType.BYTE_ARR)) + return false; + + writer.incrementState(); + + case 1: + if (!writer.writeByteArray("nodeMetrics", nodeMetrics)) + return false; + + writer.incrementState(); + + } + + return true; + } + + /** {@inheritDoc} */ + @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) { + reader.setBuffer(buf); + + if (!reader.beforeMessageRead()) + return false; + + switch (reader.state()) { + case 0: + allNodesMetrics = reader.readMap("allNodesMetrics", MessageCollectionItemType.UUID, MessageCollectionItemType.BYTE_ARR, false); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + case 1: + nodeMetrics = reader.readByteArray("nodeMetrics"); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + } + + return reader.afterMessageRead(ClusterMetricsUpdateMessage.class); + } + + /** {@inheritDoc} */ + @Override public short directType() { + return 133; + } + + /** {@inheritDoc} */ + @Override public byte fieldsCount() { + return 2; + } + + /** {@inheritDoc} */ + @Override public void onAckReceived() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ClusterMetricsUpdateMessage.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterNodeMetrics.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterNodeMetrics.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterNodeMetrics.java new file mode 100644 index 0000000..22a385f --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterNodeMetrics.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cluster; + +import java.io.Serializable; +import java.util.Collections; +import java.util.Map; +import org.apache.ignite.cache.CacheMetrics; +import org.apache.ignite.cluster.ClusterMetrics; +import org.apache.ignite.internal.ClusterMetricsSnapshot; + +/** + * + */ +class ClusterNodeMetrics implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + private final byte[] metrics; + + /** */ + private final Map<Integer, CacheMetrics> cacheMetrics; + + /** + * @param metrics Metrics. + * @param cacheMetrics Cache metrics. + */ + ClusterNodeMetrics(ClusterMetrics metrics, Map<Integer, CacheMetrics> cacheMetrics) { + this.metrics = ClusterMetricsSnapshot.serialize(metrics); + this.cacheMetrics = cacheMetrics; + } + + /** + * @return Metrics. + */ + byte[] metrics() { + return metrics; + } + + /** + * @return Cache metrics. + */ + Map<Integer, CacheMetrics> cacheMetrics() { + return cacheMetrics != null ? cacheMetrics : Collections.<Integer, CacheMetrics>emptyMap(); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java index 5f2c66c..8796302 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterProcessor.java @@ -33,6 +33,8 @@ import org.apache.ignite.IgniteSystemProperties; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.events.DiscoveryEvent; import org.apache.ignite.events.Event; +import org.apache.ignite.internal.ClusterMetricsSnapshot; +import org.apache.ignite.internal.GridDirectMap; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.IgniteDiagnosticInfo; import org.apache.ignite.internal.IgniteDiagnosticMessage; @@ -42,21 +44,29 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; import org.apache.ignite.internal.cluster.IgniteClusterImpl; import org.apache.ignite.internal.managers.communication.GridIoPolicy; import org.apache.ignite.internal.managers.communication.GridMessageListener; +import org.apache.ignite.internal.managers.discovery.DiscoCache; +import org.apache.ignite.internal.managers.discovery.IgniteClusterNode; import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener; import org.apache.ignite.internal.processors.GridProcessorAdapter; +import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.internal.processors.timeout.GridTimeoutObject; import org.apache.ignite.internal.util.GridTimerTask; import org.apache.ignite.internal.util.future.GridFinishedFuture; import org.apache.ignite.internal.util.future.GridFutureAdapter; import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl; import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.typedef.CI1; +import org.apache.ignite.internal.util.typedef.internal.LT; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteClosure; import org.apache.ignite.lang.IgniteFuture; +import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.marshaller.jdk.JdkMarshaller; import org.apache.ignite.spi.discovery.DiscoveryDataBag; import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData; +import org.apache.ignite.spi.discovery.DiscoveryMetricsProvider; +import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; import org.jetbrains.annotations.Nullable; import static org.apache.ignite.IgniteSystemProperties.IGNITE_DIAGNOSTIC_ENABLED; @@ -66,6 +76,7 @@ import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.CLUSTER_PROC; import static org.apache.ignite.internal.GridTopic.TOPIC_INTERNAL_DIAGNOSTIC; +import static org.apache.ignite.internal.GridTopic.TOPIC_METRICS; import static org.apache.ignite.internal.IgniteVersionUtils.VER_STR; /** @@ -102,6 +113,18 @@ public class ClusterProcessor extends GridProcessorAdapter { /** */ private final AtomicLong diagFutId = new AtomicLong(); + /** */ + private final Map<UUID, byte[]> allNodesMetrics = new ConcurrentHashMap<>(); + + /** */ + private final JdkMarshaller marsh = new JdkMarshaller(); + + /** */ + private DiscoveryMetricsProvider metricsProvider; + + /** */ + private boolean sndMetrics; + /** * @param ctx Kernal context. */ @@ -111,6 +134,8 @@ public class ClusterProcessor extends GridProcessorAdapter { notifyEnabled.set(IgniteSystemProperties.getBoolean(IGNITE_UPDATE_NOTIFIER, true)); cluster = new IgniteClusterImpl(ctx); + + sndMetrics = !(ctx.config().getDiscoverySpi() instanceof TcpDiscoverySpi); } /** @@ -120,33 +145,31 @@ public class ClusterProcessor extends GridProcessorAdapter { return getBoolean(IGNITE_DIAGNOSTIC_ENABLED, true); } - /** */ - private final JdkMarshaller marsh = new JdkMarshaller(); - /** * @throws IgniteCheckedException If failed. */ public void initDiagnosticListeners() throws IgniteCheckedException { ctx.event().addLocalEventListener(new GridLocalEventListener() { - @Override public void onEvent(Event evt) { - assert evt instanceof DiscoveryEvent; - assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT; + @Override public void onEvent(Event evt) { + assert evt instanceof DiscoveryEvent; + assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT; - DiscoveryEvent discoEvt = (DiscoveryEvent)evt; + DiscoveryEvent discoEvt = (DiscoveryEvent)evt; - UUID nodeId = discoEvt.eventNode().id(); + UUID nodeId = discoEvt.eventNode().id(); - ConcurrentHashMap<Long, InternalDiagnosticFuture> futs = diagnosticFutMap.get(); + ConcurrentHashMap<Long, InternalDiagnosticFuture> futs = diagnosticFutMap.get(); - if (futs != null) { - for (InternalDiagnosticFuture fut : futs.values()) { - if (fut.nodeId.equals(nodeId)) - fut.onDone(new IgniteDiagnosticInfo("Target node failed: " + nodeId)); - } + if (futs != null) { + for (InternalDiagnosticFuture fut : futs.values()) { + if (fut.nodeId.equals(nodeId)) + fut.onDone(new IgniteDiagnosticInfo("Target node failed: " + nodeId)); } } - }, - EVT_NODE_FAILED, EVT_NODE_LEFT); + + allNodesMetrics.remove(nodeId); + } + }, EVT_NODE_FAILED, EVT_NODE_LEFT); ctx.io().addMessageListener(TOPIC_INTERNAL_DIAGNOSTIC, new GridMessageListener() { @Override public void onMessage(UUID nodeId, Object msg, byte plc) { @@ -233,6 +256,17 @@ public class ClusterProcessor extends GridProcessorAdapter { U.warn(diagnosticLog, "Received unexpected message: " + msg); } }); + + if (sndMetrics) { + ctx.io().addMessageListener(TOPIC_METRICS, new GridMessageListener() { + @Override public void onMessage(UUID nodeId, Object msg, byte plc) { + if (msg instanceof ClusterMetricsUpdateMessage) + processMetricsUpdateMessage(nodeId, (ClusterMetricsUpdateMessage)msg); + else + U.warn(log, "Received unexpected message for TOPIC_METRICS: " + msg); + } + }); + } } /** @@ -296,7 +330,6 @@ public class ClusterProcessor extends GridProcessorAdapter { } } - /** * @param vals collection to seek through. */ @@ -334,6 +367,14 @@ public class ClusterProcessor extends GridProcessorAdapter { log.debug("Failed to create GridUpdateNotifier: " + e); } } + + if (sndMetrics) { + metricsProvider = ctx.discovery().createMetricsProvider(); + + long updateFreq = ctx.config().getMetricsUpdateFrequency(); + + ctx.timeout().addTimeoutObject(new MetricsUpdateTimeoutObject(updateFreq)); + } } /** {@inheritDoc} */ @@ -352,6 +393,133 @@ public class ClusterProcessor extends GridProcessorAdapter { } /** + * @param sndNodeId Sender node ID. + * @param msg Message. + */ + private void processMetricsUpdateMessage(UUID sndNodeId, ClusterMetricsUpdateMessage msg) { + byte[] nodeMetrics = msg.nodeMetrics(); + + if (nodeMetrics != null) { + assert msg.allNodesMetrics() == null; + + allNodesMetrics.put(sndNodeId, nodeMetrics); + + updateNodeMetrics(ctx.discovery().discoCache(), sndNodeId, nodeMetrics); + } + else { + Map<UUID, byte[]> allNodesMetrics = msg.allNodesMetrics(); + + assert allNodesMetrics != null; + + DiscoCache discoCache = ctx.discovery().discoCache(); + + for (Map.Entry<UUID, byte[]> e : allNodesMetrics.entrySet()) { + if (!ctx.localNodeId().equals(e.getKey())) + updateNodeMetrics(discoCache, e.getKey(), e.getValue()); + } + } + } + + /** + * @param discoCache Discovery data cache. + * @param nodeId Node ID. + * @param metricsBytes Marshalled metrics. + */ + private void updateNodeMetrics(DiscoCache discoCache, UUID nodeId, byte[] metricsBytes) { + ClusterNode node = discoCache.node(nodeId); + + if (node == null || !discoCache.alive(nodeId)) + return; + + try { + ClusterNodeMetrics metrics = U.unmarshalZip(ctx.config().getMarshaller(), metricsBytes, null); + + assert node instanceof IgniteClusterNode : node; + + IgniteClusterNode node0 = (IgniteClusterNode)node; + + node0.setMetrics(ClusterMetricsSnapshot.deserialize(metrics.metrics(), 0)); + node0.setCacheMetrics(metrics.cacheMetrics()); + + ctx.discovery().metricsUpdateEvent(discoCache, node0); + } + catch (IgniteCheckedException e) { + U.warn(log, "Failed to unmarshal node metrics: " + e); + } + } + + /** + * + */ + private void updateMetrics() { + if (ctx.isStopping() || ctx.clientDisconnected()) + return; + + ClusterNode oldest = ctx.discovery().oldestAliveServerNode(AffinityTopologyVersion.NONE); + + if (oldest == null) + return; + + if (ctx.localNodeId().equals(oldest.id())) { + IgniteClusterNode locNode = (IgniteClusterNode)ctx.discovery().localNode(); + + locNode.setMetrics(metricsProvider.metrics()); + locNode.setCacheMetrics(metricsProvider.cacheMetrics()); + + ClusterNodeMetrics metrics = new ClusterNodeMetrics(locNode.metrics(), locNode.cacheMetrics()); + + try { + byte[] metricsBytes = U.zip(U.marshal(ctx.config().getMarshaller(), metrics)); + + allNodesMetrics.put(ctx.localNodeId(), metricsBytes); + } + catch (IgniteCheckedException e) { + U.warn(log, "Failed to marshal local node metrics: " + e, e); + } + + ctx.discovery().metricsUpdateEvent(ctx.discovery().discoCache(), locNode); + + Collection<ClusterNode> allNodes = ctx.discovery().allNodes(); + + ClusterMetricsUpdateMessage msg = new ClusterMetricsUpdateMessage(new HashMap<>(allNodesMetrics)); + + for (ClusterNode node : allNodes) { + if (ctx.localNodeId().equals(node.id()) || !ctx.discovery().alive(node.id())) + continue; + + try { + ctx.io().sendToGridTopic(node, TOPIC_METRICS, msg, GridIoPolicy.SYSTEM_POOL); + } + catch (ClusterTopologyCheckedException e) { + if (log.isDebugEnabled()) + log.debug("Failed to send metrics update, node failed: " + e); + } + catch (IgniteCheckedException e) { + U.warn(log, "Failed to send metrics update: " + e, e); + } + } + } + else { + ClusterNodeMetrics metrics = new ClusterNodeMetrics(metricsProvider.metrics(), metricsProvider.cacheMetrics()); + + try { + byte[] metricsBytes = U.zip(U.marshal(ctx.config().getMarshaller(), metrics)); + + ClusterMetricsUpdateMessage msg = new ClusterMetricsUpdateMessage(metricsBytes); + + ctx.io().sendToGridTopic(oldest, TOPIC_METRICS, msg, GridIoPolicy.SYSTEM_POOL); + } + catch (ClusterTopologyCheckedException e) { + if (log.isDebugEnabled()) + log.debug("Failed to send metrics update to oldest, node failed: " + e); + } + catch (IgniteCheckedException e) { + LT.warn(log, e, "Failed to send metrics update to oldest: " + e, false, false); + } + } + } + + /** * Disables update notifier. */ public void disableUpdateNotifier() { @@ -571,4 +739,51 @@ public class ClusterProcessor extends GridProcessorAdapter { return S.toString(InternalDiagnosticFuture.class, this); } } + + /** + * + */ + private class MetricsUpdateTimeoutObject implements GridTimeoutObject, Runnable { + /** */ + private final IgniteUuid id = IgniteUuid.randomUuid(); + + /** */ + private long endTime; + + /** */ + private final long timeout; + + /** + * @param timeout Timeout. + */ + MetricsUpdateTimeoutObject(long timeout) { + this.timeout = timeout; + + endTime = U.currentTimeMillis() + timeout; + } + + /** {@inheritDoc} */ + @Override public IgniteUuid timeoutId() { + return id; + } + + /** {@inheritDoc} */ + @Override public long endTime() { + return endTime; + } + + /** {@inheritDoc} */ + @Override public void run() { + updateMetrics(); + + endTime = U.currentTimeMillis() + timeout; + + ctx.timeout().addTimeoutObject(this); + } + + /** {@inheritDoc} */ + @Override public void onTimeout() { + ctx.getSystemExecutorService().execute(this); + } + } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/AbstractContinuousMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/AbstractContinuousMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/AbstractContinuousMessage.java index e9754d1..928c619 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/AbstractContinuousMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/AbstractContinuousMessage.java @@ -63,6 +63,11 @@ public abstract class AbstractContinuousMessage implements DiscoveryCustomMessag } /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ @Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, DiscoCache discoCache) { throw new UnsupportedOperationException(); http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java new file mode 100644 index 0000000..fc0f181 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.continuous; + +import java.io.Serializable; +import java.util.UUID; +import org.apache.ignite.internal.util.typedef.internal.S; + +/** + * + */ +class ContinuousRoutineInfo implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + UUID srcNodeId; + + /** */ + final UUID routineId; + + /** */ + final byte[] hnd; + + /** */ + final byte[] nodeFilter; + + /** */ + final int bufSize; + + /** */ + final long interval; + + /** */ + final boolean autoUnsubscribe; + + /** */ + transient boolean disconnected; + + /** + * @param srcNodeId Source node ID. + * @param routineId Routine ID. + * @param hnd Marshalled handler. + * @param nodeFilter Marshalled node filter. + * @param bufSize Handler buffer size. + * @param interval Time interval. + * @param autoUnsubscribe Auto unsubscribe flag. + */ + ContinuousRoutineInfo( + UUID srcNodeId, + UUID routineId, + byte[] hnd, + byte[] nodeFilter, + int bufSize, + long interval, + boolean autoUnsubscribe) + { + this.srcNodeId = srcNodeId; + this.routineId = routineId; + this.hnd = hnd; + this.nodeFilter = nodeFilter; + this.bufSize = bufSize; + this.interval = interval; + this.autoUnsubscribe = autoUnsubscribe; + } + + /** + * @param srcNodeId Source node ID. + */ + void sourceNodeId(UUID srcNodeId) { + this.srcNodeId = srcNodeId; + } + + /** + * + */ + void onDisconnected() { + disconnected = true; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinuousRoutineInfo.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineStartResultMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineStartResultMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineStartResultMessage.java new file mode 100644 index 0000000..581ac60 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineStartResultMessage.java @@ -0,0 +1,206 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.continuous; + +import java.nio.ByteBuffer; +import java.util.UUID; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.jetbrains.annotations.Nullable; + +/** + * + */ +public class ContinuousRoutineStartResultMessage implements Message { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + private static final int ERROR_FLAG = 0x01; + + /** */ + private UUID routineId; + + /** */ + private byte[] errBytes; + + /** */ + private byte[] cntrsMapBytes; + + /** */ + private int flags; + + /** + * + */ + public ContinuousRoutineStartResultMessage() { + // No-op. + } + + /** + * @param routineId Routine ID. + * @param cntrsMapBytes Marshalled {@link CachePartitionPartialCountersMap}. + * @param errBytes Error bytes. + * @param err {@code True} if failed to start routine. + */ + ContinuousRoutineStartResultMessage(UUID routineId, byte[] cntrsMapBytes, byte[] errBytes, boolean err) { + this.routineId = routineId; + this.cntrsMapBytes = cntrsMapBytes; + this.errBytes = errBytes; + + if (err) + flags |= ERROR_FLAG; + } + + /** + * @return Marshalled {@link CachePartitionPartialCountersMap}. + */ + @Nullable byte[] countersMapBytes() { + return cntrsMapBytes; + } + + /** + * @return {@code True} if failed to start routine. + */ + boolean error() { + return (flags & ERROR_FLAG) != 0; + } + + /** + * @return Routine ID. + */ + UUID routineId() { + return routineId; + } + + /** + * @return Error bytes. + */ + @Nullable byte[] errorBytes() { + return errBytes; + } + + /** {@inheritDoc} */ + @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) { + writer.setBuffer(buf); + + if (!writer.isHeaderWritten()) { + if (!writer.writeHeader(directType(), fieldsCount())) + return false; + + writer.onHeaderWritten(); + } + + switch (writer.state()) { + case 0: + if (!writer.writeByteArray("cntrsMapBytes", cntrsMapBytes)) + return false; + + writer.incrementState(); + + case 1: + if (!writer.writeByteArray("errBytes", errBytes)) + return false; + + writer.incrementState(); + + case 2: + if (!writer.writeInt("flags", flags)) + return false; + + writer.incrementState(); + + case 3: + if (!writer.writeUuid("routineId", routineId)) + return false; + + writer.incrementState(); + + } + + return true; + } + + /** {@inheritDoc} */ + @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) { + reader.setBuffer(buf); + + if (!reader.beforeMessageRead()) + return false; + + switch (reader.state()) { + case 0: + cntrsMapBytes = reader.readByteArray("cntrsMapBytes"); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + case 1: + errBytes = reader.readByteArray("errBytes"); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + case 2: + flags = reader.readInt("flags"); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + case 3: + routineId = reader.readUuid("routineId"); + + if (!reader.isLastRead()) + return false; + + reader.incrementState(); + + } + + return reader.afterMessageRead(ContinuousRoutineStartResultMessage.class); + } + + /** {@inheritDoc} */ + @Override public short directType() { + return 134; + } + + /** {@inheritDoc} */ + @Override public byte fieldsCount() { + return 4; + } + + /** {@inheritDoc} */ + @Override public void onAckReceived() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinuousRoutineStartResultMessage.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesCommonDiscoveryData.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesCommonDiscoveryData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesCommonDiscoveryData.java new file mode 100644 index 0000000..d29de89 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesCommonDiscoveryData.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.continuous; + +import java.io.Serializable; +import java.util.List; +import org.apache.ignite.internal.util.typedef.internal.S; + +/** + * + */ +public class ContinuousRoutinesCommonDiscoveryData implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + final List<ContinuousRoutineInfo> startedRoutines; + + /** + * @param startedRoutines Routines started in cluster. + */ + ContinuousRoutinesCommonDiscoveryData(List<ContinuousRoutineInfo> startedRoutines) { + this.startedRoutines = startedRoutines; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinuousRoutinesCommonDiscoveryData.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesInfo.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesInfo.java new file mode 100644 index 0000000..ad24ff1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesInfo.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.continuous; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.spi.discovery.DiscoveryDataBag; + +import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.CONTINUOUS_PROC; + +/** + * + */ +class ContinuousRoutinesInfo { + /** */ + private final Map<UUID, ContinuousRoutineInfo> startedRoutines = new HashMap<>(); + + /** + * @param dataBag Discovery data bag. + */ + void collectGridNodeData(DiscoveryDataBag dataBag) { + synchronized (startedRoutines) { + if (!dataBag.commonDataCollectedFor(CONTINUOUS_PROC.ordinal())) + dataBag.addGridCommonData(CONTINUOUS_PROC.ordinal(), + new ContinuousRoutinesCommonDiscoveryData(new ArrayList<>(startedRoutines.values()))); + } + } + + /** + * @param dataBag Discovery data bag. + */ + void collectJoiningNodeData(DiscoveryDataBag dataBag) { + synchronized (startedRoutines) { + for (ContinuousRoutineInfo info : startedRoutines.values()) { + if (info.disconnected) + info.sourceNodeId(dataBag.joiningNodeId()); + } + + dataBag.addJoiningNodeData(CONTINUOUS_PROC.ordinal(), + new ContinuousRoutinesJoiningNodeDiscoveryData(new ArrayList<>(startedRoutines.values()))); + } + } + + /** + * @param info Routine info. + */ + void addRoutineInfo(ContinuousRoutineInfo info) { + synchronized (startedRoutines) { + startedRoutines.put(info.routineId, info); + } + } + + /** + * @param routineId Routine ID. + * @return {@code True} if routine exists. + */ + boolean routineExists(UUID routineId) { + synchronized (startedRoutines) { + return startedRoutines.containsKey(routineId); + } + } + + /** + * @param routineId Routine ID. + */ + void removeRoutine(UUID routineId) { + synchronized (startedRoutines) { + startedRoutines.remove(routineId); + } + } + + /** + * @param locRoutines Routines IDs which can survive reconnect. + */ + void onClientDisconnected(Collection<UUID> locRoutines) { + synchronized (startedRoutines) { + for (Iterator<Map.Entry<UUID, ContinuousRoutineInfo>> it = startedRoutines.entrySet().iterator(); it.hasNext();) { + Map.Entry<UUID, ContinuousRoutineInfo> e = it.next(); + + ContinuousRoutineInfo info = e.getValue(); + + if (!locRoutines.contains(info.routineId)) + it.remove(); + else + info.onDisconnected(); + } + } + } + + /** + * Removes all routines with autoUnsubscribe=false started by given node. + * + * @param nodeId Node ID. + */ + void onNodeFail(UUID nodeId) { + synchronized (startedRoutines) { + for (Iterator<Map.Entry<UUID, ContinuousRoutineInfo>> it = startedRoutines.entrySet().iterator(); it.hasNext();) { + Map.Entry<UUID, ContinuousRoutineInfo> e = it.next(); + + ContinuousRoutineInfo info = e.getValue(); + + if (info.autoUnsubscribe && info.srcNodeId.equals(nodeId)) + it.remove(); + } + } + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinuousRoutinesInfo.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/a64b941d/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java new file mode 100644 index 0000000..9be6ef8 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.continuous; + +import java.io.Serializable; +import java.util.List; +import org.apache.ignite.internal.util.typedef.internal.S; + +/** + * + */ +public class ContinuousRoutinesJoiningNodeDiscoveryData implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + final List<ContinuousRoutineInfo> startedRoutines; + + /** + * @param startedRoutines Routines registered on nodes, to be started in cluster. + */ + ContinuousRoutinesJoiningNodeDiscoveryData(List<ContinuousRoutineInfo> startedRoutines) { + this.startedRoutines = startedRoutines; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinuousRoutinesJoiningNodeDiscoveryData.class, this); + } +}
