common/util/src/test/java/org/opendaylight/yangtools/util/concurrent/ThreadPoolExecutorTest.java

   1 /*
   2  * Copyright (c) 2014 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.yangtools.util.concurrent;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.fail;
  12
  13 import java.util.Map;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import java.util.concurrent.ConcurrentMap;
  16 import java.util.concurrent.CountDownLatch;
  17 import java.util.concurrent.ExecutorService;
  18 import java.util.concurrent.RejectedExecutionException;
  19 import java.util.concurrent.TimeUnit;
  20 import java.util.concurrent.atomic.AtomicLong;
  21 import java.util.concurrent.atomic.AtomicReference;
  22
  23 import org.junit.After;
  24 import org.junit.Test;
  25
  26 import com.google.common.base.Stopwatch;
  27
  28 /**
  29  * Tests various ThreadPoolExecutor implementations.
  30  *
  31  * @author Thomas Pantelis
  32  */
  33 public class ThreadPoolExecutorTest {
  34
  35     private ExecutorService executor;
  36
  37     @After
  38     public void tearDown() {
  39         if( executor != null ) {
  40             executor.shutdownNow();
  41         }
  42     }
  43
  44     @Test
  45     public void testFastThreadPoolExecution() throws Exception {
  46
  47         testThreadPoolExecution(
  48                 SpecialExecutors.newBoundedFastThreadPool( 50, 100000, "TestPool" ),
  49                 100000, "TestPool", 0 );
  50     }
  51
  52     @Test(expected=RejectedExecutionException.class)
  53     public void testFastThreadPoolRejectingTask() throws Exception {
  54
  55         executor = SpecialExecutors.newBoundedFastThreadPool( 1, 1, "TestPool" );
  56
  57         for( int i = 0; i < 5; i++ ) {
  58             executor.execute( new Task( null, null, null, null,
  59                     TimeUnit.MICROSECONDS.convert( 5, TimeUnit.SECONDS ) ) );
  60         }
  61     }
  62
  63     @Test
  64     public void testBlockingFastThreadPoolExecution() throws Exception {
  65
  66         // With a queue capacity of 1, it should block at some point.
  67         testThreadPoolExecution(
  68                 SpecialExecutors.newBlockingBoundedFastThreadPool( 2, 1, "TestPool" ),
  69                 1000, null, 10 );
  70     }
  71
  72     @Test
  73     public void testCachedThreadPoolExecution() throws Exception {
  74
  75         testThreadPoolExecution(
  76                 SpecialExecutors.newBoundedCachedThreadPool( 10, 100000, "TestPool" ),
  77                 100000, "TestPool", 0 );
  78     }
  79
  80     @Test(expected=RejectedExecutionException.class)
  81     public void testCachedThreadRejectingTask() throws Exception {
  82
  83         ExecutorService executor = SpecialExecutors.newBoundedCachedThreadPool( 1, 1, "TestPool" );
  84
  85         for( int i = 0; i < 5; i++ ) {
  86             executor.execute( new Task( null, null, null, null,
  87                     TimeUnit.MICROSECONDS.convert( 5, TimeUnit.SECONDS ) ) );
  88         }
  89     }
  90
  91     @Test
  92     public void testBlockingCachedThreadPoolExecution() throws Exception {
  93
  94         testThreadPoolExecution(
  95                 SpecialExecutors.newBlockingBoundedCachedThreadPool( 2, 1, "TestPool" ),
  96                 1000, null, 10 );
  97     }
  98
  99     void testThreadPoolExecution( final ExecutorService executor,
 100             final int numTasksToRun, final String expThreadPrefix, final long taskDelay ) throws Exception {
 101
 102         this.executor = executor;
 103
 104         System.out.println( "\nTesting " + executor.getClass().getSimpleName() + " with " +
 105                 numTasksToRun + " tasks." );
 106
 107         final CountDownLatch tasksRunLatch = new CountDownLatch( numTasksToRun );
 108         final ConcurrentMap<Thread, AtomicLong> taskCountPerThread = new ConcurrentHashMap<>();
 109         final AtomicReference<AssertionError> threadError = new AtomicReference<>();
 110
 111         Stopwatch stopWatch = new Stopwatch();
 112         stopWatch.start();
 113
 114         new Thread() {
 115             @Override
 116             public void run() {
 117                 for( int i = 0; i < numTasksToRun; i++ ) {
 118 //                    if(i%100 == 0) {
 119 //                        Uninterruptibles.sleepUninterruptibly( 20, TimeUnit.MICROSECONDS );
 120 //                    }
 121
 122                     executor.execute( new Task( tasksRunLatch, taskCountPerThread,
 123                                                 threadError, expThreadPrefix, taskDelay ) );
 124                 }
 125             }
 126         }.start();
 127
 128         boolean done = tasksRunLatch.await( 15, TimeUnit.SECONDS );
 129
 130         stopWatch.stop();
 131
 132         if( !done ) {
 133             fail( (numTasksToRun - tasksRunLatch.getCount()) + " tasks out of " +
 134                    numTasksToRun + " executed" );
 135         }
 136
 137         if( threadError.get() != null ) {
 138             throw threadError.get();
 139         }
 140
 141         System.out.println( taskCountPerThread.size() + " threads used:" );
 142         for( Map.Entry<Thread, AtomicLong> e : taskCountPerThread.entrySet() ) {
 143             System.out.println( "  " + e.getKey().getName() + " - " + e.getValue() + " tasks" );
 144         }
 145
 146         System.out.println( "\n" + executor );
 147         System.out.println( "\nElapsed time: " + stopWatch );
 148         System.out.println();
 149     }
 150
 151     static class Task implements Runnable {
 152         final CountDownLatch tasksRunLatch;
 153         final CountDownLatch blockLatch;
 154         final ConcurrentMap<Thread, AtomicLong> taskCountPerThread;
 155         final AtomicReference<AssertionError> threadError;
 156         final String expThreadPrefix;
 157         final long delay;
 158
 159         Task( CountDownLatch tasksRunLatch, ConcurrentMap<Thread, AtomicLong> taskCountPerThread,
 160                 AtomicReference<AssertionError> threadError, String expThreadPrefix, long delay ) {
 161             this.tasksRunLatch = tasksRunLatch;
 162             this.taskCountPerThread = taskCountPerThread;
 163             this.threadError = threadError;
 164             this.expThreadPrefix = expThreadPrefix;
 165             this.delay = delay;
 166             blockLatch = null;
 167         }
 168
 169         Task( CountDownLatch tasksRunLatch, CountDownLatch blockLatch ) {
 170             this.tasksRunLatch = tasksRunLatch;
 171             this.blockLatch = blockLatch;
 172             this.taskCountPerThread = null;
 173             this.threadError = null;
 174             this.expThreadPrefix = null;
 175             this.delay = 0;
 176         }
 177
 178         @Override
 179         public void run() {
 180             try {
 181                 try {
 182                     if( delay > 0 ) {
 183                         TimeUnit.MICROSECONDS.sleep( delay );
 184                     } else if( blockLatch != null ) {
 185                         blockLatch.await();
 186                     }
 187                 } catch( InterruptedException e ) {}
 188
 189                 if( expThreadPrefix != null ) {
 190                     assertEquals( "Thread name starts with " + expThreadPrefix, true,
 191                             Thread.currentThread().getName().startsWith( expThreadPrefix ) );
 192                 }
 193
 194                 if( taskCountPerThread != null ) {
 195                     AtomicLong count = taskCountPerThread.get( Thread.currentThread() );
 196                     if( count == null ) {
 197                         count = new AtomicLong( 0 );
 198                         AtomicLong prev = taskCountPerThread.putIfAbsent( Thread.currentThread(), count );
 199                         if( prev != null ) {
 200                             count = prev;
 201                         }
 202                     }
 203
 204                     count.incrementAndGet();
 205                 }
 206
 207             } catch( AssertionError e ) {
 208                 if( threadError != null ) {
 209                     threadError.set( e );
 210                 }
 211             } finally {
 212                 if( tasksRunLatch != null ) {
 213                     tasksRunLatch.countDown();
 214                 }
 215             }
 216         }
 217     }
 218 }