IGNITE-5658 Optimizations for data streamer

Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/aa81dd14
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/aa81dd14
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/aa81dd14

Branch: refs/heads/ignite-5872
Commit: aa81dd14a7d0e21d448acf2bbbb3a8a3d4eca0f8
Parents: 0e0c2c7
Author: Igor Seliverstov <gvvinbl...@gmail.com>
Authored: Wed Aug 9 19:29:39 2017 +0300
Committer: Yakov Zhdanov <yzhda...@gridgain.com>
Committed: Wed Aug 9 19:29:39 2017 +0300

----------------------------------------------------------------------
 .../org/apache/ignite/IgniteDataStreamer.java   |  26 +-
 .../apache/ignite/IgniteSystemProperties.java   |   8 +
 .../ignite/internal/GridKernalContext.java      |   2 +-
 .../ignite/internal/GridKernalContextImpl.java  |   6 +-
 .../apache/ignite/internal/IgniteKernal.java    |   5 +-
 .../ignite/internal/IgniteNodeAttributes.java   |   3 +
 .../org/apache/ignite/internal/IgnitionEx.java  |  23 +-
 .../ignite/internal/jdbc2/JdbcConnection.java   |   5 +-
 .../managers/communication/GridIoManager.java   |   6 +
 .../managers/communication/GridIoMessage.java   |   3 +
 .../cache/persistence/freelist/PagesList.java   |  67 +++--
 .../wal/reader/StandaloneGridKernalContext.java |   2 +-
 .../datastreamer/DataStreamerImpl.java          | 281 ++++++++++++-------
 .../datastreamer/DataStreamerRequest.java       |  59 +++-
 .../ignite/internal/util/StripedExecutor.java   |  89 +++++-
 .../org/apache/ignite/thread/IgniteThread.java  |  10 +
 .../DataStreamProcessorSelfTest.java            |  14 +-
 .../datastreamer/DataStreamerImplSelfTest.java  |   3 +-
 18 files changed, 427 insertions(+), 185 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java 
b/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
index e2473dc..b1f5851 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
@@ -20,6 +20,7 @@ package org.apache.ignite;
 import java.util.Collection;
 import java.util.Map;
 import javax.cache.CacheException;
+import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.lang.IgniteBiPredicate;
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.stream.StreamReceiver;
@@ -72,7 +73,7 @@ import org.jetbrains.annotations.Nullable;
  *      this setting limits maximum allowed number of parallel buffered stream 
