test/loops/CachedThreadPoolLoops.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class CachedThreadPoolLoops {
    static final AtomicInteger remaining = new AtomicInteger();
    static final int maxIters = 1000000;

    public static void main(String[] args) throws Exception {
        int maxThreads = 100;

        if (args.length > 0) 
            maxThreads = Integer.parseInt(args[0]);
        
        int k = 1;
        for (int i = 1; i <= maxThreads;) {
            System.out.println("Threads:" + i);
            oneTest(i);
            Thread.sleep(100);
            if (i == k) {
                k = i << 1;
                i = i + (i >>> 1);
            } 
            else 
                i = k;
        }
   }

    static void oneTest(int nThreads) throws Exception {
        System.out.print("SynchronousQueue        ");
        oneRun(new SynchronousQueue<Runnable>(), nThreads);
        System.out.print("SynchronousQueue(fair)  ");
        oneRun(new SynchronousQueue<Runnable>(true), nThreads);
    }

    static final class Task implements Runnable {
        final ThreadPoolExecutor pool;
        final CountDownLatch done;
        Task(ThreadPoolExecutor p, CountDownLatch d) { 
            pool = p; 
            done = d;
        }
        public void run() {
            done.countDown();
            remaining.incrementAndGet();
            int n;
            while (!Thread.interrupted() &&
                   (n = remaining.get()) > 0 && 
                   done.getCount() > 0) {
                if (remaining.compareAndSet(n, n-1)) {
                    try {
                        pool.execute(this);
                    }
                    catch (RuntimeException ex) {
                        System.out.print("*");
                        while (done.getCount() > 0) done.countDown();
                        return;
                    }
                }
            }
        }
    }
    
    static void oneRun(BlockingQueue<Runnable> q, int nThreads) throws Exception {
       
        ThreadPoolExecutor pool = 
            new ThreadPoolExecutor(nThreads+1, Integer.MAX_VALUE,
                                   1L, TimeUnit.SECONDS, q);

        CountDownLatch done = new CountDownLatch(maxIters);
        remaining.set(nThreads-1);
        pool.prestartAllCoreThreads();
        Task t = new Task(pool, done);
        long start = System.nanoTime();
        pool.execute(t);
        done.await();
        long time = System.nanoTime() - start;
        System.out.println("\t: " + LoopHelpers.rightJustify(time / maxIters) + " ns per task");
        pool.shutdownNow();
    }

}
Revision:	1.1
Committed:	Tue May 31 15:08:32 2005 UTC (18 years, 11 months ago) by dl
Branch:	MAIN
Log Message:	Add tests
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain. Use, modify, and
4			* redistribute this code in any way without acknowledgement.
5			*/
6
7			import java.util.concurrent.*;
8			import java.util.concurrent.atomic.*;
9
10			public class CachedThreadPoolLoops {
11			static final AtomicInteger remaining = new AtomicInteger();
12			static final int maxIters = 1000000;
13
14			public static void main(String[] args) throws Exception {
15			int maxThreads = 100;
16
17			if (args.length > 0)
18			maxThreads = Integer.parseInt(args[0]);
19
20			int k = 1;
21			for (int i = 1; i <= maxThreads;) {
22			System.out.println("Threads:" + i);
23			oneTest(i);
24			Thread.sleep(100);
25			if (i == k) {
26			k = i << 1;
27			i = i + (i >>> 1);
28			}
29			else
30			i = k;
31			}
32			}
33
34			static void oneTest(int nThreads) throws Exception {
35			System.out.print("SynchronousQueue ");
36			oneRun(new SynchronousQueue<Runnable>(), nThreads);
37			System.out.print("SynchronousQueue(fair) ");
38			oneRun(new SynchronousQueue<Runnable>(true), nThreads);
39			}
40
41			static final class Task implements Runnable {
42			final ThreadPoolExecutor pool;
43			final CountDownLatch done;
44			Task(ThreadPoolExecutor p, CountDownLatch d) {
45			pool = p;
46			done = d;
47			}
48			public void run() {
49			done.countDown();
50			remaining.incrementAndGet();
51			int n;
52			while (!Thread.interrupted() &&
53			(n = remaining.get()) > 0 &&
54			done.getCount() > 0) {
55			if (remaining.compareAndSet(n, n-1)) {
56			try {
57			pool.execute(this);
58			}
59			catch (RuntimeException ex) {
60			System.out.print("*");
61			while (done.getCount() > 0) done.countDown();
62			return;
63			}
64			}
65			}
66			}
67			}
68
69			static void oneRun(BlockingQueue<Runnable> q, int nThreads) throws Exception {
70
71			ThreadPoolExecutor pool =
72			new ThreadPoolExecutor(nThreads+1, Integer.MAX_VALUE,
73			1L, TimeUnit.SECONDS, q);
74
75			CountDownLatch done = new CountDownLatch(maxIters);
76			remaining.set(nThreads-1);
77			pool.prestartAllCoreThreads();
78			Task t = new Task(pool, done);
79			long start = System.nanoTime();
80			pool.execute(t);
81			done.await();
82			long time = System.nanoTime() - start;
83			System.out.println("\t: " + LoopHelpers.rightJustify(time / maxIters) + " ns per task");
84			pool.shutdownNow();
85			}
86
87			}