deltaspike-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmannibu...@apache.org
Subject [1/2] deltaspike git commit: DELTASPIKE-1302 configuration for ThreadPoolManager
Date Sun, 26 Nov 2017 16:12:22 GMT
Repository: deltaspike
Updated Branches:
  refs/heads/master d1316e517 -> 8e27c82fe


DELTASPIKE-1302 configuration for ThreadPoolManager


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

Branch: refs/heads/master
Commit: c4006cc3b5fda1eea5551fee5a0a7e15c19264d9
Parents: d1316e5
Author: Romain Manni-Bucau <rmannibucau@apache.org>
Authored: Sun Nov 26 15:31:46 2017 +0100
Committer: Romain Manni-Bucau <rmannibucau@apache.org>
Committed: Sun Nov 26 15:31:46 2017 +0100

----------------------------------------------------------------------
 .../core/impl/future/ThreadPoolManager.java     | 195 ++++++++++++++++---
 .../core/impl/future/ThreadPoolManagerTest.java | 116 +++++++++++
 documentation/src/main/asciidoc/core.adoc       |  26 +++
 3 files changed, 312 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/deltaspike/blob/c4006cc3/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/ThreadPoolManager.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/ThreadPoolManager.java
b/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/ThreadPoolManager.java
index 9dc6667..4e1f2a1 100644
--- a/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/ThreadPoolManager.java
+++ b/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/ThreadPoolManager.java
@@ -18,23 +18,48 @@
  */
 package org.apache.deltaspike.core.impl.future;
 
+import org.apache.deltaspike.core.api.config.ConfigResolver;
 import org.apache.deltaspike.core.api.config.base.CoreBaseConfig;
 
 import javax.annotation.PreDestroy;
 import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.context.spi.CreationalContext;