messages that
  *      are being processed on remote nodes. If this number is exceeded, then
  *      {@link #addData(Object, Object)} method will block to control memory 
utilization.
- *      Default is defined by {@link #DFLT_MAX_PARALLEL_OPS} value.
+ *      Default is equal to CPU count on remote node multiply by {@link 
#DFLT_PARALLEL_OPS_MULTIPLIER}.
  *  </li>
  *  <li>
  *      {@link #autoFlushFrequency(long)} - automatic flush frequency in 
milliseconds. Essentially,
@@ -100,11 +101,23 @@ import org.jetbrains.annotations.Nullable;
  * </ul>
  */
 public interface IgniteDataStreamer<K, V> extends AutoCloseable {
-    /** Default max concurrent put operations count. */
+    /**
+     * Default max concurrent put operations count.
+     * @deprecated Is not used anymore.
+     */
+    @Deprecated
     public static final int DFLT_MAX_PARALLEL_OPS = 16;
 
-    /** Default per node buffer size. */
-    public static final int DFLT_PER_NODE_BUFFER_SIZE = 1024;
+    /**
+     * Default multiplier for data streamer pool size to get concurrent 
batches count for each remote node.
+     *
+     * @see IgniteConfiguration#getDataStreamerThreadPoolSize()
+     * @see #perNodeParallelOperations()
+     */
+    public static final int DFLT_PARALLEL_OPS_MULTIPLIER = 8;
+
+    /** Default operations batch size to sent to remote node for loading. */
+    public static final int DFLT_PER_NODE_BUFFER_SIZE = 512;
 
     /** Default timeout for streamer's operations. */
     public static final long DFLT_UNLIMIT_TIMEOUT = -1;
@@ -203,9 +216,11 @@ public interface IgniteDataStreamer<K, V> extends 
AutoCloseable {
      * <p>
      * This method should be called prior to {@link #addData(Object, Object)} 
call.
      * <p>
-     * If not provided, default value is {@link #DFLT_MAX_PARALLEL_OPS}.
+     * If not provided, default value is calculated as follows
+     * {@link #DFLT_PARALLEL_OPS_MULTIPLIER} * {@code 
DATA_STREAMER_POOL_SIZE_ON_REMOTE_NODE}.
      *
      * @param parallelOps Maximum number of parallel stream operations for a 
single node.
+     * @see IgniteConfiguration#getDataStreamerThreadPoolSize()
      */
     public void perNodeParallelOperations(int parallelOps);
 
@@ -450,5 +465,4 @@ public interface IgniteDataStreamer<K, V> extends 
AutoCloseable {
      */
     @Override public void close() throws CacheException, 
IgniteInterruptedException,
         IgniteDataStreamerTimeoutException;
-
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java 
b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index 2fa52b6..8af66c4 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -703,6 +703,14 @@ public final class IgniteSystemProperties {
     public static final String IGNITE_ENABLE_FORCIBLE_NODE_KILL = 
"IGNITE_ENABLE_FORCIBLE_NODE_KILL";
 
     /**
+     * Tasks stealing will be started if tasks queue size per data-streamer 
thread exceeds this threshold.
+     * <p>
+     * Default value is {@code 4}.
+     */
+    public static final String 
IGNITE_DATA_STREAMING_EXECUTOR_SERVICE_TASKS_STEALING_THRESHOLD =
+            "IGNITE_DATA_STREAMING_EXECUTOR_SERVICE_TASKS_STEALING_THRESHOLD";
+
+    /**
      * Enforces singleton.
      */
     private IgniteSystemProperties() {

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java 
b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index ef4001f..48a1e3e 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -532,7 +532,7 @@ public interface GridKernalContext extends 
Iterable<GridComponent> {
      *
      * @return Thread pool implementation to be used for data stream messages.
      */
-    public ExecutorService getDataStreamerExecutorService();
+    public StripedExecutor getDataStreamerExecutorService();
 
     /**
      * Should return an instance of fully configured thread pool to be used for

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 42a9b2c..89ead1a 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -317,7 +317,7 @@ public class GridKernalContextImpl implements 
GridKernalContext, Externalizable
 
     /** */
     @GridToStringExclude
-    private ExecutorService dataStreamExecSvc;
+    private StripedExecutor dataStreamExecSvc;
 
     /** */
     @GridToStringExclude
@@ -423,7 +423,7 @@ public class GridKernalContextImpl implements 
GridKernalContext, Externalizable
         ExecutorService p2pExecSvc,
         ExecutorService mgmtExecSvc,
         ExecutorService igfsExecSvc,
-        ExecutorService dataStreamExecSvc,
+        StripedExecutor dataStreamExecSvc,
         ExecutorService restExecSvc,
         ExecutorService affExecSvc,
         @Nullable ExecutorService idxExecSvc,
@@ -976,7 +976,7 @@ public class GridKernalContextImpl implements 
GridKernalContext, Externalizable
     }
 
     /** {@inheritDoc} */
-    @Override public ExecutorService getDataStreamerExecutorService() {
+    @Override public StripedExecutor getDataStreamerExecutorService() {
         return dataStreamExecSvc;
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java 
b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 756f2ab..f5111c3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -213,6 +213,7 @@ import static 
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_BUILD_VER;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_CLIENT_MODE;
 import static 
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_CONSISTENCY_CHECK_SKIPPED;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DAEMON;
+import static 
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_STREAMER_POOL_SIZE;
 import static 
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DEPLOYMENT_MODE;
 import static 
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IPS;
@@ -718,7 +719,7 @@ public class IgniteKernal implements IgniteEx, 
IgniteMXBean, Externalizable {
         ExecutorService p2pExecSvc,
         ExecutorService mgmtExecSvc,
         ExecutorService igfsExecSvc,
-        ExecutorService dataStreamExecSvc,
+        StripedExecutor dataStreamExecSvc,
         ExecutorService restExecSvc,
         ExecutorService affExecSvc,
         @Nullable ExecutorService idxExecSvc,
@@ -1427,6 +1428,8 @@ public class IgniteKernal implements IgniteEx, 
IgniteMXBean, Externalizable {
      */
     @SuppressWarnings({"SuspiciousMethodCalls", "unchecked", 
"TypeMayBeWeakened"})
     private void fillNodeAttributes(boolean notifyEnabled) throws 
IgniteCheckedException {
+        ctx.addNodeAttribute(ATTR_DATA_STREAMER_POOL_SIZE, 
configuration().getDataStreamerThreadPoolSize());
+
         final String[] incProps = cfg.getIncludeProperties();
 
         try {

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
index a45f991..fe23f92 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
@@ -177,6 +177,9 @@ public final class IgniteNodeAttributes {
     /** Ignite security compatibility mode. */
     public static final String ATTR_SECURITY_COMPATIBILITY_MODE = ATTR_PREFIX 
+ ".security.compatibility.enabled";
 
+    /** */
+    public static final String ATTR_DATA_STREAMER_POOL_SIZE = ATTR_PREFIX + 
".data.streamer.pool.size";
+
     /**
      * Enforces singleton.
      */

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java 
b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
index 23baeb3..07a5c43 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
@@ -1508,7 +1508,7 @@ public class IgnitionEx {
         private ThreadPoolExecutor igfsExecSvc;
 
         /** Data streamer executor service. */
-        private ThreadPoolExecutor dataStreamerExecSvc;
+        private StripedExecutor dataStreamerExecSvc;
 
         /** REST requests executor service. */
         private ThreadPoolExecutor restExecSvc;
@@ -1728,7 +1728,11 @@ public class IgnitionEx {
 
             validateThreadPoolSize(cfg.getStripedPoolSize(), "stripedPool");
 
-            stripedExecSvc = new StripedExecutor(cfg.getStripedPoolSize(), 
cfg.getIgniteInstanceName(), "sys", log);
+            stripedExecSvc = new StripedExecutor(
+                cfg.getStripedPoolSize(),
+                cfg.getIgniteInstanceName(),
+                "sys",
+                log);
 
             // Note that since we use 'LinkedBlockingQueue', number of
             // maximum threads has no effect.
@@ -1763,17 +1767,12 @@ public class IgnitionEx {
 
             p2pExecSvc.allowCoreThreadTimeOut(true);
 
-            // Note that we do not pre-start threads here as this pool may not 
be needed.
-            dataStreamerExecSvc = new IgniteThreadPoolExecutor(
-                "data-streamer",
-                cfg.getIgniteInstanceName(),
-                cfg.getDataStreamerThreadPoolSize(),
+            dataStreamerExecSvc = new StripedExecutor(
                 cfg.getDataStreamerThreadPoolSize(),
-                DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>(),
-                GridIoPolicy.DATA_STREAMER_POOL);
-
-            dataStreamerExecSvc.allowCoreThreadTimeOut(true);
+                cfg.getIgniteInstanceName(),
+                "data-streamer",
+                log,
+                true);
 
             // Note that we do not pre-start threads here as igfs pool may not 
be needed.
             validateThreadPoolSize(cfg.getIgfsThreadPoolSize(), "IGFS");

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcConnection.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcConnection.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcConnection.java
index 1bf51f2..6fb723e 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcConnection.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcConnection.java
@@ -185,8 +185,9 @@ public class JdbcConnection implements Connection {
         streamFlushTimeout = 
Long.parseLong(props.getProperty(PROP_STREAMING_FLUSH_FREQ, "0"));
         streamNodeBufSize = 
Integer.parseInt(props.getProperty(PROP_STREAMING_PER_NODE_BUF_SIZE,
             String.valueOf(IgniteDataStreamer.DFLT_PER_NODE_BUFFER_SIZE)));
-        streamNodeParOps = 
Integer.parseInt(props.getProperty(PROP_STREAMING_PER_NODE_PAR_OPS,
-            String.valueOf(IgniteDataStreamer.DFLT_MAX_PARALLEL_OPS)));
+        // If value is zero, server data-streamer pool size multiplied
+        // by IgniteDataStreamer.DFLT_PARALLEL_OPS_MULTIPLIER will be used
+        streamNodeParOps = 
Integer.parseInt(props.getProperty(PROP_STREAMING_PER_NODE_PAR_OPS, "0"));
 
         String nodeIdProp = props.getProperty(PROP_NODE_ID);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index bb36b26..2005032 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@ -1125,6 +1125,12 @@ public class GridIoManager extends 
GridManagerAdapter<CommunicationSpi<Serializa
             return;
         }
 
+        if (plc == GridIoPolicy.DATA_STREAMER_POOL && msg.partition() != 
GridIoMessage.STRIPE_DISABLED_PART) {
+            ctx.getDataStreamerExecutorService().execute(msg.partition(), c);
+
+            return;
+        }
+
         if (msg.topicOrdinal() == TOPIC_IO_TEST.ordinal()) {
             IgniteIoTestMessage msg0 = (IgniteIoTestMessage)msg.message();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
index dccd336..fe61aec 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
 import org.apache.ignite.internal.ExecutorAwareMessage;
 import org.apache.ignite.internal.GridDirectTransient;
 import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.plugin.extensions.communication.Message;
@@ -336,6 +337,8 @@ public class GridIoMessage implements Message {
     public int partition() {
         if (msg instanceof GridCacheMessage)
             return ((GridCacheMessage)msg).partition();
+        if (msg instanceof DataStreamerRequest)
+            return ((DataStreamerRequest)msg).partition();
         else
             return STRIPE_DISABLED_PART;
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
index 39a6865..6c355f6 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/PagesList.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.util.GridLongList;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.thread.IgniteThread;
 import org.jetbrains.annotations.Nullable;
 
 import static java.lang.Boolean.FALSE;
@@ -66,7 +67,7 @@ public abstract class PagesList extends DataStructure {
     /** */
     private static final int MAX_STRIPES_PER_BUCKET =
         
IgniteSystemProperties.getInteger("IGNITE_PAGES_LIST_STRIPES_PER_BUCKET",
-            Math.min(8, Runtime.getRuntime().availableProcessors() * 2));
+            Math.max(8, Runtime.getRuntime().availableProcessors()));
 
     /** */
     protected final AtomicLong[] bucketsSize;
@@ -507,6 +508,21 @@ public abstract class PagesList extends DataStructure {
      * @throws IgniteCheckedException If failed.
      */
     private Stripe getPageForPut(int bucket) throws IgniteCheckedException {
+        // Striped pool optimization.
+        int stripeIdx; IgniteThread igniteThread = IgniteThread.current();
+
+        if (igniteThread != null && (stripeIdx = igniteThread.stripe()) != -1) 
{
+            Stripe[] tails = getBucket(bucket);
+
+            while (tails == null || stripeIdx >= tails.length) {
+                addStripe(bucket, true);
+
+                tails = getBucket(bucket);
+            }
+
+            return tails[stripeIdx];
+        }
+
         Stripe[] tails = getBucket(bucket);
 
         if (tails == null)
@@ -607,12 +623,8 @@ public abstract class PagesList extends DataStructure {
             try {
                 long tailAddr = writeLockPage(tailId, tailPage, bucket, 
lockAttempt++); // Explicit check.
 
-                if (tailAddr == 0L) {
-                    if (isReuseBucket(bucket) && lockAttempt == 
TRY_LOCK_ATTEMPTS)
-                        addStripeForReuseBucket(bucket);
-
+                if (tailAddr == 0L)
                     continue;
-                }
 
                 assert PageIO.getPageId(tailAddr) == tailId : "pageId = " + 
PageIO.getPageId(tailAddr) + ", tailId = " + tailId;
                 assert PageIO.getType(tailAddr) == PageIO.T_PAGE_LIST_NODE;
@@ -912,13 +924,26 @@ public abstract class PagesList extends DataStructure {
      * @param bucket Bucket index.
      * @return Page for take.
      */
-    private Stripe getPageForTake(int bucket) {
+    private Stripe getPageForTake(int bucket) throws IgniteCheckedException {
         Stripe[] tails = getBucket(bucket);
 
         if (tails == null || bucketsSize[bucket].get() == 0)
             return null;
 
         int len = tails.length;
+
+        // Striped pool optimization.
+        int stripeIdx; IgniteThread igniteThread = IgniteThread.current();
+
+        if (igniteThread != null && (stripeIdx = igniteThread.stripe()) != -1) 
{
+            if (stripeIdx >= len)
+                return null;
+
+            Stripe stripe = tails[stripeIdx];
+
+            return stripe.empty ? null : stripe;
+        }
+
         int init = randomInt(len);
         int cur = init;
 
@@ -943,6 +968,12 @@ public abstract class PagesList extends DataStructure {
      */
     private long writeLockPage(long pageId, long page, int bucket, int 
lockAttempt)
         throws IgniteCheckedException {
+        // Striped pool optimization.
+        IgniteThread igniteThread = IgniteThread.current();
+
+        if (igniteThread != null && igniteThread.stripe() != -1)
+            return writeLock(pageId, page);
+
         long pageAddr = tryWriteLock(pageId, page);
 
         if (pageAddr != 0L)
@@ -952,8 +983,7 @@ public abstract class PagesList extends DataStructure {
             Stripe[] stripes = getBucket(bucket);
 
             if (stripes == null || stripes.length < MAX_STRIPES_PER_BUCKET) {
-                if (!isReuseBucket(bucket))
-                    addStripe(bucket, true);
+                addStripe(bucket, !isReuseBucket(bucket));
 
                 return 0L;
             }
@@ -963,19 +993,6 @@ public abstract class PagesList extends DataStructure {
     }
 
     /**
-     * @param bucket Bucket.
-     * @throws IgniteCheckedException If failed.
-     */
-    private void addStripeForReuseBucket(int bucket) throws 
IgniteCheckedException {
-        assert isReuseBucket(bucket);
-
-        Stripe[] stripes = getBucket(bucket);
-
-        if (stripes == null || stripes.length < MAX_STRIPES_PER_BUCKET)
-            addStripe(bucket, false);
-    }
-
-    /**
      * @param bucket Bucket index.
      * @param initIoVers Optional IO to initialize page.
      * @return Removed page ID.
@@ -994,12 +1011,8 @@ public abstract class PagesList extends DataStructure {
             try {
                 long tailAddr = writeLockPage(tailId, tailPage, bucket, 
lockAttempt++); // Explicit check.
 
-                if (tailAddr == 0L) {
-                    if (isReuseBucket(bucket) && lockAttempt == 
TRY_LOCK_ATTEMPTS)
-                        addStripeForReuseBucket(bucket);
-
+                if (tailAddr == 0L)
                     continue;
-                }
 
                 if (stripe.empty) {
                     // Another thread took the last page.

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
index 02b9352..8752eaa 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
@@ -421,7 +421,7 @@ public class StandaloneGridKernalContext implements 
GridKernalContext {
     }
 
     /** {@inheritDoc} */
-    @Override public ExecutorService getDataStreamerExecutorService() {
+    @Override public StripedExecutor getDataStreamerExecutorService() {
         return null;
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
index df51fac..eaced66 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.datastreamer;
 
+import java.lang.reflect.Array;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -60,8 +61,10 @@ import 
org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
 import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.IgniteNodeAttributes;
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
 import 
org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
 import org.apache.ignite.internal.managers.communication.GridMessageListener;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
 import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
@@ -152,7 +155,7 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
     private int bufSize = DFLT_PER_NODE_BUFFER_SIZE;
 
     /** */
-    private int parallelOps = DFLT_MAX_PARALLEL_OPS;
+    private int parallelOps;
 
     /** */
     private long timeout = DFLT_UNLIMIT_TIMEOUT;
@@ -791,6 +794,9 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
                             initPda = false;
                         }
 
+                        if (key.partition() == -1)
+                            key.partition(cctx.affinity().partition(key, 
false));
+
                         nodes = nodes(key, topVer, cctx);
                     }
                     catch (IgniteCheckedException e) {
@@ -932,10 +938,10 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
                         }
                     };
 
-                    final GridFutureAdapter<?> f;
+                    final List<GridFutureAdapter<?>> futs;
 
                     try {
-                        f = buf.update(entriesForNode, topVer, lsnr, remap);
+                        futs = buf.update(entriesForNode, topVer, lsnr, remap);
                     }
                     catch (IgniteInterruptedCheckedException e1) {
                         resFut.onDone(e1);
@@ -951,9 +957,13 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
                                 @Override public void run() {
                                     buf0.onNodeLeft();
 
-                                    if (f != null)
-                                        f.onDone(new 
ClusterTopologyCheckedException("Failed to wait for request completion " +
-                                            "(node has left): " + nodeId));
+                                    if (futs != null) {
+                                        Throwable ex = new 
ClusterTopologyCheckedException(
+                                                "Failed to wait for request 
completion (node has left): " + nodeId);
+
+                                        for (int i = 0; i < futs.size(); i++)
+                                            futs.get(i).onDone(ex);
+                                    }
                                 }
                             }, ctx.discovery().topologyVersion(), false);
                         }
@@ -1311,11 +1321,7 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
         private final Collection<IgniteInternalFuture<Object>> locFuts;
 
         /** Buffered entries. */
-        private List<DataStreamerEntry> entries;
-
-        /** */
-        @GridToStringExclude
-        private GridFutureAdapter<Object> curFut;
+        private final PerStripeBuffer[] stripes;
 
         /** Local node flag. */
         private final boolean isLocNode;
@@ -1329,16 +1335,17 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
         /** */
         private final Semaphore sem;
 
-        /** Batch topology. */
-        private AffinityTopologyVersion batchTopVer;
+        /** */
+        private final int perNodeParallelOps;
 
         /** Closure to signal on task finish. */
         @GridToStringExclude
-        private final IgniteInClosure<IgniteInternalFuture<Object>> signalC = 
new IgniteInClosure<IgniteInternalFuture<Object>>() {
-            @Override public void apply(IgniteInternalFuture<Object> t) {
-                signalTaskFinished(t);
-            }
-        };
+        private final IgniteInClosure<IgniteInternalFuture<Object>> signalC =
+            new IgniteInClosure<IgniteInternalFuture<Object>>() {
+                @Override public void apply(IgniteInternalFuture<Object> t) {
+                    signalTaskFinished(t);
+                }
+            };
 
         /**
          * @param node Node.
@@ -1354,24 +1361,19 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
             // Cache local node flag.
             isLocNode = node.equals(ctx.discovery().localNode());
 
-            entries = newEntries();
-            curFut = new GridFutureAdapter<>();
-            curFut.listen(signalC);
+            Integer attrStreamerPoolSize = 
node.attribute(IgniteNodeAttributes.ATTR_DATA_STREAMER_POOL_SIZE);
 
-            sem = new Semaphore(parallelOps);
-        }
+            int streamerPoolSize = attrStreamerPoolSize != null ? 
attrStreamerPoolSize : node.metrics().getTotalCpus();
 
-        /**
-         * @param remap Remapping flag.
-         */
-        private void renewBatch(boolean remap) {
-            entries = newEntries();
-            curFut = new GridFutureAdapter<>();
+            perNodeParallelOps = parallelOps != 0 ? parallelOps :
+                streamerPoolSize * 
IgniteDataStreamer.DFLT_PARALLEL_OPS_MULTIPLIER;
 
-            batchTopVer = null;
+            sem = new Semaphore(perNodeParallelOps);
 
-            if (!remap)
-                curFut.listen(signalC);
+            stripes = 
(PerStripeBuffer[])Array.newInstance(PerStripeBuffer.class, streamerPoolSize);
+
+            for (int i = 0; i < stripes.length; i++)
+                stripes[i] = new PerStripeBuffer(i, signalC);
         }
 
         /**
@@ -1382,61 +1384,69 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
          * @return Future for operation.
          * @throws IgniteInterruptedCheckedException If failed.
          */
-        @Nullable GridFutureAdapter<?> update(Iterable<DataStreamerEntry> 
newEntries,
+        @Nullable List<GridFutureAdapter<?>> update(
+            Iterable<DataStreamerEntry> newEntries,
             AffinityTopologyVersion topVer,
             IgniteInClosure<IgniteInternalFuture<?>> lsnr,
-            boolean remap) throws IgniteInterruptedCheckedException {
-            List<DataStreamerEntry> entries0 = null;
+            boolean remap
+        ) throws IgniteInterruptedCheckedException {
+            List<GridFutureAdapter<?>> res = null;
 
-            GridFutureAdapter<Object> curFut0;
+            for (DataStreamerEntry entry : newEntries) {
+                List<DataStreamerEntry> entries0 = null;
+                AffinityTopologyVersion curBatchTopVer;
+
+                // Init buffer.
+                int part = entry.getKey().partition();
 
-            AffinityTopologyVersion curBatchTopVer;
+                GridFutureAdapter<Object> curFut0;
+                PerStripeBuffer b = stripes[part % stripes.length];
 
-            synchronized (this) {
-                curFut0 = curFut;
+                synchronized (b) {
+                    curFut0 = b.curFut;
 
-                curFut0.listen(lsnr);
+                    // Listener should be added only once per whole entries 
collection.
+                    // Should we simplify the model and get rid of all futures?
+                    curFut0.listen(lsnr);
 
-                if (batchTopVer == null)
-                    batchTopVer = topVer;
+                    if (b.batchTopVer == null)
+                        b.batchTopVer = topVer;
 
-                curBatchTopVer = batchTopVer;
+                    curBatchTopVer = b.batchTopVer;
 
-                for (DataStreamerEntry entry : newEntries)
-                    entries.add(entry);
+                    b.entries.add(entry);
 
-                if (entries.size() >= bufSize) {
-                    entries0 = entries;
+                    if (b.entries.size() >= bufSize) {
+                        entries0 = b.entries;
 
-                    renewBatch(remap);
+                        b.renewBatch(remap);
+                    }
                 }
-            }
 
-            if (!allowOverwrite() && !topVer.equals(curBatchTopVer)) {
-                renewBatch(remap);
+                if (res == null)
+                    res = new ArrayList<>();
 
-                curFut0.onDone(null, new IgniteCheckedException("Topology 
changed during batch preparation." +
-                    "[batchTopVer=" + curBatchTopVer + ", topVer=" + topVer + 
"]"));
-            }
-            else if (entries0 != null) {
-                submit(entries0, curBatchTopVer, curFut0, remap);
-
-                if (cancelled)
-                    curFut0.onDone(new IgniteCheckedException("Data streamer 
has been cancelled: " +
-                        DataStreamerImpl.this));
-                else if (ctx.clientDisconnected())
-                    curFut0.onDone(new 
IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
-                        "Client node disconnected."));
-            }
+                res.add(curFut0);
 
-            return curFut0;
-        }
+                if (!allowOverwrite() && !topVer.equals(curBatchTopVer)) {
+                    b.renewBatch(remap);
 
-        /**
-         * @return Fresh collection with some space for outgrowth.
-         */
-        private List<DataStreamerEntry> newEntries() {
-            return new ArrayList<>((int)(bufSize * 1.2));
+                    curFut0.onDone(null, new IgniteCheckedException("Topology 
changed during batch preparation." +
+                        "[batchTopVer=" + curBatchTopVer + ", topVer=" + 
topVer + "]"));
+                }
+                else if (entries0 != null) {
+                    submit(entries0, curBatchTopVer, curFut0, remap, b.partId);
+
+                    if (cancelled)
+                        curFut0.onDone(new IgniteCheckedException("Data 
streamer has been cancelled: " +
+                            DataStreamerImpl.this));
+                    else if (ctx.clientDisconnected())
+                        curFut0.onDone(new 
IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
+                            "Client node disconnected."));
+                }
+            }
+
+            return res;
         }
 
         /**
@@ -1444,24 +1454,26 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
          * @throws IgniteInterruptedCheckedException If thread has been 
interrupted.
          */
         @Nullable IgniteInternalFuture<?> flush() throws 
IgniteInterruptedCheckedException {
-            List<DataStreamerEntry> entries0 = null;
-            GridFutureAdapter<Object> curFut0 = null;
-
             acquireRemapSemaphore();
 
-            synchronized (this) {
-                if (!entries.isEmpty()) {
-                    entries0 = entries;
-                    curFut0 = curFut;
+            for (PerStripeBuffer b : stripes) {
+                AffinityTopologyVersion batchTopVer = null;
+                List<DataStreamerEntry> entries0 = null;
+                GridFutureAdapter<Object> curFut0 = null;
+
+                synchronized (b) {
+                    if (!b.entries.isEmpty()) {
+                        entries0 = b.entries;
+                        curFut0 = b.curFut;
+                        batchTopVer = b.batchTopVer;
 
-                    entries = newEntries();
-                    curFut = new GridFutureAdapter<>();
-                    curFut.listen(signalC);
+                        b.renewBatch(false);
+                    }
                 }
-            }
 
-            if (entries0 != null)
-                submit(entries0, batchTopVer, curFut0, false);
+                if (entries0 != null)
+                    submit(entries0, batchTopVer, curFut0, false, b.partId);
+            }
 
             // Create compound future for this flush.
             GridCompoundFuture<Object, Object> res = null;
@@ -1601,13 +1613,16 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
          * @param topVer Topology version.
          * @param curFut Current future.
          * @param remap Remapping flag.
+         * @param partId Partition ID.
          * @throws IgniteInterruptedCheckedException If interrupted.
          */
-        private void submit(final Collection<DataStreamerEntry> entries,
+        private void submit(
+            final Collection<DataStreamerEntry> entries,
             @Nullable AffinityTopologyVersion topVer,
             final GridFutureAdapter<Object> curFut,
-            boolean remap)
-            throws IgniteInterruptedCheckedException {
+            boolean remap,
+            int partId
+        ) throws IgniteInterruptedCheckedException {
             assert entries != null;
             assert !entries.isEmpty();
             assert curFut != null;
@@ -1668,7 +1683,8 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
                             cache.context().deploy().onEnter();
                     }
                     catch (IgniteCheckedException e) {
-                        U.error(log, "Failed to deploy class (request will not 
be sent): " + jobPda0.deployClass(), e);
+                        U.error(log, "Failed to deploy class (request will not 
be sent): " +
+                            jobPda0.deployClass(), e);
 
                         return;
                     }
@@ -1701,7 +1717,9 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
                     dep != null ? dep.participants() : null,
                     dep != null ? dep.classLoaderId() : null,
                     dep == null,
-                    topVer);
+                    topVer,
+                    (rcvr == ISOLATED_UPDATER) ?
+                        partId : GridIoMessage.STRIPE_DISABLED_PART);
 
                 try {
                     ctx.io().sendToGridTopic(node, TOPIC_DATASTREAM, req, plc);
@@ -1750,11 +1768,13 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
             // Make sure to complete current future.
             GridFutureAdapter<Object> curFut0;
 
-            synchronized (this) {
-                curFut0 = curFut;
-            }
+            for (PerStripeBuffer b : stripes) {
+                synchronized (b) {
+                    curFut0 = b.curFut;
+                }
 
-            curFut0.onDone(e);
+                curFut0.onDone(e);
+            }
         }
 
         /**
@@ -1828,10 +1848,14 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
 
         /** {@inheritDoc} */
         @Override public String toString() {
-            int size;
+            int size = 0;
 
-            synchronized (this) {
-                size = entries.size();
+            for (int i = 0; i < stripes.length; i++) {
+                PerStripeBuffer b = stripes[i];
+
+                synchronized (b) {
+                    size += b.entries.size();
+                }
             }
 
             return S.toString(Buffer.class, this,
@@ -1920,8 +1944,10 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
         private static final long serialVersionUID = 0L;
 
         /** {@inheritDoc} */
-        @Override public void receive(IgniteCache<KeyCacheObject, CacheObject> 
cache,
-            Collection<Map.Entry<KeyCacheObject, CacheObject>> entries) {
+        @Override public void receive(
+            IgniteCache<KeyCacheObject, CacheObject> cache,
+            Collection<Map.Entry<KeyCacheObject, CacheObject>> entries
+        ) {
             IgniteCacheProxy<KeyCacheObject, CacheObject> proxy = 
(IgniteCacheProxy<KeyCacheObject, CacheObject>)cache;
 
             GridCacheAdapter<KeyCacheObject, CacheObject> internalCache = 
proxy.context().cache();
@@ -2080,4 +2106,63 @@ public class DataStreamerImpl<K, V> implements 
IgniteDataStreamer<K, V>, Delayed
             return S.toString(KeyCacheObjectWrapper.class, this);
         }
     }
+
+    /**
+     *
+     */
+    private class PerStripeBuffer {
+        /** */
+        private final int partId;
+
+        /** */
+        private List<DataStreamerEntry> entries;
+
+        /** */
+        private GridFutureAdapter<Object> curFut;
+
+        /** Batch topology. */
+        private AffinityTopologyVersion batchTopVer;
+
+        /** */
+        private final IgniteInClosure<? super IgniteInternalFuture<Object>> 
signalC;
+
+        /**
+         * @param partId Partition ID.
+         * @param c Signal closure.
+         */
+        public PerStripeBuffer(
+            int partId,
+            IgniteInClosure<? super IgniteInternalFuture<Object>> c
+        ) {
+            this.partId = partId;
+            signalC = c;
+
+            renewBatch(false);
+        }
+
+        /**
+         * @param remap Remap.
+         */
+        synchronized void renewBatch(boolean remap) {
+            entries = newEntries();
+            curFut = new GridFutureAdapter<>();
+
+            batchTopVer = null;
+
+            if (!remap)
+                curFut.listen(signalC);
+        }
+
+        /**
+         * @return Fresh collection with some space for outgrowth.
+         */
+        private List<DataStreamerEntry> newEntries() {
+            return new ArrayList<>((int)(bufSize * 1.2));
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(PerStripeBuffer.class, this, super.toString());
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
index b4cbf66..f70ee9c 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
@@ -90,6 +90,9 @@ public class DataStreamerRequest implements Message {
     /** Topology version. */
     private AffinityTopologyVersion topVer;
 
+    /** */
+    private int partId;
+
     /**
      * {@code Externalizable} support.
      */
@@ -113,8 +116,10 @@ public class DataStreamerRequest implements Message {
      * @param clsLdrId Class loader ID.
      * @param forceLocDep Force local deployment.
      * @param topVer Topology version.
+     * @param partId Partition ID.
      */
-    public DataStreamerRequest(long reqId,
+    public DataStreamerRequest(
+        long reqId,
         byte[] resTopicBytes,
         @Nullable String cacheName,
         byte[] updaterBytes,
@@ -128,7 +133,9 @@ public class DataStreamerRequest implements Message {
         Map<UUID, IgniteUuid> ldrParticipants,
         IgniteUuid clsLdrId,
         boolean forceLocDep,
-        @NotNull AffinityTopologyVersion topVer) {
+        @NotNull AffinityTopologyVersion topVer,
+        int partId
+    ) {
         assert topVer != null;
 
         this.reqId = reqId;
@@ -146,6 +153,7 @@ public class DataStreamerRequest implements Message {
         this.clsLdrId = clsLdrId;
         this.forceLocDep = forceLocDep;
         this.topVer = topVer;
+        this.partId = partId;
     }
 
     /**
@@ -253,6 +261,13 @@ public class DataStreamerRequest implements Message {
         return topVer;
     }
 
+    /**
+     * @return Partition ID.
+     */
+    public int partition() {
+        return partId;
+    }
+
     /** {@inheritDoc} */
     @Override public void onAckReceived() {
         // No-op.
@@ -324,42 +339,48 @@ public class DataStreamerRequest implements Message {
                 writer.incrementState();
 
             case 8:
-                if (!writer.writeLong("reqId", reqId))
+                if (!writer.writeInt("partId", partId))
                     return false;
 
                 writer.incrementState();
 
             case 9:
-                if (!writer.writeByteArray("resTopicBytes", resTopicBytes))
+                if (!writer.writeLong("reqId", reqId))
                     return false;
 
                 writer.incrementState();
 
             case 10:
-                if (!writer.writeString("sampleClsName", sampleClsName))
+                if (!writer.writeByteArray("resTopicBytes", resTopicBytes))
                     return false;
 
                 writer.incrementState();
 
             case 11:
-                if (!writer.writeBoolean("skipStore", skipStore))
+                if (!writer.writeString("sampleClsName", sampleClsName))
                     return false;
 
                 writer.incrementState();
 
             case 12:
-                if (!writer.writeMessage("topVer", topVer))
+                if (!writer.writeBoolean("skipStore", skipStore))
                     return false;
 
                 writer.incrementState();
 
             case 13:
-                if (!writer.writeByteArray("updaterBytes", updaterBytes))
+                if (!writer.writeMessage("topVer", topVer))
                     return false;
 
                 writer.incrementState();
 
             case 14:
+                if (!writer.writeByteArray("updaterBytes", updaterBytes))
+                    return false;
+
+                writer.incrementState();
+
+            case 15:
                 if (!writer.writeString("userVer", userVer))
                     return false;
 
@@ -447,7 +468,7 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 8:
-                reqId = reader.readLong("reqId");
+                partId = reader.readInt("partId");
 
                 if (!reader.isLastRead())
                     return false;
@@ -455,7 +476,7 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 9:
-                resTopicBytes = reader.readByteArray("resTopicBytes");
+                reqId = reader.readLong("reqId");
 
                 if (!reader.isLastRead())
                     return false;
@@ -463,7 +484,7 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 10:
-                sampleClsName = reader.readString("sampleClsName");
+                resTopicBytes = reader.readByteArray("resTopicBytes");
 
                 if (!reader.isLastRead())
                     return false;
@@ -471,7 +492,7 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 11:
-                skipStore = reader.readBoolean("skipStore");
+                sampleClsName = reader.readString("sampleClsName");
 
                 if (!reader.isLastRead())
                     return false;
@@ -479,7 +500,7 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 12:
-                topVer = reader.readMessage("topVer");
+                skipStore = reader.readBoolean("skipStore");
 
                 if (!reader.isLastRead())
                     return false;
@@ -487,7 +508,7 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 13:
-                updaterBytes = reader.readByteArray("updaterBytes");
+                topVer = reader.readMessage("topVer");
 
                 if (!reader.isLastRead())
                     return false;
@@ -495,6 +516,14 @@ public class DataStreamerRequest implements Message {
                 reader.incrementState();
 
             case 14:
+                updaterBytes = reader.readByteArray("updaterBytes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 15:
                 userVer = reader.readString("userVer");
 
                 if (!reader.isLastRead())
@@ -514,6 +543,6 @@ public class DataStreamerRequest implements Message {
 
     /** {@inheritDoc} */
     @Override public byte fieldsCount() {
-        return 15;
+        return 16;
     }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
index 6d5dc71..630d34c 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedExecutor.java
@@ -20,10 +20,12 @@ package org.apache.ignite.internal.util;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Deque;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -36,6 +38,8 @@ import java.util.concurrent.locks.LockSupport;
 import org.apache.ignite.IgniteInterruptedException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -62,6 +66,17 @@ public class StripedExecutor implements ExecutorService {
      * @param log Logger.
      */
     public StripedExecutor(int cnt, String igniteInstanceName, String 
poolName, final IgniteLogger log) {
+        this(cnt, igniteInstanceName, poolName, log, false);
+    }
+
+    /**
+     * @param cnt Count.
+     * @param igniteInstanceName Node name.
+     * @param poolName Pool name.
+     * @param log Logger.
+     * @param stealTasks {@code True} to steal tasks.
+     */
+    public StripedExecutor(int cnt, String igniteInstanceName, String 
poolName, final IgniteLogger log, boolean stealTasks) {
         A.ensure(cnt > 0, "cnt > 0");
 
         boolean success = false;
@@ -76,14 +91,19 @@ public class StripedExecutor implements ExecutorService {
 
         try {
             for (int i = 0; i < cnt; i++) {
-                stripes[i] = new StripeConcurrentQueue(
+                stripes[i] = stealTasks ? new StripeConcurrentQueue(
                     igniteInstanceName,
                     poolName,
                     i,
-                    log);
+                    log, stripes) : new StripeConcurrentQueue(
+                        igniteInstanceName,
+                        poolName,
+                        i,
+                        log);
+            }
 
+            for (int i = 0; i < cnt; i++)
                 stripes[i].start();
-            }
 
             success = true;
         }
@@ -397,7 +417,7 @@ public class StripedExecutor implements ExecutorService {
         private final String poolName;
 
         /** */
-        private final int idx;
+        protected final int idx;
 
         /** */
         private final IgniteLogger log;
@@ -536,8 +556,17 @@ public class StripedExecutor implements ExecutorService {
      * Stripe.
      */
     private static class StripeConcurrentQueue extends Stripe {
+        /** */
+        private static final int IGNITE_TASKS_STEALING_THRESHOLD =
+            IgniteSystemProperties.getInteger(
+                
IgniteSystemProperties.IGNITE_DATA_STREAMING_EXECUTOR_SERVICE_TASKS_STEALING_THRESHOLD,
 4);
+
         /** Queue. */
-        private final Queue<Runnable> queue = new ConcurrentLinkedQueue<>();
+        private final Queue<Runnable> queue;
+
+        /** */
+        @GridToStringExclude
+        private final Stripe[] others;
 
         /** */
         private volatile boolean parked;
@@ -548,16 +577,37 @@ public class StripedExecutor implements ExecutorService {
          * @param idx Stripe index.
          * @param log Logger.
          */
-        public StripeConcurrentQueue(
+        StripeConcurrentQueue(
             String igniteInstanceName,
             String poolName,
             int idx,
             IgniteLogger log
         ) {
-            super(igniteInstanceName,
+            this(igniteInstanceName, poolName, idx, log, null);
+        }
+
+        /**
+         * @param igniteInstanceName Ignite instance name.
+         * @param poolName Pool name.
+         * @param idx Stripe index.
+         * @param log Logger.
+         */
+        StripeConcurrentQueue(
+            String igniteInstanceName,
+            String poolName,
+            int idx,
+            IgniteLogger log,
+            Stripe[] others
+        ) {
+            super(
+                igniteInstanceName,
                 poolName,
                 idx,
                 log);
+
+            this.others = others;
+
+            this.queue = others == null ? new 
ConcurrentLinkedQueue<Runnable>() : new ConcurrentLinkedDeque<Runnable>();
         }
 
         /** {@inheritDoc} */
@@ -580,6 +630,24 @@ public class StripedExecutor implements ExecutorService {
                     if (r != null)
                         return r;
 
+                    if(others != null) {
+                        int len = others.length;
+                        int init = ThreadLocalRandom.current().nextInt(len);
+                        int cur = init;
+
+                        while (true) {
+                            if(cur != idx) {
+                                Deque<Runnable> queue = (Deque<Runnable>) 
((StripeConcurrentQueue) others[cur]).queue;
+
+                                if(queue.size() > 
IGNITE_TASKS_STEALING_THRESHOLD && (r = queue.pollLast()) != null)
+                                    return r;
+                            }
+
+                            if ((cur = (cur + 1) % len) == init)
+                                break;
+                        }
+                    }
+
                     LockSupport.park();
 
                     if (Thread.interrupted())
@@ -597,6 +665,13 @@ public class StripedExecutor implements ExecutorService {
 
             if (parked)
                 LockSupport.unpark(thread);
+
+            if(others != null && queueSize() > 
IGNITE_TASKS_STEALING_THRESHOLD) {
+                for (Stripe other : others) {
+                    if(((StripeConcurrentQueue)other).parked)
+                        LockSupport.unpark(other.thread);
+                }
+            }
         }
 
         /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/main/java/org/apache/ignite/thread/IgniteThread.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/thread/IgniteThread.java 
b/modules/core/src/main/java/org/apache/ignite/thread/IgniteThread.java
index c814625..14c3d4e 100644
--- a/modules/core/src/main/java/org/apache/ignite/thread/IgniteThread.java
+++ b/modules/core/src/main/java/org/apache/ignite/thread/IgniteThread.java
@@ -149,6 +149,16 @@ public class IgniteThread extends Thread {
     }
 
     /**
+     * @return IgniteThread or {@code null} if current thread is not an 
instance of IgniteThread.
+     */
+    public static IgniteThread current(){
+        Thread thread = Thread.currentThread();
+
+        return thread.getClass() == IgniteThread.class || thread instanceof 
IgniteThread ?
+            ((IgniteThread)thread) : null;
+    }
+
+    /**
      * Creates new thread name.
      *
      * @param num Thread number.

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java
 
b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java
index ec5e6d0..ac89021 100644
--- 
a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java
+++ 
b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java
@@ -962,8 +962,7 @@ public class DataStreamProcessorSelfTest extends 
GridCommonAbstractTest {
 
             final IgniteCache<String, String> cache = 
ignite.cache(DEFAULT_CACHE_NAME);
 
-            IgniteDataStreamer<String, String> ldr = 
ignite.dataStreamer(DEFAULT_CACHE_NAME);
-            try {
+            try (IgniteDataStreamer<String, String> ldr = 
ignite.dataStreamer(DEFAULT_CACHE_NAME)) {
                 ldr.receiver(new StreamReceiver<String, String>() {
                     @Override public void receive(IgniteCache<String, String> 
cache,
                         Collection<Map.Entry<String, String>> entries) throws 
IgniteException {
@@ -972,6 +971,7 @@ public class DataStreamProcessorSelfTest extends 
GridCommonAbstractTest {
                         cache.put("key", threadName);
                     }
                 });
+
                 ldr.addData("key", "value");
 
                 ldr.tryFlush();
@@ -982,9 +982,6 @@ public class DataStreamProcessorSelfTest extends 
GridCommonAbstractTest {
                     }
                 }, 3_000);
             }
-            finally {
-                ldr.close(true);
-            }
 
             assertNotNull(cache.get("key"));
 
@@ -1011,9 +1008,7 @@ public class DataStreamProcessorSelfTest extends 
GridCommonAbstractTest {
 
             final IgniteCache<String, String> cache = 
ignite.cache(DEFAULT_CACHE_NAME);
 
-            IgniteDataStreamer<String, String> ldr = 
client.dataStreamer(DEFAULT_CACHE_NAME);
-
-            try {
+            try (IgniteDataStreamer<String, String> ldr = 
client.dataStreamer(DEFAULT_CACHE_NAME)) {
                 ldr.receiver(new StringStringStreamReceiver());
 
                 ldr.addData("key", "value");
@@ -1026,9 +1021,6 @@ public class DataStreamProcessorSelfTest extends 
GridCommonAbstractTest {
                     }
                 }, 3_000);
             }
-            finally {
-                ldr.close(true);
-            }
 
             assertNotNull(cache.get("key"));
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/aa81dd14/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
 
b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
index e72a9b4..6d3466b 100644
--- 
a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
+++ 
b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
@@ -346,7 +346,8 @@ public class DataStreamerImplSelfTest extends 
GridCommonAbstractTest {
                             req.participants(),
                             req.classLoaderId(),
                             req.forceLocalDeployment(),
-                            staleTop);
+                            staleTop,
+                            -1);
 
                         msg = new GridIoMessage(
                             GridTestUtils.<Byte>getFieldValue(ioMsg, "plc"),

Reply via email to