Rebalance stats

examples/config/example-rebalance.xml

@@ -0,0 +1,147 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+  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.
+-->
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans
+http://www.springframework.org/schema/beans/spring-beans.xsd">
+    <bean id="walArchivePath" class="java.lang.String">
+        <constructor-arg index="0" value="/sdb/ssd/mmuzaf/apache-ignite/work/walArchive" />
+    </bean>
+
+    <bean id="walStorePath" class="java.lang.String">
+        <constructor-arg index="0" value="/sdb/ssd/mmuzaf/apache-ignite/work/walStore" />
+    </bean>
+
+    <bean id="igniteConfiguration" class="org.apache.ignite.configuration.IgniteConfiguration">
+        <property name="failureDetectionTimeout" value="180000"/>
+        <property name="metricsLogFrequency" value="30000"/>
+        <property name="publicThreadPoolSize" value="256"/>
+        <property name="systemThreadPoolSize" value="256"/>
+	<property name="rebalanceThreadPoolSize" value="1"/>
+
+        <property name="peerClassLoadingEnabled" value="true"/>
+
+        <property name="binaryConfiguration">
+            <ref bean="binaryConfiguration"/>
+        </property>
+
+        <property name="communicationSpi">
+            <ref bean="tcpCommunicationSpi"/>
+        </property>
+
+        <property name="discoverySpi">
+            <ref bean="discoverySpi"/>
+        </property>
+
+        <property name="connectorConfiguration">
+            <ref bean="connectorConfiguration"/>
+        </property>
+
+        <property name="memoryConfiguration">
+            <ref bean="memoryConfiguration"/>
+        </property>
+
+        <property name="transactionConfiguration">
+            <ref bean="transactionConfiguration"/>
+        </property>
+
+        <property name="atomicConfiguration">
+            <ref bean="atomicConfiguration"/>
+        </property>
+
+        <property name="persistentStoreConfiguration">
+            <ref bean="persistentStoreConfiguration"/>
+        </property>
+    </bean>
+
+    <bean id="tcpCommunicationSpi" class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi">
+        <property name="sharedMemoryPort" value="-1"/>
+        <property name="connectTimeout" value="20000"/>
+        <property name="maxConnectTimeout" value="60000"/>
+        <property name="reconnectCount" value="2"/>
+        <property name="idleConnectionTimeout" value="600000"/>
+        <property name="socketWriteTimeout" value="30000"/>
+        <property name="selectorsCount" value="18"/>
+        <property name="connectionsPerNode" value="8"/>
+        <property name="messageQueueLimit" value="0"/>
+        <property name="localPort" value="48100"/>
+    </bean>
+
+    <bean id="discoverySpi" class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
+        <property name="localPort" value="48501" />
+        <property name="ipFinder">
+            <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
+                <property name="multicastPort" value="48601" />
+                <property name="addresses">
+                    <list>
+                        <value>10.116.172.13:48601..48609</value>
+			            <value>10.116.172.15:48601..48609</value>
+                    </list>
+                </property>
+            </bean>
+        </property>
+    </bean>
+
+    <bean id="binaryConfiguration" class="org.apache.ignite.configuration.BinaryConfiguration">
+        <property name="compactFooter" value="true"/>
+    </bean>
+
+    <bean id="connectorConfiguration" class="org.apache.ignite.configuration.ConnectorConfiguration">
+        <property name="port" value="11311" />
+        <property name="idleTimeout" value="180000"/>
+    </bean>
+
+    <bean id="memoryConfiguration" class="org.apache.ignite.configuration.MemoryConfiguration">
+        <property name="pageSize" value="#{4 * 1024}"/>
+        <property name="defaultMemoryPolicyName" value="dpl_mem_plc"/>
+        <property name="memoryPolicies">
+            <list>
+                <ref bean="memoryPolicyConfiguration"/>
+            </list>
+        </property>
+    </bean>
+
+    <bean id="memoryPolicyConfiguration" class="org.apache.ignite.configuration.MemoryPolicyConfiguration">
+        <property name="name" value="dpl_mem_plc"/>
+
+        <property name="initialSize" value="#{200 * 1024 * 1024 * 1024L}"/>
+        <property name="maxSize" value="#{600 * 1024 * 1024 * 1024L}"/>
+    </bean>
+
+    <bean id="persistentStoreConfiguration" class="org.apache.ignite.configuration.PersistentStoreConfiguration">
+        <property name="checkpointingFrequency" value="10000"/>
+        <property name="walMode" value="LOG_ONLY"/>
+        <property name="tlbSize" value="10485760"/>
+        <property name="checkpointingPageBufferSize" value="#{16 * 1024 * 1024 * 1024L}" />
+
+        <property name="walArchivePath" ref="walArchivePath" />
+        <property name="walStorePath" ref="walStorePath" />
+    </bean>
+
+    <bean id="transactionConfiguration" class="org.apache.ignite.configuration.TransactionConfiguration">
+        <property name="defaultTxIsolation" value="READ_COMMITTED"/>
+        <property name="defaultTxTimeout" value="300000"/>
+    </bean>
+
+    <bean id="atomicConfiguration" class="org.apache.ignite.configuration.AtomicConfiguration">
+        <property name="atomicSequenceReserveSize" value="50000"/>
+        <property name="cacheMode" value="REPLICATED"/>
+    </bean>
+</beans>

modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java

@@ -36,6 +36,7 @@
 import org.apache.ignite.internal.processors.cache.mvcc.MvccProcessor;
 import org.apache.ignite.internal.stat.IoStatisticsManager;
 import org.apache.ignite.internal.processors.compress.CompressionProcessor;
+import org.apache.ignite.internal.processors.diag.DiagnosticProcessor;
 import org.apache.ignite.internal.worker.WorkersRegistry;
 import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
 import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
@@ -467,6 +468,9 @@ public interface GridKernalContext extends Iterable<GridComponent> {
      */
     public FailureProcessor failure();
 
+    /** */
+    public DiagnosticProcessor diagnostic();
+
     /**
      * Print grid kernal memory stats (sizes of internal structures, etc.).
      *

modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java

@@ -48,6 +48,10 @@
 import org.apache.ignite.internal.managers.failover.GridFailoverManager;
 import org.apache.ignite.internal.managers.indexing.GridIndexingManager;
 import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccProcessor;
+import org.apache.ignite.internal.processors.compress.CompressionProcessor;
+import org.apache.ignite.internal.processors.diag.DiagnosticProcessor;
+import org.apache.ignite.internal.worker.WorkersRegistry;
 import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
 import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
 import org.apache.ignite.internal.processors.cache.CacheConflictResolutionManager;
@@ -423,6 +427,9 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
     /** Failure processor. */
     private FailureProcessor failureProc;
 
+    /** */
+    private DiagnosticProcessor diagProc;
+
     /** Recovery mode flag. Flag is set to {@code false} when discovery manager started. */
     private boolean recoveryMode = true;
 
@@ -585,7 +592,8 @@ else if (comp instanceof GridEncryptionManager)
          * Processors.
          * ==========
          */
-
+        else if (comp instanceof DiagnosticProcessor)
+            diagProc = (DiagnosticProcessor)comp;
         else if (comp instanceof FailureProcessor)
             failureProc = (FailureProcessor)comp;
         else if (comp instanceof GridTaskProcessor)
@@ -1196,6 +1204,11 @@ void disconnected(boolean disconnected) {
         return failureProc;
     }
 
+    /** {@inheritDoc} */
+    @Override public DiagnosticProcessor diagnostic() {
+        return diagProc;
+    }
+
     /** {@inheritDoc} */
     @Override public Thread.UncaughtExceptionHandler uncaughtExceptionHandler() {
         return hnd;

modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java

@@ -140,6 +140,7 @@
 import org.apache.ignite.internal.processors.continuous.GridContinuousProcessor;
 import org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor;
 import org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor;
+import org.apache.ignite.internal.processors.diag.DiagnosticProcessor;
 import org.apache.ignite.internal.processors.failure.FailureProcessor;
 import org.apache.ignite.internal.processors.hadoop.Hadoop;
 import org.apache.ignite.internal.processors.hadoop.HadoopProcessorAdapter;
@@ -963,6 +964,8 @@ public void start(
 
             startProcessor(new FailureProcessor(ctx));
 
+            startProcessor(new DiagnosticProcessor(ctx));
+
             startProcessor(new PoolProcessor(ctx));
 
             // Closure processor should be started before all others

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java

@@ -76,6 +76,7 @@
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersionEx;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersionedEntryEx;
+import org.apache.ignite.internal.processors.diag.DiagnosticTopics;
 import org.apache.ignite.internal.processors.dr.GridDrType;
 import org.apache.ignite.internal.processors.query.IgniteSQLException;
 import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheFilter;
@@ -121,6 +122,7 @@
 import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.RowData.NO_KEY;
 import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.DUPLICATE_KEY;
 import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.TRANSACTION_SERIALIZATION_ERROR;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.PRELOAD_UPDATED;
 import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
 
 /**
@@ -2978,6 +2980,8 @@ protected final void update(@Nullable CacheObject val, long expireTime, long ttl
         assert lock.isHeldByCurrentThread();
         assert ttl != CU.TTL_ZERO && ttl != CU.TTL_NOT_CHANGED && ttl >= 0 : ttl;
 
+        cctx.kernalContext().diagnostic().beginTrack(PRELOAD_UPDATED);
+
         boolean trackNear = addTracked && isNear() && cctx.config().isEagerTtl();
 
         long oldExpireTime = expireTimeExtras();
@@ -2994,6 +2998,8 @@ protected final void update(@Nullable CacheObject val, long expireTime, long ttl
 
         if (trackNear && expireTime != 0 && (expireTime != oldExpireTime || isStartVersion()))
             cctx.ttl().addTrackedEntry((GridNearCacheEntry)this);
+
+        cctx.kernalContext().diagnostic().endTrack(PRELOAD_UPDATED);
     }
 
     /**
@@ -3479,6 +3485,7 @@ else if (deletedUnlocked())
                             mvccVer == null ? MvccUtils.INITIAL_VERSION : mvccVer
                         )));
                     } else {
+                        cctx.kernalContext().diagnostic().beginTrack(DiagnosticTopics.PRELOAD_ON_WAL_LOG);
                         cctx.shared().wal().log(new DataRecord(new DataEntry(
                             cctx.cacheId(),
                             key,
@@ -3490,12 +3497,14 @@ else if (deletedUnlocked())
                             partition(),
                             updateCntr
                         )));
+                        cctx.kernalContext().diagnostic().endTrack(DiagnosticTopics.PRELOAD_ON_WAL_LOG);
                     }
                 }
 
                 drReplicate(drType, val, ver, topVer);
 
                 if (!skipQryNtf) {
+                    cctx.kernalContext().diagnostic().beginTrack(DiagnosticTopics.PRELOAD_ON_ENTRY_UPDATED);
                     cctx.continuousQueries().onEntryUpdated(
                         key,
                         val,
@@ -3507,6 +3516,7 @@ else if (deletedUnlocked())
                         updateCntr,
                         null,
                         topVer);
+                    cctx.kernalContext().diagnostic().endTrack(DiagnosticTopics.PRELOAD_ON_ENTRY_UPDATED);
                 }
 
                 onUpdateFinished(updateCntr);
@@ -4295,10 +4305,14 @@ protected boolean storeValue(
         @Nullable IgnitePredicate<CacheDataRow> predicate) throws IgniteCheckedException {
         assert lock.isHeldByCurrentThread();
 
+        cctx.kernalContext().diagnostic().beginTrack(DiagnosticTopics.PRELOAD_OFFHEAP_INVOKE);
+
         UpdateClosure closure = new UpdateClosure(this, val, ver, expireTime, predicate);
 
         cctx.offheap().invoke(cctx, key, localPartition(), closure);
 
+        cctx.kernalContext().diagnostic().endTrack(DiagnosticTopics.PRELOAD_OFFHEAP_INVOKE);
+
         return closure.treeOp != IgniteTree.OperationType.NOOP;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java

@@ -138,6 +138,7 @@
 import static org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion.NONE;
 import static org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture.nextDumpTimeout;
 import static org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader.DFLT_PRELOAD_RESEND_TIMEOUT;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.TOTAL;
 
 /**
  * Partition exchange manager.
@@ -3035,6 +3036,8 @@ else if (task instanceof ForceRebalanceExchangeTask) {
                         if (task instanceof ForceRebalanceExchangeTask)
                             forcedRebFut = ((ForceRebalanceExchangeTask)task).forcedRebalanceFuture();
 
+                        cctx.kernalContext().diagnostic().beginTrack(TOTAL);
+
                         for (Integer order : orderMap.descendingKeySet()) {
                             for (Integer grpId : orderMap.get(order)) {
                                 CacheGroupContext grp = cctx.cache().cacheGroup(grpId);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java

@@ -128,6 +128,11 @@
 import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.unexpectedStateException;
 import static org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager.EMPTY_CURSOR;
 import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.PRELOAD_PENDING_TREE_PUT;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.PRELOAD_PENDING_TREE_REMOVE;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.PRELOAD_TREE_ADD_ROW;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.PRELOAD_TREE_FINISH_UPDATE;
+import static org.apache.ignite.internal.processors.diag.DiagnosticTopics.PRELOAD_TREE_INVOKE;
 import static org.apache.ignite.internal.util.IgniteTree.OperationType.NOOP;
 import static org.apache.ignite.internal.util.IgniteTree.OperationType.PUT;
 
@@ -1660,15 +1665,20 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo
             try {
                 assert cctx.shared().database().checkpointLockIsHeldByThread();
 
+                ctx.kernalContext().diagnostic().beginTrack(PRELOAD_TREE_INVOKE);
+
                 dataTree.invoke(row, CacheDataRowAdapter.RowData.NO_KEY, c);
 
+                ctx.kernalContext().diagnostic().endTrack(PRELOAD_TREE_INVOKE);
+
                 switch (c.operationType()) {
                     case PUT: {
                         assert c.newRow() != null : c;
 
                         CacheDataRow oldRow = c.oldRow();
-
+                        ctx.kernalContext().diagnostic().beginTrack(PRELOAD_TREE_FINISH_UPDATE);
                         finishUpdate(cctx, c.newRow(), oldRow);
+                        ctx.kernalContext().diagnostic().endTrack(PRELOAD_TREE_FINISH_UPDATE);
 
                         break;
                     }
@@ -1705,6 +1715,8 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo
 
             DataRow dataRow = makeDataRow(key, val, ver, expireTime, cacheId);
 
+            ctx.kernalContext().diagnostic().beginTrack(PRELOAD_TREE_ADD_ROW);
+
             if (canUpdateOldRow(cctx, oldRow, dataRow) && rowStore.updateRow(oldRow.link(), dataRow, grp.statisticsHolderData()))
                 dataRow.link(oldRow.link());
             else {
@@ -1715,6 +1727,7 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo
 
                 rowStore.addRow(dataRow, grp.statisticsHolderData());
             }
+            ctx.kernalContext().diagnostic().endTrack(PRELOAD_TREE_ADD_ROW);
 
             assert dataRow.link() != 0 : dataRow;
 
@@ -2687,13 +2700,17 @@ private void updatePendingEntries(GridCacheContext cctx, CacheDataRow newRow, @N
             if (oldRow != null) {
                 assert oldRow.link() != 0 : oldRow;
 
-                if (pendingTree() != null && oldRow.expireTime() != 0)
+                if (pendingTree() != null && oldRow.expireTime() != 0) {
+                    cctx.kernalContext().diagnostic().beginTrack(PRELOAD_PENDING_TREE_REMOVE);
                     pendingTree().removex(new PendingRow(cacheId, oldRow.expireTime(), oldRow.link()));
+                    cctx.kernalContext().diagnostic().endTrack(PRELOAD_PENDING_TREE_REMOVE);
+                }
             }
 
             if (pendingTree() != null && expireTime != 0) {
+                cctx.kernalContext().diagnostic().beginTrack(PRELOAD_PENDING_TREE_PUT);
                 pendingTree().putx(new PendingRow(cacheId, expireTime, newRow.link()));
-
+                cctx.kernalContext().diagnostic().endTrack(PRELOAD_PENDING_TREE_PUT);
                 hasPendingEntries = true;
             }
         }