/*
* 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));
}
}
}