infernap12/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2026-01-31 21:48:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

8223783: sun/net/www/http/HttpClient/MultiThreadTest.java sometimes detect threads+1 connections

Browse Source 
Reviewed-by: djelinski, michaelm
This commit is contained in:
Daniel Fuchs 2022-11-24 16:18:28 +00:00
parent ab1f9ffa78
commit 8effaa8f1f
1 changed files with 236 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

327
test/jdk/sun/net/www/http/HttpClient/MultiThreadTest.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2002, 2019, Oracle and/or its affiliates. All rights reserved.
* 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
@ -25,6 +25,7 @@
* @test
* @bug 4636628
* @summary HttpURLConnection duplicates HTTP GET requests when used with multiple threads
* @run main MultiThreadTest
*/
/*
@ -40,9 +41,14 @@
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 {
@ -51,8 +57,13 @@ public class MultiThreadTest extends Thread {
*/
static boolean debug = true; // disable debug once stability proven
static Object threadlock = new Object ();
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; }
@ -63,6 +74,9 @@ public class MultiThreadTest extends Thread {
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();
@ -79,42 +93,49 @@ public class MultiThreadTest extends Thread {
http.disconnect();
}
String uri;
byte[] b;
int requests;
final String uri;
final byte[] b;
final int requests;
final CountDownLatch countDown;
MultiThreadTest(String authority, int requests) throws Exception {
MultiThreadTest(String authority, int requests, CountDownLatch latch) throws Exception {
countDown = latch;
uri = "http://" + authority + "/foo.html";
b = new byte [256];
this.requests = requests;
synchronized (threadlock) {
threadCounter ++;
}
}
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 {
synchronized (threadlock) {
threadCounter --;
if (threadCounter == 0) {
threadlock.notifyAll();
}
}
countDown.countDown();
}
debug("client: end - " + Duration.ofNanos(System.nanoTime() - start));
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();
@ -138,62 +159,98 @@ public class MultiThreadTest extends Thread {
ss.bind(new InetSocketAddress(loopback, 0));
Server svr = new Server(ss);
svr.start();
var latch = new CountDownLatch(threads);
Object lock = MultiThreadTest.getLock();
synchronized (lock) {
for (int i=0; i<threads; i++) {
MultiThreadTest t = new MultiThreadTest(svr.getAuthority(), requests);
t.start ();
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);
}
try {
lock.wait();
} catch (InterruptedException e) {}
}
// 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;
for (Worker w : svr.workers()) {
if (w.headerFound) validConnections++;
}
if (validConnections > threads + 1 || validConnections == 0) { // could be less
throw new RuntimeException ("Expected " + threads + " connections: used " + validConnections);
}
// Sometimes the client drops a connection after a while and
// spawns a new one. Why this is happening is not clear,
// and JDK-8223783 is logged to follow up on this. For the sake
// of test stabilization we don't fail on `threads + 1` connections
// but log a warning instead.
if (validConnections == threads + 1) {
debug("WARNING: " + validConnections
+ " have been used, where only " + threads
+ " were expected!");
}
if (validConnections != cnt) {
debug("WARNING: got " + (cnt - validConnections) + " unexpected connections!");
}
if (validConnections == cnt && reqs != threads*requests) {
throw new RuntimeException ("Expected "+ threads*requests+ " requests: got " +reqs);
}
for (Thread worker : svr.workers()) {
worker.join(60_000);
}
debug("main thread end - " + Duration.ofNanos(System.nanoTime() - start));
debug("main thread end - " + at() + "ms");
}
}
@ -231,34 +288,45 @@ public class MultiThreadTest extends Thread {
public synchronized void shutdown() {
shutdown = true;
try {
ss.close();
} catch (IOException x) {
}
}
public void run() {
try {
ss.setSoTimeout(2000);
ss.setSoTimeout(6000);
long startServer = System.nanoTime();
for (;;) {
Socket s;
long acceptTime;
try {
MultiThreadTest.debug("server: calling accept.");
s = ss.accept();
MultiThreadTest.debug("server: return accept.");
} catch (SocketTimeoutException te) {
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.");
MultiThreadTest.debug("server: Shuting down at: "
+ MultiThreadTest.at() + "ms");
return;
}
}
continue;
if (te instanceof SocketTimeoutException)
continue;
throw te;
}
int id;
Worker w;
synchronized (this) {
id = connectionCount++;
w = new Worker(s, id);
w = new Worker(s, id, acceptTime);
workers.add(w);
}
w.start();
@ -280,14 +348,39 @@ public class MultiThreadTest extends Thread {
* multiple http requests on same connection.
*/
class Worker extends Thread {
Socket s;
int id;
volatile boolean headerFound;
final long TIMEOUT = MultiThreadTest.KEEP_ALIVE; // seconds
final long KEEP_ALIVE_NS = Duration.ofSeconds(TIMEOUT).toNanos(); // nanos
final Socket s;
final int id;
Worker(Socket s, 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;
@ -305,12 +398,42 @@ public class MultiThreadTest extends Thread {
}
}
/**
* {@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]) {
@ -327,6 +450,9 @@ public class MultiThreadTest extends Thread {
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];
@ -337,38 +463,49 @@ public class MultiThreadTest extends Thread {
s.getOutputStream() ));
for (;;) {
// read entire request from client
int n=0;
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");
MultiThreadTest.debug("worker: " + id + ": Shutdown at "
+ MultiThreadTest.at() + "ms");
s.close();
return;
}
if (headers.toString().contains("/foo.html?foo=")) {
headerFound = true;
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).");
"(" + n + " bytes) at " + MultiThreadTest.at() + "ms");
incRequests();
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=15, max="+max+"\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;
@ -378,11 +515,19 @@ public class MultiThreadTest extends Thread {
} catch (Exception e) {
e.printStackTrace();
} finally {
long end = stopTime = System.nanoTime();
try {
s.close();
} catch (Exception e) { }
MultiThreadTest.debug("worker: " + id + " end - " +
Duration.ofNanos(System.nanoTime() - start));
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));
}
}
}