534 lines
20 KiB
Java
534 lines
20 KiB
Java
/*
|
|
* Copyright (c) 2002, 2022, Oracle and/or its affiliates. All rights reserved.
|
|
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
|
*
|
|
* This code is free software; you can redistribute it and/or modify it
|
|
* under the terms of the GNU General Public License version 2 only, as
|
|
* published by the Free Software Foundation.
|
|
*
|
|
* This code is distributed in the hope that it will be useful, but WITHOUT
|
|
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
|
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
|
* version 2 for more details (a copy is included in the LICENSE file that
|
|
* accompanied this code).
|
|
*
|
|
* You should have received a copy of the GNU General Public License version
|
|
* 2 along with this work; if not, write to the Free Software Foundation,
|
|
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
|
|
*
|
|
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
|
|
* or visit www.oracle.com if you need additional information or have any
|
|
* questions.
|
|
*/
|
|
|
|
/*
|
|
* @test
|
|
* @bug 4636628
|
|
* @summary HttpURLConnection duplicates HTTP GET requests when used with multiple threads
|
|
* @run main MultiThreadTest
|
|
*/
|
|
|
|
/*
|
|
* This tests keep-alive behavior using chunkedinputstreams
|
|
* It checks that keep-alive connections are used and also
|
|
* that requests are not being repeated (due to errors)
|
|
*
|
|
* It also checks that the keepalive connections are closed eventually
|
|
* because the test will not terminate if the connections
|
|
* are not closed by the keep-alive timer.
|
|
*/
|
|
|
|
import java.net.*;
|
|
import java.io.*;
|
|
import java.time.Duration;
|
|
import java.util.ArrayList;
|
|
import java.util.List;
|
|
import java.util.Queue;
|
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
|
import java.util.concurrent.CountDownLatch;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|
|
|
public class MultiThreadTest extends Thread {
|
|
|
|
/*
|
|
* Is debugging enabled - start with -d to enable.
|
|
*/
|
|
static boolean debug = true; // disable debug once stability proven
|
|
|
|
static final Object threadlock = new Object ();
|
|
static int threadCounter = 0;
|
|
// KEEP_ALIVE sent by the server
|
|
static final int KEEP_ALIVE = 1; // seconds
|
|
// The sending thread will sleep for this time after sending
|
|
// half the number of its requests
|
|
static final int SLEEP = KEEP_ALIVE * 1000 + 500; // ms
|
|
|
|
static Object getLock() { return threadlock; }
|
|
|
|
static void debug(String msg) {
|
|
if (debug)
|
|
System.out.println(msg);
|
|
}
|
|
|
|
static final AtomicInteger reqnum = new AtomicInteger();
|
|
|
|
// Set to true after all requests have been sent
|
|
static final AtomicBoolean DONE = new AtomicBoolean();
|
|
|
|
void doRequest(String uri) throws Exception {
|
|
URL url = new URL(uri + "?foo="+reqnum.getAndIncrement());
|
|
HttpURLConnection http = (HttpURLConnection)url.openConnection();
|
|
InputStream in = http.getInputStream();
|
|
byte b[] = new byte[100];
|
|
int total = 0;
|
|
int n;
|
|
do {
|
|
n = in.read(b);
|
|
if (n > 0) total += n;
|
|
} while (n > 0);
|
|
debug ("client: read " + total + " bytes");
|
|
in.close();
|
|
http.disconnect();
|
|
}
|
|
|
|
final String uri;
|
|
final byte[] b;
|
|
final int requests;
|
|
final CountDownLatch countDown;
|
|
|
|
MultiThreadTest(String authority, int requests, CountDownLatch latch) throws Exception {
|
|
countDown = latch;
|
|
uri = "http://" + authority + "/foo.html";
|
|
b = new byte [256];
|
|
this.requests = requests;
|
|
}
|
|
|
|
public void run() {
|
|
long start = System.nanoTime();
|
|
try {
|
|
for (int i=0; i<requests; i++) {
|
|
doRequest (uri);
|
|
// sleep after sending half of the requests, that
|
|
// should cause the connections to be closed as idle
|
|
// if sleeping more than KeepAlive.
|
|
if (i == requests/2) Thread.sleep(SLEEP);
|
|
}
|
|
} catch (Exception e) {
|
|
throw new RuntimeException (e.getMessage());
|
|
} finally {
|
|
countDown.countDown();
|
|
}
|
|
debug("client: end at " + at() + "ms, thread duration "
|
|
+ Duration.ofNanos(System.nanoTime() - start).toMillis() + "ms");
|
|
}
|
|
|
|
static int threads=5;
|
|
// time at which main() started its work.
|
|
static volatile long MAIN_START;
|
|
|
|
// number of millis since MAIN_START
|
|
public static long at() {
|
|
return at(System.nanoTime());
|
|
}
|
|
// number of millis between MAIN_START and the given time stamp
|
|
public static long at(long nanoTime) {
|
|
return Duration.ofNanos(nanoTime - MAIN_START).toMillis();
|
|
}
|
|
|
|
public static void main(String args[]) throws Exception {
|
|
long start = System.nanoTime();
|
|
|
|
int x = 0, arg_len = args.length;
|
|
int requests = 20;
|
|
|
|
if (arg_len > 0 && args[0].equals("-d")) {
|
|
debug = true;
|
|
x = 1;
|
|
arg_len --;
|
|
}
|
|
if (arg_len > 0) {
|
|
threads = Integer.parseInt (args[x]);
|
|
requests = Integer.parseInt (args[x+1]);
|
|
}
|
|
|
|
/* start the server */
|
|
InetAddress loopback = InetAddress.getLoopbackAddress();
|
|
ServerSocket ss = new ServerSocket();
|
|
ss.bind(new InetSocketAddress(loopback, 0));
|
|
Server svr = new Server(ss);
|
|
svr.start();
|
|
var latch = new CountDownLatch(threads);
|
|
|
|
MAIN_START = System.nanoTime();
|
|
try {
|
|
Object lock = MultiThreadTest.getLock();
|
|
List<MultiThreadTest> tests = new ArrayList<>();
|
|
for (int i = 0; i < threads; i++) {
|
|
MultiThreadTest t = new MultiThreadTest(svr.getAuthority(), requests, latch);
|
|
tests.add(t);
|
|
t.start();
|
|
}
|
|
|
|
latch.await();
|
|
long end = System.nanoTime();
|
|
DONE.compareAndSet(false, true);
|
|
for (var test : tests) test.join();
|
|
|
|
MultiThreadTest.debug("DONE at " + at(end) + "ms");
|
|
|
|
// shutdown server - we're done.
|
|
svr.shutdown();
|
|
|
|
int cnt = svr.connectionCount();
|
|
MultiThreadTest.debug("Connections = " + cnt);
|
|
int reqs = Worker.getRequests();
|
|
MultiThreadTest.debug("Requests = " + reqs);
|
|
System.out.println("Connection count = " + cnt + " Request count = " + reqs);
|
|
|
|
// We may have received traffic from something else than
|
|
// our client. We should only count those workers for which
|
|
// the expected header has been found.
|
|
int validConnections = 0;
|
|
// We detect worker threads that may have timed out, so we don't include them in
|
|
// the count to compare with the number of connections.
|
|
int becameIdle = 0;
|
|
for (Worker w : svr.workers()) {
|
|
if (w.headerFound > 0) {
|
|
validConnections++;
|
|
if (w.mayHaveTimedOut(end)) {
|
|
debug("Worker " + w.id + " may have timed out");
|
|
becameIdle++;
|
|
} else {
|
|
long at0 = at(w.lastReading);
|
|
long at1 = at(w.lastReplied);
|
|
debug("Worker " + w.id +" has not timed out - last used at " +
|
|
Math.max(at0, at1));
|
|
}
|
|
} else {
|
|
debug("Worker " + w.id + " is not a valid connection");
|
|
}
|
|
}
|
|
|
|
if (validConnections > threads) {
|
|
if (SLEEP > KEEP_ALIVE) {
|
|
debug("INFO: " + validConnections
|
|
+ " have been used, with " + becameIdle
|
|
+ " becoming idle for more than " + KEEP_ALIVE + "s"
|
|
+ " while using " + threads
|
|
+ " threads to make concurrent connections");
|
|
} else {
|
|
debug("WARNING: " + validConnections
|
|
+ " have been used, with " + becameIdle
|
|
+ " becoming idle for more than " + KEEP_ALIVE + "s"
|
|
+ " where only " + threads
|
|
+ " connections and none idle were expected!");
|
|
}
|
|
}
|
|
|
|
if (validConnections > threads + becameIdle || validConnections == 0) { // could be less
|
|
throw new RuntimeException("Expected " + (threads + becameIdle) + " connections: used " + validConnections);
|
|
}
|
|
|
|
if (validConnections != cnt) {
|
|
debug("INFO: got " + (cnt - validConnections) + " unexpected connections");
|
|
}
|
|
if (reqs != threads * requests) {
|
|
throw new RuntimeException("Expected " + threads * requests + " requests: got " + reqs);
|
|
}
|
|
|
|
} finally {
|
|
debug("waiting for worker to shutdown at " + at() +"ms");
|
|
for (Worker worker : svr.workers()) {
|
|
// We want to verify that the client will eventually
|
|
// close the idle connections. So just join the worker
|
|
// and wait... This shouldn't take more than the granularity
|
|
// of the keep-alive cache timer - so we're not actually
|
|
// going to have to wait for one full minute here.
|
|
worker.join(60_000);
|
|
}
|
|
}
|
|
|
|
debug("main thread end - " + at() + "ms");
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Server thread to accept connection and create worker threads
|
|
* to service each connection.
|
|
*/
|
|
class Server extends Thread {
|
|
ServerSocket ss;
|
|
int connectionCount;
|
|
boolean shutdown = false;
|
|
private final Queue<Worker> workers = new ConcurrentLinkedQueue<>();
|
|
|
|
Server(ServerSocket ss) {
|
|
this.ss = ss;
|
|
}
|
|
|
|
public String getAuthority() {
|
|
InetAddress address = ss.getInetAddress();
|
|
String hostaddr = address.isAnyLocalAddress()
|
|
? "localhost" : address.getHostAddress();
|
|
if (hostaddr.indexOf(':') > -1) {
|
|
hostaddr = "[" + hostaddr + "]";
|
|
}
|
|
return hostaddr + ":" + ss.getLocalPort();
|
|
}
|
|
|
|
public Queue<Worker> workers() {
|
|
return workers;
|
|
}
|
|
|
|
public synchronized int connectionCount() {
|
|
return connectionCount;
|
|
}
|
|
|
|
public synchronized void shutdown() {
|
|
shutdown = true;
|
|
try {
|
|
ss.close();
|
|
} catch (IOException x) {
|
|
}
|
|
}
|
|
|
|
public void run() {
|
|
try {
|
|
ss.setSoTimeout(6000);
|
|
long startServer = System.nanoTime();
|
|
|
|
for (;;) {
|
|
Socket s;
|
|
long acceptTime;
|
|
try {
|
|
MultiThreadTest.debug("server: calling accept.");
|
|
s = ss.accept();
|
|
acceptTime = System.nanoTime();
|
|
MultiThreadTest.debug("server: return accept (at " +
|
|
MultiThreadTest.at(acceptTime)+ "ms)");
|
|
} catch (IOException te) {
|
|
MultiThreadTest.debug("server: STE");
|
|
synchronized (this) {
|
|
if (shutdown) {
|
|
MultiThreadTest.debug("server: Shuting down at: "
|
|
+ MultiThreadTest.at() + "ms");
|
|
return;
|
|
}
|
|
}
|
|
if (te instanceof SocketTimeoutException)
|
|
continue;
|
|
throw te;
|
|
}
|
|
|
|
int id;
|
|
Worker w;
|
|
synchronized (this) {
|
|
id = connectionCount++;
|
|
w = new Worker(s, id, acceptTime);
|
|
workers.add(w);
|
|
}
|
|
w.start();
|
|
MultiThreadTest.debug("server: Started worker " + id);
|
|
}
|
|
|
|
} catch (Exception e) {
|
|
e.printStackTrace();
|
|
} finally {
|
|
try {
|
|
ss.close();
|
|
} catch (Exception e) { }
|
|
}
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Worker thread to service single connection - can service
|
|
* multiple http requests on same connection.
|
|
*/
|
|
class Worker extends Thread {
|
|
final long TIMEOUT = MultiThreadTest.KEEP_ALIVE; // seconds
|
|
final long KEEP_ALIVE_NS = Duration.ofSeconds(TIMEOUT).toNanos(); // nanos
|
|
final Socket s;
|
|
final int id;
|
|
|
|
// time at which the connection was accepted (nanos)
|
|
final long acceptTime;
|
|
|
|
// number of requests that had the expected URI
|
|
volatile int headerFound;
|
|
// time at which the worker thread exited
|
|
volatile long stopTime;
|
|
// Time at which the first call to is.read() for the last request
|
|
// returned. This includes cases where -1 was returned.
|
|
volatile long startReading;
|
|
// Lat time at which a byte was read from the stream.
|
|
volatile long lastReading;
|
|
// number of times that the time between two consecutive received requests
|
|
// exceeded the KEEP_ALIVE timeout.
|
|
volatile int timeoutExceeded;
|
|
// Number of requests handled by this worker
|
|
volatile int requestHandled;
|
|
// Time at which the last byte of the last reply was sent
|
|
volatile long lastReplied;
|
|
// Whether the worker was asked to stop
|
|
volatile boolean done;
|
|
|
|
Worker(Socket s, int id, long acceptTime) {
|
|
super("Worker-" + id);
|
|
this.s = s;
|
|
this.id = id;
|
|
// no time can have a value before accepTime
|
|
this.acceptTime = lastReading = lastReplied = startReading = acceptTime;
|
|
}
|
|
|
|
static int requests = 0;
|
|
static final Object rlock = new Object();
|
|
|
|
public static int getRequests () {
|
|
synchronized (rlock) {
|
|
return requests;
|
|
}
|
|
}
|
|
|
|
public static void incRequests () {
|
|
synchronized (rlock) {
|
|
requests++;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* {@return Whether this worker might have been idle for more
|
|
* than the KEEP_ALIVE timeout}
|
|
* This will be true if the worker detected that the idle timeout
|
|
* was exceeded between two consecutive request, or
|
|
* if the time between the last reply and `nanosNow` exceeds
|
|
* the keep-alive time.
|
|
* @param nanosNow a timestamp in nano seconds
|
|
*/
|
|
public boolean mayHaveTimedOut(long nanosNow) {
|
|
// the minimum time elapsed between nanosNow and:
|
|
// - the time the socket was accepted
|
|
// - the last time a byte was received
|
|
// - the last time a reply was sent.
|
|
// We must not use `startReading` here because `startReading` may
|
|
// be set if the client asynchronously closes the connection
|
|
// after all requests have been sent. We should really only
|
|
// take into account `lastReading` and `lastReplied`.
|
|
long idle = Math.min(nanosNow - lastReading, nanosNow - lastReplied);
|
|
return timeoutExceeded > 0 || idle >= KEEP_ALIVE_NS;
|
|
}
|
|
|
|
int readUntil(InputStream in, StringBuilder headers, char[] seq) throws IOException {
|
|
int i=0, count=0;
|
|
long last;
|
|
while (true) {
|
|
int c = in.read();
|
|
last = System.nanoTime();
|
|
if (count == 0) {
|
|
// time at which the first byte of the request (or EOF) was received
|
|
startReading = last;
|
|
}
|
|
if (c == -1)
|
|
return -1;
|
|
// time at which the last byte of the request was received (excludes EOF)
|
|
lastReading = last;
|
|
headers.append((char)c);
|
|
count++;
|
|
if (c == seq[i]) {
|
|
i++;
|
|
if (i == seq.length)
|
|
return count;
|
|
continue;
|
|
} else {
|
|
i = 0;
|
|
}
|
|
}
|
|
}
|
|
|
|
public void run() {
|
|
long start = System.nanoTime();
|
|
|
|
// lastUsed starts when the connection was accepted
|
|
long lastUsed = acceptTime;
|
|
int expectedReqs = 0;
|
|
try {
|
|
int max = 400;
|
|
byte b[] = new byte[1000];
|
|
InputStream in = new BufferedInputStream(s.getInputStream());
|
|
// response to client
|
|
PrintStream out = new PrintStream(
|
|
new BufferedOutputStream(
|
|
s.getOutputStream() ));
|
|
|
|
for (;;) {
|
|
// read entire request from client
|
|
int n;
|
|
StringBuilder headers = new StringBuilder();
|
|
n = readUntil(in, headers, new char[] {'\r','\n', '\r','\n'});
|
|
long idle = startReading - lastUsed;
|
|
if (idle >= KEEP_ALIVE_NS) {
|
|
if (!MultiThreadTest.DONE.get()) {
|
|
// avoid increasing timeoutExceeded after the test is no
|
|
// longer sending requests.
|
|
timeoutExceeded++;
|
|
}
|
|
}
|
|
if (n <= 0) {
|
|
MultiThreadTest.debug("worker: " + id + ": Shutdown at "
|
|
+ MultiThreadTest.at() + "ms");
|
|
s.close();
|
|
return;
|
|
}
|
|
if (headers.toString().contains("/foo.html?foo=")) {
|
|
headerFound = ++expectedReqs;
|
|
incRequests();
|
|
} else {
|
|
MultiThreadTest.debug("worker: " + id + ": Unexpected request received: " + headers);
|
|
s.close();
|
|
return;
|
|
}
|
|
|
|
MultiThreadTest.debug("worker " + id +
|
|
": Read request from client " +
|
|
"(" + n + " bytes) at " + MultiThreadTest.at() + "ms");
|
|
|
|
out.print("HTTP/1.1 200 OK\r\n");
|
|
out.print("Transfer-Encoding: chunked\r\n");
|
|
out.print("Content-Type: text/html\r\n");
|
|
out.print("Connection: Keep-Alive\r\n");
|
|
out.print("Keep-Alive: timeout=" + TIMEOUT + ", max="+max+"\r\n");
|
|
out.print("\r\n");
|
|
out.print("6\r\nHello \r\n");
|
|
out.print("5\r\nWorld\r\n");
|
|
out.print("0\r\n\r\n");
|
|
out.flush();
|
|
requestHandled++;
|
|
lastUsed = lastReplied = System.nanoTime();
|
|
if (--max == 0) {
|
|
s.close();
|
|
return;
|
|
}
|
|
}
|
|
|
|
} catch (Exception e) {
|
|
e.printStackTrace();
|
|
} finally {
|
|
long end = stopTime = System.nanoTime();
|
|
try {
|
|
s.close();
|
|
} catch (Exception e) { }
|
|
MultiThreadTest.debug("worker: " + id + " end at " +
|
|
MultiThreadTest.at() + "ms, elapsed since worker start: " +
|
|
Duration.ofNanos(end - start).toMillis() + "ms, elapsed since accept: " +
|
|
Duration.ofNanos(end - acceptTime).toMillis() +
|
|
"ms, timeout exceeded: " + timeoutExceeded +
|
|
", successfuly handled " + requestHandled + "/" +
|
|
expectedReqs + " genuine requests, " +
|
|
", mayHaveTimedOut: " + mayHaveTimedOut(end));
|
|
}
|
|
}
|
|
|
|
}
|