ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [16/40] ignite git commit: IGNITE-4041: Created separate processor for thread pools and refactored IO manager. This closes #1150.
Date Mon, 17 Oct 2016 17:59:13 GMT
IGNITE-4041: Created separate processor for thread pools and refactored IO manager. This closes
#1150.


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

Branch: refs/heads/ignite-ssl-hotfix
Commit: f9a0676fad7fd6c23e3c91c10d7e0412ccb27c06
Parents: b9c776a
Author: vozerov-gridgain <vozerov@gridgain.com>
Authored: Tue Oct 11 10:23:01 2016 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Tue Oct 11 10:23:01 2016 +0300

----------------------------------------------------------------------
 .../ignite/internal/GridKernalContext.java      |  15 ++
 .../ignite/internal/GridKernalContextImpl.java  |  24 +++
 .../apache/ignite/internal/IgniteKernal.java    |   6 +
 .../org/apache/ignite/internal/IgnitionEx.java  |  19 ++-
 .../managers/communication/GridIoManager.java   | 161 ++-----------------
 .../processors/closure/GridClosurePolicy.java   |  51 ------
 .../closure/GridClosureProcessor.java           |  88 +++-------
 .../internal/processors/pool/PoolProcessor.java | 149 +++++++++++++++++
 .../resources/META-INF/classnames.properties    |   1 -
 .../managers/GridManagerStopSelfTest.java       |   2 +
 .../util/future/GridFutureAdapterSelfTest.java  |   3 +
 .../junits/GridTestKernalContext.java           |   1 +
 12 files changed, 250 insertions(+), 270 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/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 b123a4a..e608af2 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
@@ -53,6 +53,7 @@ import org.apache.ignite.internal.processors.odbc.OdbcProcessor;
 import org.apache.ignite.internal.processors.offheap.GridOffHeapProcessor;
 import org.apache.ignite.internal.processors.platform.PlatformProcessor;
 import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.processors.port.GridPortProcessor;
 import org.apache.ignite.internal.processors.query.GridQueryProcessor;
 import org.apache.ignite.internal.processors.resource.GridResourceProcessor;
@@ -286,6 +287,13 @@ public interface GridKernalContext extends Iterable<GridComponent>
{
     public HadoopProcessorAdapter hadoop();
 
     /**
+     * Gets pool processor.
+     *
+     * @return Pool processor.
+     */
+    public PoolProcessor pools();
+
+    /**
      * Gets Hadoop helper.
      *
      * @return Hadoop helper.
@@ -533,6 +541,13 @@ public interface GridKernalContext extends Iterable<GridComponent>
{
     public ExecutorService getRestExecutorService();
 
     /**
+     * Get affinity executor service.
+     *
+     * @return Affinity executor service.
+     */
+    public ExecutorService getAffinityExecutorService();
+
+    /**
      * Gets exception registry.
      *
      * @return Exception registry.

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/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 c7e26e9..ddef345 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
@@ -70,6 +70,7 @@ import org.apache.ignite.internal.processors.odbc.OdbcProcessor;
 import org.apache.ignite.internal.processors.offheap.GridOffHeapProcessor;
 import org.apache.ignite.internal.processors.platform.PlatformProcessor;
 import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.processors.port.GridPortProcessor;
 import org.apache.ignite.internal.processors.query.GridQueryProcessor;
 import org.apache.ignite.internal.processors.resource.GridResourceProcessor;
@@ -259,6 +260,10 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
 
     /** */
     @GridToStringExclude
+    private PoolProcessor poolProc;
+
+    /** */
+    @GridToStringExclude
     private IgnitePluginProcessor pluginProc;
 
     /** */
@@ -311,6 +316,10 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
 
     /** */
     @GridToStringExclude
+    protected ExecutorService affExecSvc;
+
+    /** */
+    @GridToStringExclude
     protected IgniteStripedThreadPoolExecutor callbackExecSvc;
 
     /** */
@@ -372,6 +381,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
      * @param mgmtExecSvc Management executor service.
      * @param igfsExecSvc IGFS executor service.
      * @param restExecSvc REST executor service.
+     * @param affExecSvc Affinity executor service.
      * @param plugins Plugin providers.
      * @throws IgniteCheckedException In case of error.
      */
@@ -389,6 +399,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
         ExecutorService mgmtExecSvc,
         ExecutorService igfsExecSvc,
         ExecutorService restExecSvc,
+        ExecutorService affExecSvc,
         IgniteStripedThreadPoolExecutor callbackExecSvc,
         List<PluginProvider> plugins) throws IgniteCheckedException {
         assert grid != null;
@@ -406,6 +417,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
         this.mgmtExecSvc = mgmtExecSvc;
         this.igfsExecSvc = igfsExecSvc;
         this.restExecSvc = restExecSvc;
+        this.affExecSvc = affExecSvc;
         this.callbackExecSvc = callbackExecSvc;
 
         String workDir = U.workDirectory(cfg.getWorkDirectory(), cfg.getIgniteHome());
@@ -533,6 +545,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
             cluster = (ClusterProcessor)comp;
         else if (comp instanceof PlatformProcessor)
             platformProc = (PlatformProcessor)comp;
+        else if (comp instanceof PoolProcessor)
+            poolProc = (PoolProcessor) comp;
         else if (!(comp instanceof DiscoveryNodeValidationProcessor))
             assert (comp instanceof GridPluginComponent) : "Unknown manager class: " + comp.getClass();
 
@@ -757,6 +771,11 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
     }
 
     /** {@inheritDoc} */
