test/loops/CachedThreadPoolLoops.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 */

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]);

        System.out.print("Warmup:");
        for (int j = 0; j < 2; ++j) {
            int k = 1;
            for (int i = 1; i <= maxThreads;) {
                System.out.print(" " + i);
                oneTest(i, 10000, false);
                Thread.sleep(100);
                if (i == k) {
                    k = i << 1;
                    i = i + (i >>> 1);
                } 
                else 
                    i = k;
            }
        }
        System.out.println();

        int k = 1;
        for (int i = 1; i <= maxThreads;) {
            System.out.println("Threads:" + i);
            oneTest(i, maxIters, true);
            Thread.sleep(100);
            if (i == k) {
                k = i << 1;
                i = i + (i >>> 1);
            } 
            else 
                i = k;
        }
   }

    static void oneTest(int nThreads, int iters, boolean print) throws Exception {
        //        if (print) System.out.print("LinkedBlockingQueue     ");
        //        oneRun(new LinkedBlockingQueue<Runnable>(256), nThreads, iters, print);
        //        if (print) System.out.print("ArrayBlockingQueue      ");
        //        oneRun(new ArrayBlockingQueue<Runnable>(256), nThreads, iters, print);
        if (print) System.out.print("SynchronousQueue        ");
        oneRun(new SynchronousQueue<Runnable>(false), nThreads, iters, print);
        if (print) System.out.print("SynchronousQueue(fair)  ");
        oneRun(new SynchronousQueue<Runnable>(true), nThreads, iters, print);
    }

    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, int iters, boolean print) throws Exception {
       
        ThreadPoolExecutor pool = 
            new ThreadPoolExecutor(nThreads+1, Integer.MAX_VALUE,
                                   1L, TimeUnit.SECONDS, q);

        CountDownLatch done = new CountDownLatch(iters);
        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;
        if (print)
            System.out.println("\t: " + LoopHelpers.rightJustify(time / iters) + " ns per task");
        q.clear();
        Thread.sleep(100);
        pool.shutdown();
        Thread.sleep(100);
        pool.shutdownNow();
    }

}
Revision:	1.3
Committed:	Mon Feb 19 00:46:06 2007 UTC (17 years, 3 months ago) by dl
Branch:	MAIN
Changes since 1.2:	+6 -2 lines
Log Message:	Uniform headers
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3	dl	1.3	* Expert Group and released to the public domain, as explained at
4			* http://creativecommons.org/licenses/publicdomain
5	dl	1.1	*/
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	dl	1.2
20			System.out.print("Warmup:");
21			for (int j = 0; j < 2; ++j) {
22			int k = 1;
23			for (int i = 1; i <= maxThreads;) {
24			System.out.print(" " + i);
25			oneTest(i, 10000, false);
26			Thread.sleep(100);
27			if (i == k) {
28			k = i << 1;
29			i = i + (i >>> 1);
30			}
31			else
32			i = k;
33			}
34			}
35			System.out.println();
36
37	dl	1.1	int k = 1;
38			for (int i = 1; i <= maxThreads;) {
39			System.out.println("Threads:" + i);
40	dl	1.2	oneTest(i, maxIters, true);
41	dl	1.1	Thread.sleep(100);
42			if (i == k) {
43			k = i << 1;
44			i = i + (i >>> 1);
45			}
46			else
47			i = k;
48			}
49			}
50
51	dl	1.2	static void oneTest(int nThreads, int iters, boolean print) throws Exception {
52	dl	1.3	// if (print) System.out.print("LinkedBlockingQueue ");
53			// oneRun(new LinkedBlockingQueue<Runnable>(256), nThreads, iters, print);
54			// if (print) System.out.print("ArrayBlockingQueue ");
55			// oneRun(new ArrayBlockingQueue<Runnable>(256), nThreads, iters, print);
56	dl	1.2	if (print) System.out.print("SynchronousQueue ");
57			oneRun(new SynchronousQueue<Runnable>(false), nThreads, iters, print);
58			if (print) System.out.print("SynchronousQueue(fair) ");
59			oneRun(new SynchronousQueue<Runnable>(true), nThreads, iters, print);
60	dl	1.1	}
61
62			static final class Task implements Runnable {
63			final ThreadPoolExecutor pool;
64			final CountDownLatch done;
65			Task(ThreadPoolExecutor p, CountDownLatch d) {
66			pool = p;
67			done = d;
68			}
69			public void run() {
70			done.countDown();
71			remaining.incrementAndGet();
72			int n;
73			while (!Thread.interrupted() &&
74			(n = remaining.get()) > 0 &&
75			done.getCount() > 0) {
76			if (remaining.compareAndSet(n, n-1)) {
77			try {
78			pool.execute(this);
79			}
80			catch (RuntimeException ex) {
81			System.out.print("*");
82			while (done.getCount() > 0) done.countDown();
83			return;
84			}
85			}
86			}
87			}
88			}
89
90	dl	1.2	static void oneRun(BlockingQueue<Runnable> q, int nThreads, int iters, boolean print) throws Exception {
91	dl	1.1
92			ThreadPoolExecutor pool =
93			new ThreadPoolExecutor(nThreads+1, Integer.MAX_VALUE,
94			1L, TimeUnit.SECONDS, q);
95
96	dl	1.2	CountDownLatch done = new CountDownLatch(iters);
97	dl	1.1	remaining.set(nThreads-1);
98			pool.prestartAllCoreThreads();
99			Task t = new Task(pool, done);
100			long start = System.nanoTime();
101			pool.execute(t);
102			done.await();
103			long time = System.nanoTime() - start;
104	dl	1.2	if (print)
105			System.out.println("\t: " + LoopHelpers.rightJustify(time / iters) + " ns per task");
106			q.clear();
107			Thread.sleep(100);
108			pool.shutdown();
109			Thread.sleep(100);
110	dl	1.1	pool.shutdownNow();
111			}
112
113			}