test/loops/TimeoutExchangerLoops.java

/*
 * Written by Bill Scherer and 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.*;
import java.util.concurrent.locks.*;

public class TimeoutExchangerLoops {
    static final int NCPUS = Runtime.getRuntime().availableProcessors();

    static final int  DEFAULT_THREADS = NCPUS + 2;
    static final long DEFAULT_PATIENCE_NANOS = 500000;
    static final long DEFAULT_TRIAL_MILLIS   = 10000;

    public static void main(String[] args) throws Exception {
        int maxThreads = DEFAULT_THREADS;
        long trialMillis = DEFAULT_TRIAL_MILLIS;
        long patienceNanos = DEFAULT_PATIENCE_NANOS;
        int nReps = 3;

        // Parse and check args
        int argc = 0;
        while (argc < args.length) {
            String option = args[argc++];
            if (option.equals("-t"))
                trialMillis = Integer.parseInt(args[argc]);
            else if (option.equals("-p"))
                patienceNanos = Long.parseLong(args[argc]);
            else if (option.equals("-r"))
                nReps = Integer.parseInt(args[argc]);
            else 
                maxThreads = Integer.parseInt(option);
            argc++;
        }

        // Display runtime parameters
        System.out.print("TimeoutExchangerTest");
        System.out.print(" -t " + trialMillis);
        System.out.print(" -p " + patienceNanos);
        System.out.print(" -r " + nReps);
        System.out.print(" max threads " + maxThreads);
        System.out.println();

        System.out.println("Warmups..");
        long warmupTime = 1000;
        long sleepTime = 500;
        if (false) {
            for (int k = 0; k < 10; ++k) {
                for (int j = 0; j < 10; ++j) {
                    oneRun(2, (j + 1) * 1000, patienceNanos);
                    Thread.sleep(sleepTime);
                }
            }
        }

        oneRun(3, warmupTime, patienceNanos);
        Thread.sleep(sleepTime);

        for (int i = maxThreads; i >= 2; i -= 1) {
            oneRun(i, warmupTime, patienceNanos);
            Thread.sleep(sleepTime);
        }

        for (int j = 0; j < nReps; ++j) {
            System.out.println("Replication " + j);
            for (int i = 2; i <= maxThreads; i += 2) {
                oneRun(i, trialMillis, patienceNanos);
                Thread.sleep(sleepTime);
            }
        }
    }

    static void oneRun(int nThreads, long trialMillis, long patienceNanos) 
        throws Exception {
        System.out.printf("%4d threads", nThreads);
        System.out.printf("%9dms", trialMillis);
        final CountDownLatch start = new CountDownLatch(1);
        Exchanger x = new Exchanger();
        Runner[] runners = new Runner[nThreads];
        Thread[] threads = new Thread[nThreads];
        for (int i = 0; i < nThreads; ++i) {
            runners[i] = new Runner(x, patienceNanos, start);
            threads[i] = new Thread(runners[i]);
            threads[i].start();
        }
        long startTime = System.nanoTime();
        start.countDown();
        Thread.sleep(trialMillis);
        for (int i = 0; i < nThreads; ++i) 
            threads[i].interrupt();
        long elapsed = System.nanoTime() - startTime;
        for (int i = 0; i < nThreads; ++i) 
            threads[i].join();
        int iters = 0;
        long fails = 0;
        for (int i = 0; i < nThreads; ++i) {
            iters += runners[i].iters;
            fails += runners[i].failures;
        }
        if (iters <= 0) iters = 1;
        long rate = iters * 1000L * 1000L * 1000L / elapsed;
        long npt = elapsed / iters;
        double failRate = (fails * 100.0) / (double)iters;
        System.out.printf("%9d it/s ", rate);
        System.out.printf("%9d ns/it", npt);
        System.out.printf("%9.5f%% fails", failRate);
        System.out.println();
        //        x.printStats();
    }


    static final class Runner implements Runnable {
        final Exchanger exchanger;
        final CountDownLatch start;
        final long patience;
        volatile int iters;
        volatile int failures;
        Runner(Exchanger x, long patience, CountDownLatch start) {
            this.exchanger = x;
            this.patience = patience;
            this.start = start;
        }

        public void run() {
            int i = 0;
            try {
                Exchanger x = exchanger;
                Object m = new Integer(17);
                long p = patience;
                start.await();
                for (;;) {
                    try {
                        Object e = x.exchange(m, p, TimeUnit.NANOSECONDS);
                        if (e == null || e == m)
                            throw new Error();
                        m = e;
                        ++i;
                    } catch (TimeoutException to) {
                        if (Thread.interrupted()) {
                            iters = i;
                            return;
                        }
                        ++i;
                        ++failures;
                    }
                }
            } catch (InterruptedException ie) {
                iters = i;
            }
        }
    }
}
        
Revision:	1.2
Committed:	Mon Feb 13 12:39:23 2006 UTC (18 years, 3 months ago) by dl
Branch:	MAIN
Changes since 1.1:	+95 -90 lines
Log Message:	Update exchanger tests
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Bill Scherer and Doug Lea with assistance from members
3			* of JCP JSR-166 Expert Group and released to the public domain. Use,
4			* modify, and redistribute this code in any way without
5			* acknowledgement.
6			*/
7
8			import java.util.concurrent.*;
9			import java.util.concurrent.atomic.*;
10			import java.util.concurrent.locks.*;
11
12			public class TimeoutExchangerLoops {
13	dl	1.2	static final int NCPUS = Runtime.getRuntime().availableProcessors();
14
15			static final int DEFAULT_THREADS = NCPUS + 2;
16	dl	1.1	static final long DEFAULT_PATIENCE_NANOS = 500000;
17	dl	1.2	static final long DEFAULT_TRIAL_MILLIS = 10000;
18	dl	1.1
19			public static void main(String[] args) throws Exception {
20			int maxThreads = DEFAULT_THREADS;
21			long trialMillis = DEFAULT_TRIAL_MILLIS;
22			long patienceNanos = DEFAULT_PATIENCE_NANOS;
23	dl	1.2	int nReps = 3;
24	dl	1.1
25			// Parse and check args
26			int argc = 0;
27	dl	1.2	while (argc < args.length) {
28			String option = args[argc++];
29			if (option.equals("-t"))
30			trialMillis = Integer.parseInt(args[argc]);
31			else if (option.equals("-p"))
32			patienceNanos = Long.parseLong(args[argc]);
33			else if (option.equals("-r"))
34			nReps = Integer.parseInt(args[argc]);
35			else
36			maxThreads = Integer.parseInt(option);
37			argc++;
38	dl	1.1	}
39
40			// Display runtime parameters
41			System.out.print("TimeoutExchangerTest");
42			System.out.print(" -t " + trialMillis);
43			System.out.print(" -p " + patienceNanos);
44	dl	1.2	System.out.print(" -r " + nReps);
45	dl	1.1	System.out.print(" max threads " + maxThreads);
46			System.out.println();
47
48	dl	1.2	System.out.println("Warmups..");
49			long warmupTime = 1000;
50			long sleepTime = 500;
51			if (false) {
52			for (int k = 0; k < 10; ++k) {
53			for (int j = 0; j < 10; ++j) {
54			oneRun(2, (j + 1) * 1000, patienceNanos);
55			Thread.sleep(sleepTime);
56			}
57			}
58			}
59
60			oneRun(3, warmupTime, patienceNanos);
61			Thread.sleep(sleepTime);
62
63			for (int i = maxThreads; i >= 2; i -= 1) {
64			oneRun(i, warmupTime, patienceNanos);
65			Thread.sleep(sleepTime);
66			}
67
68			for (int j = 0; j < nReps; ++j) {
69			System.out.println("Replication " + j);
70			for (int i = 2; i <= maxThreads; i += 2) {
71			oneRun(i, trialMillis, patienceNanos);
72			Thread.sleep(sleepTime);
73			}
74	dl	1.1	}
75			}
76
77			static void oneRun(int nThreads, long trialMillis, long patienceNanos)
78			throws Exception {
79	dl	1.2	System.out.printf("%4d threads", nThreads);
80			System.out.printf("%9dms", trialMillis);
81			final CountDownLatch start = new CountDownLatch(1);
82			Exchanger x = new Exchanger();
83	dl	1.1	Runner[] runners = new Runner[nThreads];
84	dl	1.2	Thread[] threads = new Thread[nThreads];
85			for (int i = 0; i < nThreads; ++i) {
86			runners[i] = new Runner(x, patienceNanos, start);
87			threads[i] = new Thread(runners[i]);
88			threads[i].start();
89			}
90			long startTime = System.nanoTime();
91			start.countDown();
92			Thread.sleep(trialMillis);
93	dl	1.1	for (int i = 0; i < nThreads; ++i)
94	dl	1.2	threads[i].interrupt();
95			long elapsed = System.nanoTime() - startTime;
96	dl	1.1	for (int i = 0; i < nThreads; ++i)
97	dl	1.2	threads[i].join();
98			int iters = 0;
99	dl	1.1	long fails = 0;
100			for (int i = 0; i < nThreads; ++i) {
101	dl	1.2	iters += runners[i].iters;
102	dl	1.1	fails += runners[i].failures;
103			}
104	dl	1.2	if (iters <= 0) iters = 1;
105			long rate = iters * 1000L * 1000L * 1000L / elapsed;
106			long npt = elapsed / iters;
107	dl	1.1	double failRate = (fails * 100.0) / (double)iters;
108	dl	1.2	System.out.printf("%9d it/s ", rate);
109			System.out.printf("%9d ns/it", npt);
110			System.out.printf("%9.5f%% fails", failRate);
111	dl	1.1	System.out.println();
112	dl	1.2	// x.printStats();
113	dl	1.1	}
114
115	dl	1.2
116	dl	1.1	static final class Runner implements Runnable {
117	dl	1.2	final Exchanger exchanger;
118			final CountDownLatch start;
119	dl	1.1	final long patience;
120	dl	1.2	volatile int iters;
121			volatile int failures;
122			Runner(Exchanger x, long patience, CountDownLatch start) {
123			this.exchanger = x;
124	dl	1.1	this.patience = patience;
125	dl	1.2	this.start = start;
126	dl	1.1	}
127
128			public void run() {
129	dl	1.2	int i = 0;
130	dl	1.1	try {
131	dl	1.2	Exchanger x = exchanger;
132			Object m = new Integer(17);
133			long p = patience;
134			start.await();
135			for (;;) {
136	dl	1.1	try {
137	dl	1.2	Object e = x.exchange(m, p, TimeUnit.NANOSECONDS);
138			if (e == null \|\| e == m)
139			throw new Error();
140			m = e;
141	dl	1.1	++i;
142			} catch (TimeoutException to) {
143	dl	1.2	if (Thread.interrupted()) {
144			iters = i;
145			return;
146			}
147			++i;
148			++failures;
149	dl	1.1	}
150	dl	1.2	}
151			} catch (InterruptedException ie) {
152			iters = i;
153	dl	1.1	}
154			}
155			}
156			}
157