ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [29/50] [abbrv] ignite git commit: Added related GridIoPolicy in IgniteThread.
Date Tue, 01 Aug 2017 09:27:14 GMT
Added related GridIoPolicy in IgniteThread.


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

Branch: refs/heads/ignite-5757
Commit: 3a3650fb2eadcd6a1186cc186fa5d471cd1e74d5
Parents: 2574beb
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Jul 27 13:57:11 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Jul 27 13:57:11 2017 +0300

----------------------------------------------------------------------
 .../org/apache/ignite/internal/IgnitionEx.java  | 34 +++++++++++++------
 .../managers/communication/GridIoPolicy.java    |  3 ++
 .../service/GridServiceProcessor.java           |  2 +-
 .../util/StripedCompositeReadWriteLock.java     |  6 ++--
 .../ignite/internal/util/StripedExecutor.java   |  4 ++-
 .../org/apache/ignite/thread/IgniteThread.java  | 35 ++++++++++----------
 .../ignite/thread/IgniteThreadFactory.java      | 15 ++++++---
 .../ignite/thread/IgniteThreadPoolExecutor.java | 33 +++++++++++++++++-
 .../GridThreadPoolExecutorServiceSelfTest.java  |  2 +-
 9 files changed, 95 insertions(+), 39 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/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 1139ec6..23baeb3 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
@@ -64,6 +64,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.configuration.MemoryConfiguration;
 import org.apache.ignite.configuration.TransactionConfiguration;
 import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor;
 import org.apache.ignite.internal.processors.igfs.IgfsThreadFactory;
 import org.apache.ignite.internal.processors.igfs.IgfsUtils;
@@ -1694,7 +1695,8 @@ public class IgnitionEx {
                 cfg.getPublicThreadPoolSize(),
                 cfg.getPublicThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.PUBLIC_POOL);
 
             execSvc.allowCoreThreadTimeOut(true);
 
@@ -1706,7 +1708,8 @@ public class IgnitionEx {
                 cfg.getServiceThreadPoolSize(),
                 cfg.getServiceThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.SERVICE_POOL);
 
             svcExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1718,7 +1721,8 @@ public class IgnitionEx {
                 cfg.getSystemThreadPoolSize(),
                 cfg.getSystemThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.SYSTEM_POOL);
 
             sysExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1738,7 +1742,8 @@ public class IgnitionEx {
                 cfg.getManagementThreadPoolSize(),
                 cfg.getManagementThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.MANAGEMENT_POOL);
 
             mgmtExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1753,7 +1758,8 @@ public class IgnitionEx {
                 cfg.getPeerClassLoadingThreadPoolSize(),
                 cfg.getPeerClassLoadingThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.P2P_POOL);
 
             p2pExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1764,7 +1770,8 @@ public class IgnitionEx {
                 cfg.getDataStreamerThreadPoolSize(),
                 cfg.getDataStreamerThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.DATA_STREAMER_POOL);
 
             dataStreamerExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1811,7 +1818,8 @@ public class IgnitionEx {
                 myCfg.getUtilityCacheThreadPoolSize(),
                 myCfg.getUtilityCacheThreadPoolSize(),
                 myCfg.getUtilityCacheKeepAliveTime(),
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.UTILITY_CACHE_POOL);
 
             utilityCacheExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1821,7 +1829,8 @@ public class IgnitionEx {
                 1,
                 1,
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.AFFINITY_POOL);
 
             affExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1834,7 +1843,8 @@ public class IgnitionEx {
                     cpus,
                     cpus * 2,
                     3000L,
-                    new LinkedBlockingQueue<Runnable>(1000)
+                    new LinkedBlockingQueue<Runnable>(1000),
+                    GridIoPolicy.IDX_POOL
                 );
             }
 
@@ -1846,7 +1856,8 @@ public class IgnitionEx {
                 cfg.getQueryThreadPoolSize(),
                 cfg.getQueryThreadPoolSize(),
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.QUERY_POOL);
 
             qryExecSvc.allowCoreThreadTimeOut(true);
 
@@ -1856,7 +1867,8 @@ public class IgnitionEx {
                 2,
                 2,
                 DFLT_THREAD_KEEP_ALIVE_TIME,
-                new LinkedBlockingQueue<Runnable>());
+                new LinkedBlockingQueue<Runnable>(),
+                GridIoPolicy.SCHEMA_POOL);
 
             schemaExecSvc.allowCoreThreadTimeOut(true);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoPolicy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoPolicy.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoPolicy.java