+import javax.enterprise.inject.spi.Bean;
+import javax.enterprise.inject.spi.BeanManager;
+import javax.inject.Inject;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+import static java.util.Arrays.asList;
+import static java.util.Locale.ENGLISH;
+
 @ApplicationScoped
 public class ThreadPoolManager
 {
     private final ConcurrentMap<String, ExecutorService> pools = new ConcurrentHashMap<String,
ExecutorService>();
-    private volatile ExecutorService defaultPool;
+    private final Collection<CreationalContext<?>> contexts = new ArrayList<CreationalContext<?>>(8);
     private volatile boolean closed = false;
 
+    @Inject
+    private BeanManager beanManager;
+
     @PreDestroy
     private void shutdown()
     {
@@ -43,21 +68,12 @@ public class ThreadPoolManager
         for (final ExecutorService es : pools.values())
         {
             es.shutdown();
-            try
-            {
-                es.awaitTermination(timeout, TimeUnit.MILLISECONDS);
-            }
-            catch (final InterruptedException e)
-            {
-                Thread.interrupted();
-            }
         }
-        if (defaultPool != null)
+        for (final ExecutorService es : pools.values())
         {
-            defaultPool.shutdown();
             try
             {
-                defaultPool.awaitTermination(timeout, TimeUnit.MILLISECONDS);
+                es.awaitTermination(timeout, TimeUnit.MILLISECONDS);
             }
             catch (final InterruptedException e)
             {
@@ -65,6 +81,12 @@ public class ThreadPoolManager
             }
         }
         pools.clear();
+
+        for (final CreationalContext<?> ctx : contexts)
+        {
+            ctx.release();
+        }
+        contexts.clear();
     }
 
     public ExecutorService find(final String name)
@@ -76,25 +98,148 @@ public class ThreadPoolManager
         ExecutorService pool = pools.get(name);
         if (pool == null)
         {
-            ensureDefaultPool();
-            pool = defaultPool;
-        }
-        return pool;
-    }
-
-    private void ensureDefaultPool()
-    {
-        if (defaultPool == null)
-        {
             synchronized (this)
             {
-                if (defaultPool == null)
+                pool = pools.get(name);
+                if (pool == null)
                 {
-                    defaultPool = Executors.newFixedThreadPool(
-                            Math.max(2, Runtime.getRuntime().availableProcessors()));
+                    // the instantiation does the following:
+                    // 1. check if there is a named bean matching this name using @Default
qualifier
+                    // 2. check if there is a JNDI entry (ManagedExecutorService case) matching
this name
+                    // 3. create a new executor service based on the DS-config
+
+                    // 1.
+                    final Set<Bean<?>> beans = beanManager.getBeans(name);
+                    if (beans != null && !beans.isEmpty())
+                    {
+                        final Bean<?> bean = beanManager.resolve(beans);
+                        if (bean.getTypes().contains(ExecutorService.class))
+                        {
+                            final CreationalContext<Object> creationalContext = beanManager.createCreationalContext(null);
+                            if (!beanManager.isNormalScope(bean.getScope()))
+                            {
+                                contexts.add(creationalContext);
+                            }
+                            pool = ExecutorService.class.cast(beanManager.getReference(bean,
ExecutorService.class, creationalContext));
+                        }
+                    }
+
+                    if (pool == null) // 2.
+                    {
+                        for (final String prefix : asList(
+                                "", "java:app/", "java:global/", "java:global/threads/",
"java:global/deltaspike/", "java:"))
+                        {
+                            try {
+                                final Object instance = new InitialContext().lookup(prefix
+ name);
+                                if (ExecutorService.class.isInstance(instance))
+                                {
+                                    pool = ExecutorService.class.cast(instance);
+                                    break;
+                                }
+                            } catch (final NamingException e) {
+                                // no-op
+                            }
+                        }
+                    }
+
+                    if (pool == null) // 3.
+                    {
+                        final String configPrefix = "futurable.pool." + name + ".";
+                        final int coreSize = ConfigResolver.resolve(configPrefix + "coreSize")
+                                .as(Integer.class)
+                                .withDefault(Math.max(2, Runtime.getRuntime().availableProcessors()))
+                                .getValue();
+                        final int maxSize = ConfigResolver.resolve(configPrefix + "maxSize")
+                                .as(Integer.class)
+                                .withDefault(coreSize)
+                                .getValue();
+                        final long keepAlive = ConfigResolver.resolve(configPrefix + "keepAlive.value")
+                                .as(Long.class)
+                                .withDefault(0L)
+                                .getValue();
+                        final String keepAliveUnit = ConfigResolver.resolve(configPrefix
+ "keepAlive.unit")
+                                .as(String.class)
+                                .withDefault("MILLISECONDS")
+                                .getValue();
+
+                        final String queueType = ConfigResolver.resolve(configPrefix + "queue.type")
+                                .as(String.class)
+                                .withDefault("LINKED")
+                                .getValue();
+                        final BlockingQueue<Runnable> queue;
+                        if ("ARRAY".equalsIgnoreCase(queueType))
+                        {
+                            final int size = ConfigResolver.resolve(configPrefix + "queue.size")
+                                    .as(Integer.class)
+                                    .withDefault(1024)
+                                    .getValue();
+                            final boolean fair = ConfigResolver.resolve(configPrefix + "queue.fair")
+                                    .as(Boolean.class)
+                                    .withDefault(false)
+                                    .getValue();
+                            queue = new ArrayBlockingQueue<Runnable>(size, fair);
+                        }
+                        else if ("SYNCHRONOUS".equalsIgnoreCase(queueType))
+                        {
+                            final boolean fair = ConfigResolver.resolve(configPrefix + "queue.fair")
+                                    .as(Boolean.class)
+                                    .withDefault(false)
+                                    .getValue();
+                            queue = new SynchronousQueue<Runnable>(fair);
+                        }
+                        else/* (queueType.equalsIgnoreCase("LINKED")) */
+                        {
+                            final int capacity = ConfigResolver.resolve(configPrefix + "queue.capacity")
+                                    .as(Integer.class)
+                                    .withDefault(Integer.MAX_VALUE)
+                                    .getValue();
+                            queue = new LinkedBlockingQueue<Runnable>(capacity);
+                        }
+
+                        final String threadFactoryName = ConfigResolver.getPropertyValue(configPrefix
+ "threadFactory.name");
+                        final ThreadFactory threadFactory;
+                        if (threadFactoryName != null)
+                        {
+                            threadFactory = lookupByName(threadFactoryName, ThreadFactory.class);
+                        }
+                        else
+                        {
+                            threadFactory = Executors.defaultThreadFactory();
+                        }
+
+                        final String rejectedHandlerName = ConfigResolver.getPropertyValue(configPrefix
+ "rejectedExecutionHandler.name");
+                        final RejectedExecutionHandler rejectedHandler;
+                        if (rejectedHandlerName != null)
+                        {
+                            rejectedHandler = lookupByName(rejectedHandlerName, RejectedExecutionHandler.class);
+                        }
+                        else
+                        {
+                            rejectedHandler = new ThreadPoolExecutor.AbortPolicy();
+                        }
+
+                        pool = new ThreadPoolExecutor(
+                                coreSize, maxSize,
+                                keepAlive, TimeUnit.valueOf(keepAliveUnit),
+                                queue, threadFactory, rejectedHandler);
+                    }
+
+                    pools.put(name, pool);
                 }
             }
         }
+        return pool;
+    }
+
+    private <T> T lookupByName(final String name, final Class<T> type) {
+        final Set<Bean<?>> tfb = beanManager.getBeans(name);
+        final Bean<?> bean = beanManager.resolve(tfb);
+        final CreationalContext<?> ctx = beanManager.createCreationalContext(null);
+        if (!beanManager.isNormalScope(bean.getScope()))
+        {
+            contexts.add(ctx);
+        }
+        return type.cast(beanManager.getReference(bean, type, ctx));
     }
 }
 

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/c4006cc3/deltaspike/core/impl/src/test/java/org/apache/deltaspike/core/impl/future/ThreadPoolManagerTest.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/test/java/org/apache/deltaspike/core/impl/future/ThreadPoolManagerTest.java
b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/core/impl/future/ThreadPoolManagerTest.java
new file mode 100644
index 0000000..111dda4
--- /dev/null
+++ b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/core/impl/future/ThreadPoolManagerTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.deltaspike.core.impl.future;
+
+import org.apache.deltaspike.core.api.config.ConfigResolver;
+import org.apache.deltaspike.core.impl.config.PropertiesConfigSource;
+import org.apache.deltaspike.core.impl.scope.conversation.ConversationBeanHolder;
+import org.apache.deltaspike.core.impl.scope.viewaccess.ViewAccessBeanAccessHistory;
+import org.apache.deltaspike.core.impl.scope.viewaccess.ViewAccessBeanHolder;
+import org.apache.deltaspike.core.impl.scope.viewaccess.ViewAccessViewHistory;
+import org.apache.deltaspike.core.impl.scope.window.DefaultWindowContextQuotaHandler;
+import org.apache.deltaspike.core.impl.scope.window.WindowBeanHolder;
+import org.apache.deltaspike.core.impl.scope.window.WindowContextProducer;
+import org.apache.deltaspike.core.impl.scope.window.WindowContextQuotaHandlerCache;
+import org.apache.deltaspike.core.impl.scope.window.WindowIdHolder;
+import org.apache.deltaspike.core.spi.config.ConfigSource;
+import org.jboss.arquillian.container.test.api.Deployment;
+import org.jboss.arquillian.junit.Arquillian;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.EmptyAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.jboss.shrinkwrap.api.spec.WebArchive;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.inject.Inject;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+@RunWith(Arquillian.class)
+public class ThreadPoolManagerTest
+{
+    @Deployment
+    public static WebArchive deploy()
+    {
+        return ShrinkWrap.create(WebArchive.class, "ThreadPoolManagerTest.war")
+                .addAsLibraries(ShrinkWrap.create(JavaArchive.class, "deltaspike-core-fake.jar")
+                    .addClasses(
+                            ThreadPoolManager.class,
+                            WindowContextProducer.class, WindowBeanHolder.class, WindowIdHolder.class,
+                            DefaultWindowContextQuotaHandler.class, WindowContextQuotaHandlerCache.class,
+                            ConversationBeanHolder.class, ViewAccessBeanHolder.class,
+                            ViewAccessBeanAccessHistory.class, ViewAccessViewHistory.class)
+                    .addAsManifestResource(EmptyAsset.INSTANCE, "beans.xml"))
+                .addAsWebInfResource(EmptyAsset.INSTANCE, "beans.xml");
+    }
+
+    @Inject
+    private ThreadPoolManager manager;
+
+    @Test
+    public void defaultPool() throws ExecutionException, InterruptedException
+    {
+        final ExecutorService auto = manager.find("auto");
+        assertEquals(auto, auto);
+        assertSame(auto, auto);
+        assertEquals(Runtime.getRuntime().availableProcessors(), ThreadPoolExecutor.class.cast(auto).getCorePoolSize());
+        assertUsable(auto);
+    }
+
+    @Test // this test validates we read the config properly but also it is lazy
+    public void configuredPool() throws ExecutionException, InterruptedException
+    {
+        ConfigResolver.addConfigSources(Collections.<ConfigSource>singletonList(new
PropertiesConfigSource(new Properties()
+        {{
+            setProperty("futurable.pool.custom.coreSize", "5");
+        }})
+        {
+            @Override
+            public String getConfigName()
+            {
+                return "configuredPool";
+            }
+        }));
+        final ExecutorService custom = manager.find("custom");
+        assertEquals(custom, custom);
+        assertSame(custom, custom);
+        assertEquals(5, ThreadPoolExecutor.class.cast(custom).getCorePoolSize());
+        assertUsable(custom);
+    }
+
+    private void assertUsable(final ExecutorService pool) throws InterruptedException, ExecutionException
+    {
+        assertEquals("ok", pool.submit(new Callable<String>()
+        {
+            @Override
+            public String call() throws Exception
+            {
+                return "ok";
+            }
+        }).get());
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/c4006cc3/documentation/src/main/asciidoc/core.adoc
----------------------------------------------------------------------
diff --git a/documentation/src/main/asciidoc/core.adoc b/documentation/src/main/asciidoc/core.adoc
index 4da3c9f..9fd5dce 100644
--- a/documentation/src/main/asciidoc/core.adoc
+++ b/documentation/src/main/asciidoc/core.adoc
@@ -1182,6 +1182,32 @@ DeltaSpike provides support for executing code in an asynchronous manner.
 The b
 - `@Locked` - Ability to prevent concurrent access to a method based on its usage of reads/writes.
 - `@Throttled` - Ability to limit how frequently a method can be invoked.
 
+== @Futureable configuration
+
+The strategy to find the `ExecutorService` associated to the `@Futureable` name is the following
one:
+
+1. Check if there is a CDI bean of type `ExecutorService` with the name of the pool, if not
try 2
+2. Check if there is a JNDI entry matching the pool name directly or prefixed with `java:app/`,
`java:global/`, `java:global/threads/`, `java:global/deltaspike/`, `java:`, if not try 3
+3. Read the configuration and create a `ThreadPoolExecutor`
+
+IMPORTANT: the instance is looked up only once so from the first time it was read you can't
change any configuration anymore.
+
+If you rely on the last option (configured executor) here are the keys you can set in DeltaSpike
configuration:
+
+|===
+| Key | Description | Default
+| futurable.pool.<pool name>.coreSize | The core size of the pool. | Number of available
processors
+| futurable.pool.<pool name>.maxSize | The max size of the pool. | coreSize value
+| futurable.pool.<pool name>.keepAlive.value | Pool keep alive (when a thread is released).
| 0
+| futurable.pool.<pool name>.keepAlive.unit | Unit of keepAlive.value. It must match
a `TIMEUNIT` name. | MILLISECONDS
+| futurable.pool.<pool name>.queue.type | The task queue type of the executor. Can
be `ARRAY` to use an `ArrayBlockingQueue`, `LINKED` for a `LinkedBlockingQueue` or `SYNCHRONOUS`
for a `SynchronousQueue`. | LINKED
+| futurable.pool.<pool name>.queue.fair | For synchronous and array queue types, if
the queue is fair. | false
+| futurable.pool.<pool name>.queue.size | For array queue type, the size of the queue.
| 1024
+| futurable.pool.<pool name>.queue.capacity | For linked queue type, the capacity of
the queue. | `Integer.MAX_VALUE`
+| futurable.pool.<pool name>.threadFactory.name | If set a CDI bean matching the value
will be looked up and used as `ThreadFactory`. | none, `Executors.defaultThreadFactory()`
is used
+| futurable.pool.<pool name>.rejectedExecutionHandler.name | If set a CDI bean matching
the value will be looked up and used as `RejectedExecutionHandler`. | none, `ThreadPoolExecutor.AbortPolicy`
is used
+|===
+
 == Utilities
 
 DeltaSpike provides many utility classes (no constructor / static


Mime
View raw message