+    @Override public PoolProcessor pools() {
+        return poolProc;
+    }
+
+    /** {@inheritDoc} */
     @Override public ExecutorService utilityCachePool() {
         return utilityCachePool;
     }
@@ -942,6 +961,11 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
     }
 
     /** {@inheritDoc} */
+    @Override public ExecutorService getAffinityExecutorService() {
+        return affExecSvc;
+    }
+
+    /** {@inheritDoc} */
     @Override public IgniteExceptionRegistry exceptionRegistry() {
         return IgniteExceptionRegistry.get();
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/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 e0a36a7..02f16af 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
@@ -126,6 +126,7 @@ import org.apache.ignite.internal.processors.offheap.GridOffHeapProcessor;
 import org.apache.ignite.internal.processors.platform.PlatformNoopProcessor;
 import org.apache.ignite.internal.processors.platform.PlatformProcessor;
 import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.processors.port.GridPortProcessor;
 import org.apache.ignite.internal.processors.port.GridPortRecord;
 import org.apache.ignite.internal.processors.query.GridQueryProcessor;
@@ -666,6 +667,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable
{
      * @param mgmtExecSvc Management executor service.
      * @param igfsExecSvc IGFS executor service.
      * @param restExecSvc Reset executor service.
+     * @param affExecSvc Affinity executor service.
      * @param errHnd Error handler to use for notification about startup problems.
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
@@ -679,6 +681,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable
{
         ExecutorService mgmtExecSvc,
         ExecutorService igfsExecSvc,
         ExecutorService restExecSvc,
+        ExecutorService affExecSvc,
         IgniteStripedThreadPoolExecutor callbackExecSvc,
         GridAbsClosure errHnd)
         throws IgniteCheckedException
@@ -784,6 +787,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable
{
                 mgmtExecSvc,
                 igfsExecSvc,
                 restExecSvc,
+                affExecSvc,
                 callbackExecSvc,
                 plugins);
 
@@ -827,6 +831,8 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable
{
 
             startProcessor(new IgnitePluginProcessor(ctx, cfg, plugins));
 
+            startProcessor(new PoolProcessor(ctx));
+
             // Off-heap processor has no dependencies.
             startProcessor(new GridOffHeapProcessor(ctx));
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/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 001f599..a6860b3 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
@@ -1477,6 +1477,9 @@ public class IgnitionEx {
         /** Marshaller cache executor service. */
         private ThreadPoolExecutor marshCacheExecSvc;
 
+        /** Affinity executor service. */
+        private ThreadPoolExecutor affExecSvc;
+
         /** Continuous query executor service. */
         private IgniteStripedThreadPoolExecutor callbackExecSvc;
 
@@ -1734,6 +1737,16 @@ public class IgnitionEx {
 
             marshCacheExecSvc.allowCoreThreadTimeOut(true);
 
+            affExecSvc = new IgniteThreadPoolExecutor(
+                "aff",
+                cfg.getGridName(),
+                1,
+                1,
+                DFLT_THREAD_KEEP_ALIVE_TIME,
+                new LinkedBlockingQueue<Runnable>());
+
+            affExecSvc.allowCoreThreadTimeOut(true);
+
             // Register Ignite MBean for current grid instance.
             registerFactoryMbean(myCfg.getMBeanServer());
 
@@ -1746,7 +1759,7 @@ public class IgnitionEx {
                 grid = grid0;
 
                 grid0.start(myCfg, utilityCacheExecSvc, marshCacheExecSvc, execSvc, sysExecSvc,
p2pExecSvc, mgmtExecSvc,
-                    igfsExecSvc, restExecSvc, callbackExecSvc,
+                    igfsExecSvc, restExecSvc, affExecSvc, callbackExecSvc,
                     new CA() {
                         @Override public void apply() {
                             startLatch.countDown();
@@ -2381,6 +2394,10 @@ public class IgnitionEx {
 
             marshCacheExecSvc = null;
 
+            U.shutdownNow(getClass(), affExecSvc, log);
+
+            affExecSvc = null;
+
             U.shutdownNow(getClass(), callbackExecSvc, log);
 
             callbackExecSvc = null;

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/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 4bc2eea..bd285b2 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
@@ -29,9 +29,6 @@ import java.util.UUID;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -54,6 +51,7 @@ import org.apache.ignite.internal.managers.deployment.GridDeployment;
 import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
 import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
 import org.apache.ignite.internal.processors.platform.message.PlatformMessageFilter;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
 import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashSet;
 import org.apache.ignite.internal.util.GridSpinReadWriteLock;
@@ -68,7 +66,6 @@ import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.lang.IgniteRunnable;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.marshaller.Marshaller;
-import org.apache.ignite.plugin.extensions.communication.IoPool;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
@@ -78,7 +75,6 @@ import org.apache.ignite.spi.IgniteSpiException;
 import org.apache.ignite.spi.communication.CommunicationListener;
 import org.apache.ignite.spi.communication.CommunicationSpi;
 import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
-import org.apache.ignite.thread.IgniteThreadPoolExecutor;
 import org.jetbrains.annotations.Nullable;
 import org.jsr166.ConcurrentHashMap8;
 import org.jsr166.ConcurrentLinkedDeque8;
@@ -130,32 +126,8 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
     /** Disconnect listeners. */
     private final Collection<GridDisconnectListener> disconnectLsnrs = new ConcurrentLinkedQueue<>();
 
-    /** Map of {@link IoPool}-s injected by Ignite plugins. */
-    private final IoPool[] ioPools = new IoPool[128];
-
-    /** Public pool. */
-    private ExecutorService pubPool;
-
-    /** Internal P2P pool. */
-    private ExecutorService p2pPool;
-
-    /** Internal system pool. */
-    private ExecutorService sysPool;
-
-    /** Internal management pool. */
-    private ExecutorService mgmtPool;
-
-    /** Affinity assignment executor service. */
-    private ExecutorService affPool;
-
-    /** Utility cache pool. */
-    private ExecutorService utilityCachePool;
-
-    /** Marshaller cache pool. */
-    private ExecutorService marshCachePool;
-
-    /** IGFS pool. */
-    private ExecutorService igfsPool;
+    /** Pool processor. */
+    private PoolProcessor pools;
 
     /** Discovery listener. */
     private GridLocalEventListener discoLsnr;
@@ -210,6 +182,10 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
     public GridIoManager(GridKernalContext ctx) {
         super(ctx, ctx.config().getCommunicationSpi());
 
+        pools = ctx.pools();
+
+        assert pools != null;
+
         locNodeId = ctx.localNodeId();
 
         discoDelay = ctx.config().getDiscoveryStartupDelay();
@@ -253,21 +229,6 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
 
         startSpi();
 
-        pubPool = ctx.getExecutorService();
-        p2pPool = ctx.getPeerClassLoadingExecutorService();
-        sysPool = ctx.getSystemExecutorService();
-        mgmtPool = ctx.getManagementExecutorService();
-        utilityCachePool = ctx.utilityCachePool();
-        marshCachePool = ctx.marshallerCachePool();
-        igfsPool = ctx.getIgfsExecutorService();
-        affPool = new IgniteThreadPoolExecutor(
-            "aff",
-            ctx.gridName(),
-            1,
-            1,
-            0,
-            new LinkedBlockingQueue<Runnable>());
-
         getSpi().setListener(commLsnr = new CommunicationListener<Serializable>() {
             @Override public void onMessage(UUID nodeId, Serializable msg, IgniteRunnable
msgC) {
                 try {
@@ -335,41 +296,6 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
 
         if (log.isDebugEnabled())
             log.debug(startInfo());
-
-        registerIoPoolExtensions();
-    }
-
-    /**
-     * Processes IO messaging pool extensions.
-     * @throws IgniteCheckedException On error.
-     */
-    private void registerIoPoolExtensions() throws IgniteCheckedException {
-        // Process custom IO messaging pool extensions:
-        final IoPool[] executorExtensions = ctx.plugins().extensions(IoPool.class);
-
-        if (executorExtensions != null) {
-            // Store it into the map and check for duplicates:
-            for (IoPool ex : executorExtensions) {
-                final byte id = ex.id();
-
-                // 1. Check the pool id is non-negative:
-                if (id < 0)
-                    throw new IgniteCheckedException("Failed to register IO executor pool
because its Id is negative " +
-                        "[id=" + id + ']');
-
-                // 2. Check the pool id is in allowed range:
-                if (isReservedGridIoPolicy(id))
-                    throw new IgniteCheckedException("Failed to register IO executor pool
because its Id in in the " +
-                        "reserved range (0-31) [id=" + id + ']');
-
-                // 3. Check the pool for duplicates:
-                if (ioPools[id] != null)
-                    throw new IgniteCheckedException("Failed to register IO executor pool
because its " +
-                        "Id as already used [id=" + id + ']');
-
-                ioPools[id] = ex;
-            }
-        }
     }
 
     /** {@inheritDoc} */
@@ -546,8 +472,6 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
             if (interrupted)
                 Thread.currentThread().interrupt();
 
-            U.shutdownNow(getClass(), affPool, log);
-
             GridEventStorageManager evtMgr = ctx.event();
 
             if (evtMgr != null && discoLsnr != null)
@@ -566,8 +490,6 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
 
         if (log.isDebugEnabled())
             log.debug(stopInfo());
-
-        Arrays.fill(ioPools, null);
     }
 
     /**
@@ -683,67 +605,6 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
     }
 
     /**
-     * Gets execution pool for policy.
-     *
-     * @param plc Policy.
-     * @return Execution pool.
-     * @throws IgniteCheckedException If failed.
-     */
-    private Executor pool(byte plc) throws IgniteCheckedException {
-        switch (plc) {
-            case P2P_POOL:
-                return p2pPool;
-            case SYSTEM_POOL:
-                return sysPool;
-            case PUBLIC_POOL:
-                return pubPool;
-            case MANAGEMENT_POOL:
-                return mgmtPool;
-            case AFFINITY_POOL:
-                return affPool;
-
-            case UTILITY_CACHE_POOL:
-                assert utilityCachePool != null : "Utility cache pool is not configured.";
-
-                return utilityCachePool;
-
-            case MARSH_CACHE_POOL:
-                assert marshCachePool != null : "Marshaller cache pool is not configured.";
-
-                return marshCachePool;
-
-            case IGFS_POOL:
-                assert igfsPool != null : "IGFS pool is not configured.";
-
-                return igfsPool;
-
-            default: {
-                assert plc >= 0 : "Negative policy: " + plc;
-
-                if (isReservedGridIoPolicy(plc))
-                    throw new IgniteCheckedException("Failed to process message with policy
of reserved" +
-                        " range (0-31), [policy=" + plc + ']');
-
-                IoPool pool = ioPools[plc];
-
-                if (pool == null)
-                    throw new IgniteCheckedException("Failed to process message because no
pool is registered " +
-                        "for policy. [policy=" + plc + ']');
-
-                assert plc == pool.id();
-
-                Executor ex = pool.executor();
-
-                if (ex == null)
-                    throw new IgniteCheckedException("Failed to process message because corresponding
executor " +
-                        "is null. [id=" + plc + ']');
-
-                return ex;
-            }
-        }
-    }
-
-    /**
      * @param nodeId Node ID.
      * @param msg Message.
      * @param msgC Closure to call when message processing finished.
@@ -778,7 +639,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
         };
 
         try {
-            p2pPool.execute(c);
+            pools.p2pPool().execute(c);
         }
         catch (RejectedExecutionException e) {
             U.error(log, "Failed to process P2P message due to execution rejection. Increase
the upper bound " +
@@ -818,7 +679,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
         };
 
         try {
-            pool(plc).execute(c);
+            pools.poolForPolicy(plc).execute(c);
         }
         catch (RejectedExecutionException e) {
             U.error(log, "Failed to process regular message due to execution rejection. Increase
the upper bound " +
@@ -1154,7 +1015,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
         };
 
         try {
-            pool(plc).execute(c);
+            pools.poolForPolicy(plc).execute(c);
         }
         catch (RejectedExecutionException e) {
             U.error(log, "Failed to process ordered message due to execution rejection. "
+
@@ -1781,7 +1642,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
 
             try {
                 for (final GridCommunicationMessageSet msgSet : msgSets) {
-                    pool(msgSet.policy()).execute(
+                    pools.poolForPolicy(msgSet.policy()).execute(
                         new Runnable() {
                             @Override public void run() {
                                 unwindMessageSet(msgSet, lsnrs0);

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosurePolicy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosurePolicy.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosurePolicy.java
deleted file mode 100644
index c17cedd..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosurePolicy.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.closure;
-
-import org.jetbrains.annotations.Nullable;
-
-/**
- * This enumeration defines different types of closure
- * processing by the closure processor.
- */
-public enum GridClosurePolicy {
-    /** Public execution pool. */
-    PUBLIC_POOL,
-
-    /** P2P execution pool. */
-    P2P_POOL,
-
-    /** System execution pool. */
-    SYSTEM_POOL,
-
-    /** IGFS pool. */
-    IGFS_POOL;
-
-    /** Enum values. */
-    private static final GridClosurePolicy[] VALS = values();
-
-    /**
-     * Efficiently gets enumerated value from its ordinal.
-     *
-     * @param ord Ordinal value.
-     * @return Enumerated value.
-     */
-    @Nullable public static GridClosurePolicy fromOrdinal(int ord) {
-        return ord >= 0 && ord < VALS.length ? VALS[ord] : null;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
index 6f878ce..a96d6eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
@@ -50,8 +50,10 @@ import org.apache.ignite.internal.GridClosureCallMode;
 import org.apache.ignite.internal.GridInternalWrapper;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.processors.GridProcessorAdapter;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.processors.resource.GridNoImplicitInjection;
 import org.apache.ignite.internal.util.GridSpinReadWriteLock;
 import org.apache.ignite.internal.util.IgniteUtils;
@@ -88,14 +90,8 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     /** Ignite version in which binarylizable versions of closures were introduced. */
     public static final IgniteProductVersion BINARYLIZABLE_CLOSURES_SINCE = IgniteProductVersion.fromString("1.6.0");
 
-    /** */
-    private final Executor sysPool;
-
-    /** */
-    private final Executor pubPool;
-
-    /** */
-    private final Executor igfsPool;
+    /** Pool processor. */
+    private final PoolProcessor pools;
 
     /** Lock to control execution after stop. */
     private final GridSpinReadWriteLock busyLock = new GridSpinReadWriteLock();
@@ -109,9 +105,9 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     public GridClosureProcessor(GridKernalContext ctx) {
         super(ctx);
 
-        sysPool = ctx.getSystemExecutorService();
-        pubPool = ctx.getExecutorService();
-        igfsPool = ctx.getIgfsExecutorService();
+        pools = ctx.pools();
+
+        assert pools != null;
     }
 
     /** {@inheritDoc} */
@@ -731,57 +727,13 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     }
 
     /**
-     * Gets pool by execution policy.
-     *
-     * @param plc Whether to get system or public pool.
-     * @return Requested worker pool.
-     */
-    private Executor pool(GridClosurePolicy plc) {
-        switch (plc) {
-            case PUBLIC_POOL:
-                return pubPool;
-
-            case SYSTEM_POOL:
-                return sysPool;
-
-            case IGFS_POOL:
-                return igfsPool;
-
-            default:
-                throw new IllegalArgumentException("Invalid closure execution policy: " +
plc);
-        }
-    }
-
-    /**
-     * Gets pool name by execution policy.
-     *
-     * @param plc Policy to choose executor pool.
-     * @return Pool name.
-     */
-    private String poolName(GridClosurePolicy plc) {
-        switch (plc) {
-            case PUBLIC_POOL:
-                return "public";
-
-            case SYSTEM_POOL:
-                return "system";
-
-            case IGFS_POOL:
-                return "igfs";
-
-            default:
-                throw new IllegalArgumentException("Invalid closure execution policy: " +
plc);
-        }
-    }
-
-    /**
      * @param c Closure to execute.
      * @param sys If {@code true}, then system pool will be used, otherwise public pool will
be used.
      * @return Future.
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
     private IgniteInternalFuture<?> runLocal(@Nullable final Runnable c, boolean sys)
throws IgniteCheckedException {
-        return runLocal(c, sys ? GridClosurePolicy.SYSTEM_POOL : GridClosurePolicy.PUBLIC_POOL);
+        return runLocal(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
     }
 
     /**
@@ -790,7 +742,8 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @return Future.
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
-    private IgniteInternalFuture<?> runLocal(@Nullable final Runnable c, GridClosurePolicy
plc) throws IgniteCheckedException {
+    private IgniteInternalFuture<?> runLocal(@Nullable final Runnable c, byte plc)
+        throws IgniteCheckedException {
         if (c == null)
             return new GridFinishedFuture();
 
@@ -830,11 +783,11 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             fut.setWorker(w);
 
             try {
-                pool(plc).execute(w);
+                pools.poolForPolicy(plc).execute(w);
             }
             catch (RejectedExecutionException e) {
                 U.error(log, "Failed to execute worker due to execution rejection " +
-                    "(increase upper bound on " + poolName(plc) + " executor service).",
e);
+                    "(increase upper bound on executor service) [policy=" + plc + ']', e);
 
                 w.run();
             }
@@ -866,7 +819,7 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @return Future.
      */
     public IgniteInternalFuture<?> runLocalSafe(Runnable c, boolean sys) {
-        return runLocalSafe(c, sys ? GridClosurePolicy.SYSTEM_POOL : GridClosurePolicy.PUBLIC_POOL);
+        return runLocalSafe(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
     }
 
     /**
@@ -877,7 +830,7 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @param plc Policy to choose executor pool.
      * @return Future.
      */
-    public IgniteInternalFuture<?> runLocalSafe(Runnable c, GridClosurePolicy plc)
{
+    public IgniteInternalFuture<?> runLocalSafe(Runnable c, byte plc) {
         try {
             return runLocal(c, plc);
         }
@@ -921,7 +874,7 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
     private <R> IgniteInternalFuture<R> callLocal(@Nullable final Callable<R>
c, boolean sys) throws IgniteCheckedException {
-        return callLocal(c, sys ? GridClosurePolicy.SYSTEM_POOL : GridClosurePolicy.PUBLIC_POOL);
+        return callLocal(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
     }
 
     /**
@@ -931,7 +884,8 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @return Future.
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
-    private <R> IgniteInternalFuture<R> callLocal(@Nullable final Callable<R>
c, GridClosurePolicy plc) throws IgniteCheckedException {
+    private <R> IgniteInternalFuture<R> callLocal(@Nullable final Callable<R>
c, byte plc)
+        throws IgniteCheckedException {
         if (c == null)
             return new GridFinishedFuture<>();
 
@@ -969,11 +923,11 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             fut.setWorker(w);
 
             try {
-                pool(plc).execute(w);
+                pools.poolForPolicy(plc).execute(w);
             }
             catch (RejectedExecutionException e) {
                 U.error(log, "Failed to execute worker due to execution rejection " +
-                    "(increase upper bound on " + poolName(plc) + " executor service).",
e);
+                    "(increase upper bound on executor service) [policy=" + plc + ']', e);
 
                 w.run();
             }
@@ -1005,7 +959,7 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @return Future.
      */
     public <R> IgniteInternalFuture<R> callLocalSafe(Callable<R> c, boolean
sys) {
-        return callLocalSafe(c, sys ? GridClosurePolicy.SYSTEM_POOL : GridClosurePolicy.PUBLIC_POOL);
+        return callLocalSafe(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
     }
 
     /**
@@ -1016,7 +970,7 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @param plc Policy to choose executor pool.
      * @return Future.
      */
-    public <R> IgniteInternalFuture<R> callLocalSafe(Callable<R> c, GridClosurePolicy
plc) {
+    private <R> IgniteInternalFuture<R> callLocalSafe(Callable<R> c, byte
plc) {
         try {
             return callLocal(c, plc);
         }

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
new file mode 100644
index 0000000..41e805e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.pool;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.processors.GridProcessorAdapter;
+import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
+import org.apache.ignite.plugin.extensions.communication.IoPool;
+
+import java.util.Arrays;
+import java.util.concurrent.Executor;
+
+/**
+ * Processor which abstracts out thread pool management.
+ */
+public class PoolProcessor extends GridProcessorAdapter {
+    /** Map of {@link IoPool}-s injected by Ignite plugins. */
+    private final IoPool[] extPools = new IoPool[128];
+
+    /**
+     * Constructor.
+     *
+     * @param ctx Kernal context.
+     */
+    public PoolProcessor(GridKernalContext ctx) {
+        super(ctx);
+
+        IgnitePluginProcessor plugins = ctx.plugins();
+
+        if (plugins != null) {
+            // Process custom IO messaging pool extensions:
+            final IoPool[] executorExtensions = ctx.plugins().extensions(IoPool.class);
+
+            if (executorExtensions != null) {
+                // Store it into the map and check for duplicates:
+                for (IoPool ex : executorExtensions) {
+                    final byte id = ex.id();
+
+                    // 1. Check the pool id is non-negative:
+                    if (id < 0)
+                        throw new IgniteException("Failed to register IO executor pool because
its ID is " +
+                            "negative: " + id);
+
+                    // 2. Check the pool id is in allowed range:
+                    if (GridIoPolicy.isReservedGridIoPolicy(id))
+                        throw new IgniteException("Failed to register IO executor pool because
its ID in in the " +
+                            "reserved range: " + id);
+
+                    // 3. Check the pool for duplicates:
+                    if (extPools[id] != null)
+                        throw new IgniteException("Failed to register IO executor pool because
its ID as " +
+                            "already used: " + id);
+
+                    extPools[id] = ex;
+                }
+            }
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void stop(boolean cancel) throws IgniteCheckedException {
+        // Avoid external thread pools GC retention.
+        Arrays.fill(extPools, null);
+    }
+
+    /**
+     * @return P2P pool.
+     */
+    public Executor p2pPool() {
+        return ctx.getPeerClassLoadingExecutorService();
+    }
+
+    /**
+     * Get executor service for policy.
+     *
+     * @param plc Policy.
+     * @return Executor service.
+     * @throws IgniteCheckedException If failed.
+     */
+    public Executor poolForPolicy(byte plc) throws IgniteCheckedException {
+        switch (plc) {
+            case GridIoPolicy.P2P_POOL:
+                return ctx.getPeerClassLoadingExecutorService();
+            case GridIoPolicy.SYSTEM_POOL:
+                return ctx.getSystemExecutorService();
+            case GridIoPolicy.PUBLIC_POOL:
+                return ctx.getExecutorService();
+            case GridIoPolicy.MANAGEMENT_POOL:
+                return ctx.getManagementExecutorService();
+            case GridIoPolicy.AFFINITY_POOL:
+                return ctx.getAffinityExecutorService();
+
+            case GridIoPolicy.UTILITY_CACHE_POOL:
+                assert ctx.utilityCachePool() != null : "Utility cache pool is not configured.";
+
+                return ctx.utilityCachePool();
+
+            case GridIoPolicy.MARSH_CACHE_POOL:
+                assert ctx.marshallerCachePool() != null : "Marshaller cache pool is not
configured.";
+
+                return ctx.marshallerCachePool();
+
+            case GridIoPolicy.IGFS_POOL:
+                assert ctx.getIgfsExecutorService() != null : "IGFS pool is not configured.";
+
+                return ctx.getIgfsExecutorService();
+
+            default: {
+                if (plc < 0)
+                    throw new IgniteCheckedException("Policy cannot be negative: " + plc);
+
+                if (GridIoPolicy.isReservedGridIoPolicy(plc))
+                    throw new IgniteCheckedException("Policy is reserved for internal usage
(range 0-31): " + plc);
+
+                IoPool pool = extPools[plc];
+
+                if (pool == null)
+                    throw new IgniteCheckedException("No pool is registered for policy: "
+ plc);
+
+                assert plc == pool.id();
+
+                Executor res = pool.executor();
+
+                if (res == null)
+                    throw new IgniteCheckedException("Thread pool for policy is null: " +
plc);
+
+                return res;
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index 57762c7..2d8d245 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -948,7 +948,6 @@ org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessorImpl
 org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessorImpl$UserKeyCacheObjectImpl
 org.apache.ignite.internal.processors.clock.GridClockDeltaSnapshotMessage
 org.apache.ignite.internal.processors.clock.GridClockDeltaVersion
-org.apache.ignite.internal.processors.closure.GridClosurePolicy
 org.apache.ignite.internal.processors.closure.GridClosureProcessor$C1
 org.apache.ignite.internal.processors.closure.GridClosureProcessor$C1MLA
 org.apache.ignite.internal.processors.closure.GridClosureProcessor$C1MLAV2

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
index 04eb8ef..328b775 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
 import org.apache.ignite.internal.managers.failover.GridFailoverManager;
 import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
 import org.apache.ignite.internal.managers.swapspace.GridSwapSpaceManager;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.processors.resource.GridResourceProcessor;
 import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
 import org.apache.ignite.resources.LoggerResource;
@@ -68,6 +69,7 @@ public class GridManagerStopSelfTest extends GridCommonAbstractTest {
 
         ctx.config().setPeerClassLoadingEnabled(true);
 
+        ctx.add(new PoolProcessor(ctx));
         ctx.add(new GridResourceProcessor(ctx));
 
         ctx.start();

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
index f44d282..adcd144 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.cluster.ClusterGroupEmptyCheckedException;
 import org.apache.ignite.internal.processors.closure.GridClosureProcessor;
+import org.apache.ignite.internal.processors.pool.PoolProcessor;
 import org.apache.ignite.internal.util.typedef.CI1;
 import org.apache.ignite.internal.util.typedef.CX1;
 import org.apache.ignite.testframework.GridTestUtils;
@@ -174,6 +175,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest
{
         ctx.setExecutorService(Executors.newFixedThreadPool(1));
         ctx.setSystemExecutorService(Executors.newFixedThreadPool(1));
 
+        ctx.add(new PoolProcessor(ctx));
         ctx.add(new GridClosureProcessor(ctx));
 
         ctx.start();
@@ -238,6 +240,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest
{
         ctx.setExecutorService(Executors.newFixedThreadPool(1));
         ctx.setSystemExecutorService(Executors.newFixedThreadPool(1));
 
+        ctx.add(new PoolProcessor(ctx));
         ctx.add(new GridClosureProcessor(ctx));
 
         ctx.start();

http://git-wip-us.apache.org/repos/asf/ignite/blob/f9a0676f/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
index e4afe73..cba67e0 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
@@ -61,6 +61,7 @@ public class GridTestKernalContext extends GridKernalContextImpl {
                 null,
                 null,
                 null,
+                null,
                 U.allPluginProviders());
 
         GridTestUtils.setFieldValue(grid(), "cfg", config());


Mime
View raw message