test/loops/RLIBar.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/publicdomain/zero/1.0/
 */
// Adapted from code that was in turn
// Derived from SocketPerformanceTest.java - BugID: 4763450
//
//

import java.io.*;
import java.net.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;

public class RLIBar {

    static int batchLimit;
    static int mseq;
    static int nReady;
    static int ExThreads;
    static int ASum;
    static final ReentrantLock Gate = new ReentrantLock();
    static final Condition GateCond = Gate.newCondition();

    static final ReentrantLock HoldQ = new ReentrantLock();
    static final Condition HoldQCond = HoldQ.newCondition();
    static boolean Hold = false;
    static int HoldPop;
    static int HoldLimit;

    private static boolean HoldCheck() {
        try {
            HoldQ.lock();
            try {
                if (!Hold) return false;
                else {
                    ++HoldPop;
                    if (HoldPop >= HoldLimit) {
                        System.out.print("Holding ");
                        Thread.sleep(1000);
                        System.out.println();
                        Hold = false;
                        HoldQCond.signalAll();
                    }
                    else
                        while (Hold)
                            HoldQCond.await();

                    if (--HoldPop == 0) HoldQCond.signalAll();
                    return true;
                }
            }
            finally {
                HoldQ.unlock();
            }
        } catch (Exception Ex) {
            System.out.println("Unexpected exception in Hold: " + Ex);
            return false;
        }
    }

    private static class Server {
        private int nClients;
        final ReentrantLock thisLock = new ReentrantLock();
        final Condition thisCond = thisLock.newCondition();

        Server(int nClients) {
            this.nClients = nClients;
            try {
                for (int i = 0; i < nClients; ++i) {
                    final int fix = i;
                    new Thread() { public void run() { runServer(fix); }}.start();
                }
            } catch (Exception e) {
                System.err.println(e);
            }
        }

        // the total number of messages received by all server threads
        // on this server
        int msgsReceived = 0;

        // incremented each time we get a complete batch of requests
        private int currentBatch = 0;

        // the number of requests received since the last time currentBatch
        // was incremented
        private int currentBatchSize = 0;