index 13bc4c4..3f31f92 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoPolicy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoPolicy.java
@@ -22,6 +22,9 @@ package org.apache.ignite.internal.managers.communication;
  * message processing by the communication manager.
  */
 public class GridIoPolicy {
+    /** */
+    public static final byte UNDEFINED = -1;
+
     /** Public execution pool. */
     public static final byte PUBLIC_POOL = 0;
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
index db632ec..46fcfea 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/GridServiceProcessor.java
@@ -153,7 +153,7 @@ public class GridServiceProcessor extends GridProcessorAdapter implements
Ignite
     private volatile GridSpinBusyLock busyLock = new GridSpinBusyLock();
 
     /** Thread factory. */
-    private ThreadFactory threadFactory = new IgniteThreadFactory(ctx.igniteInstanceName());
+    private ThreadFactory threadFactory = new IgniteThreadFactory(ctx.igniteInstanceName(),
"service");
 
     /** Thread local for service name. */
     private ThreadLocal<String> svcName = new ThreadLocal<>();

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
index e215663..18ef06c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
@@ -67,8 +67,10 @@ public class StripedCompositeReadWriteLock implements ReadWriteLock {
     @NotNull @Override public Lock readLock() {
         int idx;
 
-        if (Thread.currentThread() instanceof IgniteThread) {
-            IgniteThread igniteThread = (IgniteThread)Thread.currentThread();
+        Thread curThread = Thread.currentThread();
+
+        if (curThread instanceof IgniteThread) {
+            IgniteThread igniteThread = (IgniteThread)curThread;
 
             idx = igniteThread.compositeRwLockIndex();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/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 6c85b32..6d5dc71 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
@@ -35,6 +35,7 @@ import java.util.concurrent.TimeoutException;
 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.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -439,7 +440,8 @@ public class StripedExecutor implements ExecutorService {
                 poolName + "-stripe-" + idx,
                 this,
                 IgniteThread.GRP_IDX_UNASSIGNED,
-                idx);
+                idx,
+                GridIoPolicy.UNDEFINED);
 
             thread.start();
         }

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/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 6005ac9..c814625 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
@@ -18,6 +18,7 @@
 package org.apache.ignite.thread;
 
 import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.worker.GridWorker;
@@ -52,13 +53,16 @@ public class IgniteThread extends Thread {
     /** */
     private final int stripe;
 
+    /** */
+    private final byte plc;
+
     /**
      * Creates thread with given worker.
      *
      * @param worker Runnable to create thread with.
      */
     public IgniteThread(GridWorker worker) {
-        this(DFLT_GRP, worker.igniteInstanceName(), worker.name(), worker, GRP_IDX_UNASSIGNED,
-1);
+        this(worker.igniteInstanceName(), worker.name(), worker, GRP_IDX_UNASSIGNED, -1,
GridIoPolicy.UNDEFINED);
     }
 
     /**
@@ -69,41 +73,28 @@ public class IgniteThread extends Thread {
      * @param r Runnable to execute.
      */
     public IgniteThread(String igniteInstanceName, String threadName, Runnable r) {
-        this(igniteInstanceName, threadName, r, GRP_IDX_UNASSIGNED, -1);
-    }
-
-    /**
-     * Creates grid thread with given name for a given Ignite instance.
-     *
-     * @param igniteInstanceName Name of the Ignite instance this thread is created for.
-     * @param threadName Name of thread.
-     * @param r Runnable to execute.
-     * @param grpIdx Index within a group.
-     * @param stripe Non-negative stripe number if this thread is striped pool thread.
-     */
-    public IgniteThread(String igniteInstanceName, String threadName, Runnable r, int grpIdx,
int stripe) {
-        this(DFLT_GRP, igniteInstanceName, threadName, r, grpIdx, stripe);
+        this(igniteInstanceName, threadName, r, GRP_IDX_UNASSIGNED, -1, GridIoPolicy.UNDEFINED);
     }
 
     /**
      * Creates grid thread with given name for a given Ignite instance with specified
      * thread group.
      *
-     * @param grp Thread group.
      * @param igniteInstanceName Name of the Ignite instance this thread is created for.
      * @param threadName Name of thread.
      * @param r Runnable to execute.
      * @param grpIdx Thread index within a group.
      * @param stripe Non-negative stripe number if this thread is striped pool thread.
      */
-    public IgniteThread(ThreadGroup grp, String igniteInstanceName, String threadName, Runnable
r, int grpIdx, int stripe) {
-        super(grp, r, createName(cntr.incrementAndGet(), threadName, igniteInstanceName));
+    public IgniteThread(String igniteInstanceName, String threadName, Runnable r, int grpIdx,
int stripe, byte plc) {
+        super(DFLT_GRP, r, createName(cntr.incrementAndGet(), threadName, igniteInstanceName));
 
         A.ensure(grpIdx >= -1, "grpIdx >= -1");
 
         this.igniteInstanceName = igniteInstanceName;
         this.compositeRwLockIdx = grpIdx;
         this.stripe = stripe;
+        this.plc = plc;
     }
 
     /**
@@ -117,6 +108,14 @@ public class IgniteThread extends Thread {
         this.igniteInstanceName = igniteInstanceName;
         this.compositeRwLockIdx = GRP_IDX_UNASSIGNED;
         this.stripe = -1;
+        this.plc = GridIoPolicy.UNDEFINED;
+    }
+
+    /**
+     * @return Related {@link GridIoPolicy} for internal Ignite pools.
+     */
+    public byte policy() {
+        return plc;
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadFactory.java
b/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadFactory.java
index d2f0b15..062c973 100644
--- a/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadFactory.java
@@ -20,6 +20,7 @@ package org.apache.ignite.thread;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.jetbrains.annotations.NotNull;
 
@@ -37,14 +38,18 @@ public class IgniteThreadFactory implements ThreadFactory {
     /** Index generator for threads. */
     private final AtomicInteger idxGen = new AtomicInteger();
 
+    /** */
+    private final byte plc;
+
     /**
      * Constructs new thread factory for given grid. All threads will belong
      * to the same default thread group.
      *
      * @param igniteInstanceName Ignite instance name.
+     * @param threadName Thread name.
      */
-    public IgniteThreadFactory(String igniteInstanceName) {
-        this(igniteInstanceName, "ignite");
+    public IgniteThreadFactory(String igniteInstanceName, String threadName) {
+        this(igniteInstanceName, threadName, GridIoPolicy.UNDEFINED);
     }
 
     /**
@@ -53,15 +58,17 @@ public class IgniteThreadFactory implements ThreadFactory {
      *
      * @param igniteInstanceName Ignite instance name.
      * @param threadName Thread name.
+     * @param plc {@link GridIoPolicy} for thread pool.
      */
-    public IgniteThreadFactory(String igniteInstanceName, String threadName) {
+    public IgniteThreadFactory(String igniteInstanceName, String threadName, byte plc) {
         this.igniteInstanceName = igniteInstanceName;
         this.threadName = threadName;
+        this.plc = plc;
     }
 
     /** {@inheritDoc} */
     @Override public Thread newThread(@NotNull Runnable r) {
-        return new IgniteThread(igniteInstanceName, threadName, r, idxGen.incrementAndGet(),
-1);
+        return new IgniteThread(igniteInstanceName, threadName, r, idxGen.incrementAndGet(),
-1, plc);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadPoolExecutor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadPoolExecutor.java
b/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadPoolExecutor.java
index 639ef94..83c64c3 100644
--- a/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadPoolExecutor.java
+++ b/modules/core/src/main/java/org/apache/ignite/thread/IgniteThreadPoolExecutor.java
@@ -22,6 +22,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 
 /**
  * An {@link ExecutorService} that executes submitted tasks using pooled grid threads.
@@ -46,13 +47,43 @@ public class IgniteThreadPoolExecutor extends ThreadPoolExecutor {
         int maxPoolSize,
         long keepAliveTime,
         BlockingQueue<Runnable> workQ) {
+        this(threadNamePrefix,
+            igniteInstanceName,
+            corePoolSize,
+            maxPoolSize,
+            keepAliveTime,
+            workQ,
+            GridIoPolicy.UNDEFINED);
+    }
+
+    /**
+     * Creates a new service with the given initial parameters.
+     *
+     * @param threadNamePrefix Will be added at the beginning of all created threads.
+     * @param igniteInstanceName Must be the name of the grid.
+     * @param corePoolSize The number of threads to keep in the pool, even if they are idle.
+     * @param maxPoolSize The maximum number of threads to allow in the pool.
+     * @param keepAliveTime When the number of threads is greater than the core, this is
the maximum time
+     *      that excess idle threads will wait for new tasks before terminating.
+     * @param workQ The queue to use for holding tasks before they are executed. This queue
will hold only
+     *      runnable tasks submitted by the {@link #execute(Runnable)} method.
+     * @param plc {@link GridIoPolicy} for thread pool.
+     */
+    public IgniteThreadPoolExecutor(
+        String threadNamePrefix,
+        String igniteInstanceName,
+        int corePoolSize,
+        int maxPoolSize,
+        long keepAliveTime,
+        BlockingQueue<Runnable> workQ,
+        byte plc) {
         super(
             corePoolSize,
             maxPoolSize,
             keepAliveTime,
             TimeUnit.MILLISECONDS,
             workQ,
-            new IgniteThreadFactory(igniteInstanceName, threadNamePrefix)
+            new IgniteThreadFactory(igniteInstanceName, threadNamePrefix, plc)
         );
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/3a3650fb/modules/core/src/test/java/org/apache/ignite/thread/GridThreadPoolExecutorServiceSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/thread/GridThreadPoolExecutorServiceSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/thread/GridThreadPoolExecutorServiceSelfTest.java
index 3948f6a..dce6328 100644
--- a/modules/core/src/test/java/org/apache/ignite/thread/GridThreadPoolExecutorServiceSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/thread/GridThreadPoolExecutorServiceSelfTest.java
@@ -64,7 +64,7 @@ public class GridThreadPoolExecutorServiceSelfTest extends GridCommonAbstractTes
      * @throws Exception If failed.
      */
     public void testSingleGridThreadExecutor() throws Exception {
-        ExecutorService exec = Executors.newSingleThreadExecutor(new IgniteThreadFactory("gridName"));
+        ExecutorService exec = Executors.newSingleThreadExecutor(new IgniteThreadFactory("gridName",
"testThread"));
 
         exec.submit(new InterruptingRunnable()).get();
 


Mime
View raw message