[GitHub] franz1981 commented on a change in pull request #2524: ARTEMIS-2240 ActiveMQThreadPoolExecutor should use LinkedTransferQueue

[GitBox <gi...@apache.org>]

franz1981 commented on a change in pull request #2524: ARTEMIS-2240 ActiveMQThreadPoolExecutor should use LinkedTransferQueue
URL: https://github.com/apache/activemq-artemis/pull/2524#discussion_r251325505
 
 

 ##########
 File path: artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadPoolExecutor.java
 ##########
 @@ -29,125 +30,58 @@
  * and will be removed after idling for a specified keep time.
  * But in contrast to a standard cached executor, tasks are queued if the
  * maximum pool size if reached, instead of rejected.
- *
- * This is achieved by using a specialized blocking queue, which checks the
- * state of the associated executor in the offer method to decide whether to
- * queue a task or have the executor create another thread.
- *
- * Since the thread pool's execute method is reentrant, more than one caller
- * could try to offer a task into the queue. There is a small chance that
- * (a few) more threads are created as it should be limited by max pool size.
- * To allow for such a case not to reject a task, the underlying thread pool
- * executor is not limited. Only the offer method checks the configured limit.
  */
 public class ActiveMQThreadPoolExecutor extends ThreadPoolExecutor {
 
-   @SuppressWarnings("serial")
-   private static class ThreadPoolQueue extends LinkedBlockingQueue<Runnable> {
+   /**
+    * The default rejected execution handler
+    */
+   private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
 
-      private ActiveMQThreadPoolExecutor executor = null;
+   // Handler executed when a task is submitted and a new thread cannot be created (because maxSize was reached)
+   // It queues the task on the executors's queue (using the add() method, see ThreadPoolQueue class below)
+   private static class QueueExecutionHandler implements RejectedExecutionHandler {
 
-      // keep track of the difference between the number of idle threads and
-      // the number of queued tasks. If the delta is > 0, we have more
-      // idle threads than queued tasks and can add more tasks into the queue.
-      // The delta is incremented if a thread becomes idle or if a task is taken from the queue.
-      // The delta is decremented if a thread leaves idle state or if a task is added to the queue.
-      private static final AtomicIntegerFieldUpdater<ThreadPoolQueue> DELTA_UPDATER = AtomicIntegerFieldUpdater.newUpdater(ThreadPoolQueue.class, "threadTaskDelta");
-      private volatile int threadTaskDelta = 0;
+      private final RejectedExecutionHandler handler;
 
-      public void setExecutor(ActiveMQThreadPoolExecutor executor) {
-         this.executor = executor;
+      private QueueExecutionHandler(RejectedExecutionHandler handler) {
+         Objects.requireNonNull(handler);
+         this.handler = handler;
       }
 
       @Override
-      public boolean offer(Runnable runnable) {
-         boolean retval = false;
-
-         if (threadTaskDelta > 0 || (executor.getPoolSize() >= executor.getMaximumPoolSize())) {
-            // A new task will be added to the queue if the maximum number of threads has been reached
-            // or if the delta is > 0, which means that there are enough idle threads.
-
-            retval = super.offer(runnable);
-
-            // Only decrement the delta if the task has actually been added to the queue
-            if (retval)
-               DELTA_UPDATER.decrementAndGet(this);
+      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+         if (executor.isShutdown() || !executor.getQueue().add(r)) {
+            handler.rejectedExecution(r, executor);
          }
-
-         return retval;
       }
+   }
 
-      @Override
-      public Runnable take() throws InterruptedException {
-         // Increment the delta as a thread becomes idle
-         // by waiting for a task to take from the queue
-         DELTA_UPDATER.incrementAndGet(this);
-
-
-         Runnable runnable = null;
+   // A specialized LinkedBlockingQueue that takes new elements by calling add() but not offer()
 
 Review comment:
   I don't know bud, just looking at the doc of `tryTransfert` is very clear (for a concurrent q doc obviously :P).
   But If you feel that adding some doc could help I will add some for sure :+1: 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services