        private void runServer(int id) {
            int msg;
            boolean held = false;
            final ReentrantLock thisLock = this.thisLock;
            final Condition thisCond = this.thisCond;

            try {

                // Startup barrier - rendezvous - wait for all threads.
                // Forces all threads to park on their LWPs, ensuring
                // proper provisioning on T1.
                // Alternately, use THR_BOUND threads
                Gate.lock(); try {
                    ++nReady;
                    if (nReady == ExThreads) {
                        GateCond.signalAll();
                    }
                    while (nReady != ExThreads)
                        GateCond.await();
                } finally { Gate.unlock(); }

                for (;;) {
                    //                    if (!held && currentBatchSize == 0) held = HoldCheck ();
                    msg = (++ mseq) ^ id;
                    thisLock.lock();
                    try {
                        ASum += msg;
                        ++msgsReceived;
                        int myBatch = currentBatch;
                        if (++currentBatchSize >= batchLimit) {
                            // this batch is full, start a new one ...
                            ++currentBatch;
                            currentBatchSize = 0;
                            // and wake up everyone in this one
                            thisCond.signalAll();
                        }
                        // Wait until our batch is complete
                        while (myBatch == currentBatch)
                            thisCond.await();
                    }
                    finally {
                        thisLock.unlock();
                    }
                }
            } catch (Exception e) {
                System.err.println("Server thread: exception "  + e);
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        int nServers = 10;
        int nClients = 10;
        int samplePeriod = 10000;
        int nSamples = 5;

        int nextArg = 0;
        while (nextArg < args.length) {
            String arg = args[nextArg++];
            if (arg.equals("-nc"))
                nClients = Integer.parseInt(args[nextArg++]);
            else if (arg.equals("-ns"))
                nServers = Integer.parseInt(args[nextArg++]);
            else if (arg.equals("-batch"))
                batchLimit = Integer.parseInt(args[nextArg++]);
            else if (arg.equals("-sample"))
                samplePeriod = Integer.parseInt(args[nextArg++]);
            else if (arg.equals("-np"))
                nSamples = Integer.parseInt(args[nextArg++]);
            else {
                System.err.println("Argument error:" + arg);
                System.exit(1);
            }
        }
        if (nClients <= 0 || nServers <= 0 || samplePeriod <= 0 || batchLimit > nClients) {
            System.err.println("Argument error");
            System.exit(1);
        }

        // default batch size is 2/3 the number of clients
        // (for no particular reason)
        if (false && batchLimit <= 0)
            batchLimit = (2 * nClients + 1) / 3;

        ExThreads = nServers * nClients;       // expected # of threads
        HoldLimit = ExThreads;

        // start up all threads
        Server[] servers = new Server[nServers];
        for (int i = 0; i < nServers; ++i) {
            servers[i] = new Server(nClients);
        }

        // Wait for consensus
        try {
            Gate.lock(); try {
                while (nReady != ExThreads ) GateCond.await();
            } finally { Gate.unlock(); }
        } catch (Exception ex) {
            System.out.println(ex);
        }
        System.out.println(
                            nReady + " Ready: nc=" + nClients + " ns=" + nServers + " batch=" + batchLimit);

        // Start sampling ...
        // Methodological problem: all the mutator threads
        // can starve the compiler threads, resulting in skewed scores.
        // In theory, over time, the scores will improve as the compiler
        // threads are granted CPU cycles, but in practice a "warm up" phase
        // might be good idea to help C2.  For this reason I've implemented
        // the "Hold" facility.

        long lastNumMsgs = 0;
        long sampleStart = System.currentTimeMillis();
        for (int j = 0; j < nSamples; ++j) {
            // when this sample period is supposed to end
            long sampleEnd = sampleStart + samplePeriod;
            for (;;) {
                long now = System.currentTimeMillis();
                if (now >= sampleEnd) {
                    // when it really did end
                    sampleEnd = now;
                    break;
                }
                Thread.sleep(sampleEnd - now);
            }

            if (false && j == 2) {
                System.out.print("Hold activated ...");
                HoldQ.lock();
                try {
                    Hold = true;
                    while (Hold) HoldQCond.await();
                }
                finally {
                    HoldQ.unlock();
                }
            }

            // there's no synchronization here, so the total i get is
            // approximate, but that's OK since any i miss for this
            // sample will get credited to the next sample, and on average
            // we'll be right
            long numMsgs = 0;
            for (int i = 0; i < nServers; ++i)
                numMsgs += servers[i].msgsReceived;
            long deltaMsgs = numMsgs - lastNumMsgs;
            long deltaT = sampleEnd - sampleStart;
            if (true || j != 2) {       // Don't report results if we issued a hold ...
                System.out.print(
                                 "Sample period = " + deltaT + " ms; "
                                 + "New msgs rcvd = " + deltaMsgs + "; "
                                 + "Throughput = " + (deltaMsgs*1000 / deltaT) + " msg/sec\n");
                //                for (int i = 0; i < nServers; ++i)
                //                    servers[i].thisLock.dump();
            }
            sampleStart = sampleEnd;
            lastNumMsgs = numMsgs;
        }
        System.exit(0);
    }
}
Revision:	1.9
Committed:	Thu Jan 15 18:34:19 2015 UTC (9 years, 4 months ago) by jsr166
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.8:	+0 -4 lines
Log Message:	delete extraneous blank lines
#	User	Rev	Content
1	dl	1.2	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain, as explained at
4	jsr166	1.8	* http://creativecommons.org/publicdomain/zero/1.0/
5	dl	1.2	*/
6	dl	1.1	// Adapted from code that was in turn
7			// Derived from SocketPerformanceTest.java - BugID: 4763450
8			//
9	jsr166	1.3	//
10	dl	1.1
11			import java.io.*;
12			import java.net.*;
13			import java.util.concurrent.*;
14			import java.util.concurrent.locks.*;
15
16			public class RLIBar {
17
18	jsr166	1.5	static int batchLimit;
19			static int mseq;
20			static int nReady;
21			static int ExThreads;
22			static int ASum;
23			static final ReentrantLock Gate = new ReentrantLock();
24			static final Condition GateCond = Gate.newCondition();
25
26			static final ReentrantLock HoldQ = new ReentrantLock();
27			static final Condition HoldQCond = HoldQ.newCondition();
28			static boolean Hold = false;
29			static int HoldPop;
30			static int HoldLimit;
31	jsr166	1.3
32	jsr166	1.6	private static boolean HoldCheck() {
33	jsr166	1.3	try {
34			HoldQ.lock();
35			try {
36	dl	1.1	if (!Hold) return false;
37			else {
38	jsr166	1.5	++HoldPop;
39	jsr166	1.3	if (HoldPop >= HoldLimit) {
40	jsr166	1.5	System.out.print("Holding ");
41			Thread.sleep(1000);
42			System.out.println();
43			Hold = false;
44			HoldQCond.signalAll();
45	jsr166	1.3	}
46	dl	1.1	else
47	jsr166	1.3	while (Hold)
48	jsr166	1.5	HoldQCond.await();
49	jsr166	1.3
50	jsr166	1.5	if (--HoldPop == 0) HoldQCond.signalAll();
51	dl	1.1	return true;
52			}
53			}
54	jsr166	1.3	finally {
55			HoldQ.unlock();
56	dl	1.1	}
57	jsr166	1.3	} catch (Exception Ex) {
58	jsr166	1.5	System.out.println("Unexpected exception in Hold: " + Ex);
59	dl	1.1	return false;
60			}
61			}
62
63			private static class Server {
64			private int nClients;
65			final ReentrantLock thisLock = new ReentrantLock();
66	jsr166	1.3	final Condition thisCond = thisLock.newCondition();
67
68	jsr166	1.5	Server(int nClients) {
69	dl	1.1	this.nClients = nClients;
70			try {
71			for (int i = 0; i < nClients; ++i) {
72	jsr166	1.5	final int fix = i;
73			new Thread() { public void run() { runServer(fix); }}.start();
74	dl	1.1	}
75			} catch (Exception e) {
76	jsr166	1.5	System.err.println(e);
77	dl	1.1	}
78			}
79
80			// the total number of messages received by all server threads
81			// on this server
82			int msgsReceived = 0;
83
84			// incremented each time we get a complete batch of requests
85			private int currentBatch = 0;
86
87			// the number of requests received since the last time currentBatch
88			// was incremented
89			private int currentBatchSize = 0;
90
91	jsr166	1.5	private void runServer(int id) {
92			int msg;
93	dl	1.1	boolean held = false;
94			final ReentrantLock thisLock = this.thisLock;
95	jsr166	1.3	final Condition thisCond = this.thisCond;
96	dl	1.1
97			try {
98
99	jsr166	1.3	// Startup barrier - rendezvous - wait for all threads.
100	dl	1.1	// Forces all threads to park on their LWPs, ensuring
101	jsr166	1.3	// proper provisioning on T1.
102			// Alternately, use THR_BOUND threads
103			Gate.lock(); try {
104	jsr166	1.5	++nReady;
105			if (nReady == ExThreads) {
106			GateCond.signalAll();
107	dl	1.1	}
108	jsr166	1.5	while (nReady != ExThreads)
109			GateCond.await();
110	dl	1.1	} finally { Gate.unlock(); }
111
112			for (;;) {
113	jsr166	1.5	// if (!held && currentBatchSize == 0) held = HoldCheck ();
114			msg = (++ mseq) ^ id;
115	jsr166	1.3	thisLock.lock();
116	dl	1.1	try {
117	jsr166	1.5	ASum += msg;
118	dl	1.1	++msgsReceived;
119			int myBatch = currentBatch;
120			if (++currentBatchSize >= batchLimit) {
121			// this batch is full, start a new one ...
122			++currentBatch;
123			currentBatchSize = 0;
124			// and wake up everyone in this one
125	jsr166	1.5	thisCond.signalAll();
126	dl	1.1	}
127			// Wait until our batch is complete
128			while (myBatch == currentBatch)
129	jsr166	1.3	thisCond.await();
130			}
131			finally {
132			thisLock.unlock();
133	dl	1.1	}
134			}
135			} catch (Exception e) {
136	jsr166	1.5	System.err.println("Server thread: exception " + e);
137	dl	1.1	e.printStackTrace();
138			}
139			}
140			}
141
142	jsr166	1.5	public static void main(String[] args) throws Exception {
143			int nServers = 10;
144			int nClients = 10;
145	dl	1.1	int samplePeriod = 10000;
146			int nSamples = 5;
147
148			int nextArg = 0;
149			while (nextArg < args.length) {
150			String arg = args[nextArg++];
151			if (arg.equals("-nc"))
152			nClients = Integer.parseInt(args[nextArg++]);
153			else if (arg.equals("-ns"))
154			nServers = Integer.parseInt(args[nextArg++]);
155			else if (arg.equals("-batch"))
156			batchLimit = Integer.parseInt(args[nextArg++]);
157			else if (arg.equals("-sample"))
158			samplePeriod = Integer.parseInt(args[nextArg++]);
159			else if (arg.equals("-np"))
160			nSamples = Integer.parseInt(args[nextArg++]);
161	jsr166	1.3	else {
162	jsr166	1.5	System.err.println("Argument error:" + arg);
163			System.exit(1);
164	jsr166	1.3	}
165	dl	1.1	}
166			if (nClients <= 0 \|\| nServers <= 0 \|\| samplePeriod <= 0 \|\| batchLimit > nClients) {
167	jsr166	1.5	System.err.println("Argument error");
168			System.exit(1);
169	dl	1.1	}
170
171			// default batch size is 2/3 the number of clients
172			// (for no particular reason)
173			if (false && batchLimit <= 0)
174			batchLimit = (2 * nClients + 1) / 3;
175
176	jsr166	1.5	ExThreads = nServers * nClients; // expected # of threads
177			HoldLimit = ExThreads;
178	dl	1.1
179			// start up all threads
180			Server[] servers = new Server[nServers];
181			for (int i = 0; i < nServers; ++i) {
182			servers[i] = new Server(nClients);
183			}
184
185	jsr166	1.3	// Wait for consensus
186	dl	1.1	try {
187	jsr166	1.3	Gate.lock(); try {
188	jsr166	1.5	while (nReady != ExThreads ) GateCond.await();
189	dl	1.1	} finally { Gate.unlock(); }
190	jsr166	1.3	} catch (Exception ex) {
191	jsr166	1.5	System.out.println(ex);
192	dl	1.1	}
193	jsr166	1.5	System.out.println(
194			nReady + " Ready: nc=" + nClients + " ns=" + nServers + " batch=" + batchLimit);
195	dl	1.1
196			// Start sampling ...
197			// Methodological problem: all the mutator threads
198			// can starve the compiler threads, resulting in skewed scores.
199			// In theory, over time, the scores will improve as the compiler
200			// threads are granted CPU cycles, but in practice a "warm up" phase
201	jsr166	1.3	// might be good idea to help C2. For this reason I've implemented
202			// the "Hold" facility.
203	dl	1.1
204			long lastNumMsgs = 0;
205			long sampleStart = System.currentTimeMillis();
206			for (int j = 0; j < nSamples; ++j) {
207			// when this sample period is supposed to end
208			long sampleEnd = sampleStart + samplePeriod;
209			for (;;) {
210			long now = System.currentTimeMillis();
211			if (now >= sampleEnd) {
212			// when it really did end
213			sampleEnd = now;
214			break;
215			}
216			Thread.sleep(sampleEnd - now);
217			}
218
219	jsr166	1.3	if (false && j == 2) {
220	jsr166	1.5	System.out.print("Hold activated ...");
221	jsr166	1.3	HoldQ.lock();
222	jsr166	1.7	try {
223	jsr166	1.5	Hold = true;
224			while (Hold) HoldQCond.await();
225	jsr166	1.3	}
226			finally {
227			HoldQ.unlock();
228	dl	1.1	}
229			}
230
231			// there's no synchronization here, so the total i get is
232			// approximate, but that's OK since any i miss for this
233			// sample will get credited to the next sample, and on average
234			// we'll be right
235			long numMsgs = 0;
236			for (int i = 0; i < nServers; ++i)
237			numMsgs += servers[i].msgsReceived;
238			long deltaMsgs = numMsgs - lastNumMsgs;
239			long deltaT = sampleEnd - sampleStart;
240	jsr166	1.4	if (true \|\| j != 2) { // Don't report results if we issued a hold ...
241	dl	1.1	System.out.print(
242			"Sample period = " + deltaT + " ms; "
243			+ "New msgs rcvd = " + deltaMsgs + "; "
244			+ "Throughput = " + (deltaMsgs*1000 / deltaT) + " msg/sec\n");
245			// for (int i = 0; i < nServers; ++i)
246			// servers[i].thisLock.dump();
247			}
248			sampleStart = sampleEnd;
249			lastNumMsgs = numMsgs;
250			}
251			System.exit(0);
252			}
253			}