8269258: java/net/httpclient/ManyRequestsLegacy.java failed with connection timeout
Reviewed-by: michaelm
This commit is contained in:
Daniel Fuchs
parent
fd8cb2dae8
commit
37921e3080
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2015, 2020, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2015, 2021, 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
|
||||
@ -32,13 +32,13 @@
|
||||
* @compile ../../../com/sun/net/httpserver/LogFilter.java
|
||||
* @compile ../../../com/sun/net/httpserver/EchoHandler.java
|
||||
* @compile ../../../com/sun/net/httpserver/FileServerHandler.java
|
||||
* @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=ssl ManyRequests
|
||||
* @run main/othervm/timeout=40 -Dtest.insertDelay=true ManyRequests
|
||||
* @run main/othervm/timeout=40 -Dtest.chunkSize=64 ManyRequests
|
||||
* @run main/othervm/timeout=40 -Dtest.insertDelay=true -Dtest.chunkSize=64 ManyRequests
|
||||
* @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=ssl,channel ManyRequests
|
||||
* @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=channel -Dtest.insertDelay=true ManyRequests
|
||||
* @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=channel -Dtest.chunkSize=64 ManyRequests
|
||||
* @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=channel -Dtest.insertDelay=true -Dtest.chunkSize=64 ManyRequests
|
||||
* @summary Send a large number of requests asynchronously
|
||||
*/
|
||||
// * @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=ssl ManyRequests
|
||||
// * @run main/othervm/timeout=40 -Djdk.httpclient.HttpClient.log=ssl,channel ManyRequests
|
||||
|
||||
import com.sun.net.httpserver.HttpsConfigurator;
|
||||
import com.sun.net.httpserver.HttpsParameters;
|
||||
@ -47,6 +47,7 @@ import com.sun.net.httpserver.HttpExchange;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.net.ConnectException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.URI;
|
||||
@ -54,13 +55,18 @@ import java.net.http.HttpClient;
|
||||
import java.net.http.HttpClient.Builder;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpRequest.BodyPublishers;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.net.http.HttpResponse.BodyHandlers;
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.Formatter;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedList;
|
||||
import java.util.Map;
|
||||
import java.util.Random;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
@ -68,17 +74,27 @@ import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.logging.Logger;
|
||||
import java.util.logging.Level;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.stream.Stream;
|
||||
import javax.net.ssl.SSLContext;
|
||||
|
||||
import jdk.test.lib.Platform;
|
||||
import jdk.test.lib.RandomFactory;
|
||||
import jdk.test.lib.net.SimpleSSLContext;
|
||||
import jdk.test.lib.net.URIBuilder;
|
||||
|
||||
public class ManyRequests {
|
||||
|
||||
volatile static int counter = 0;
|
||||
static final int MAX_COUNT = 20;
|
||||
static final int MAX_LIMIT = 40;
|
||||
static final AtomicInteger COUNT = new AtomicInteger();
|
||||
static final AtomicInteger LIMIT = new AtomicInteger(MAX_LIMIT);
|
||||
static final Random RANDOM = RandomFactory.getRandom();
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
Logger logger = Logger.getLogger("com.sun.net.httpserver");
|
||||
logger.setLevel(Level.ALL);
|
||||
logger.info("TEST");
|
||||
Stream.of(Logger.getLogger("").getHandlers()).forEach((h) -> h.setLevel(Level.ALL));
|
||||
System.out.println("Sending " + REQUESTS
|
||||
+ " requests; delay=" + INSERT_DELAY
|
||||
+ ", chunks=" + CHUNK_SIZE
|
||||
@ -106,14 +122,14 @@ public class ManyRequests {
|
||||
}
|
||||
|
||||
//static final int REQUESTS = 1000;
|
||||
static final int REQUESTS = 20;
|
||||
static final int REQUESTS = MAX_COUNT;
|
||||
static final boolean INSERT_DELAY = Boolean.getBoolean("test.insertDelay");
|
||||
static final int CHUNK_SIZE = Math.max(0,
|
||||
Integer.parseInt(System.getProperty("test.chunkSize", "0")));
|
||||
static final boolean XFIXED = Boolean.getBoolean("test.XFixed");
|
||||
|
||||
static class TestEchoHandler extends EchoHandler {
|
||||
final Random rand = jdk.test.lib.RandomFactory.getRandom();
|
||||
final Random rand = RANDOM;
|
||||
@Override
|
||||
public void handle(HttpExchange e) throws IOException {
|
||||
System.out.println("Server: received " + e.getRequestURI());
|
||||
@ -139,60 +155,126 @@ public class ManyRequests {
|
||||
}
|
||||
}
|
||||
|
||||
static String now(long start) {
|
||||
long elapsed = System.nanoTime() - start;
|
||||
long ms = elapsed / 1000_000L;
|
||||
long s = ms / 1000L;
|
||||
if (s == 0) return ms + "ms: ";
|
||||
return s + "s, " + (ms - s * 1000L) + "ms: ";
|
||||
}
|
||||
|
||||
static String failure(Throwable t) {
|
||||
String s = "\n\t failed: " + t;
|
||||
for (t = t.getCause(); t != null ; t = t.getCause()) {
|
||||
s = s + "\n\t\t Caused by: " + t;
|
||||
}
|
||||
return s;
|
||||
}
|
||||
|
||||
static void test(HttpsServer server, HttpClient client) throws Exception {
|
||||
int port = server.getAddress().getPort();
|
||||
URI baseURI = new URI("https://localhost:" + port + "/foo/x");
|
||||
|
||||
URI baseURI = URIBuilder.newBuilder()
|
||||
.scheme("https")
|
||||
.host(InetAddress.getLoopbackAddress().getHostName())
|
||||
.port(port)
|
||||
.path("/foo/x").build();
|
||||
server.createContext("/foo", new TestEchoHandler());
|
||||
server.start();
|
||||
|
||||
RequestLimiter limiter = new RequestLimiter(40);
|
||||
Random rand = new Random();
|
||||
CompletableFuture<?>[] results = new CompletableFuture<?>[REQUESTS];
|
||||
HashMap<HttpRequest,byte[]> bodies = new HashMap<>();
|
||||
// This loop implements a retry mechanism to work around an issue
|
||||
// on some systems (observed on Windows 10) that seem to be trying to
|
||||
// throttle the number of connections that can be made concurrently by
|
||||
// rejecting connection attempts.
|
||||
// On the first iteration of this loop, we will attempt 20 concurrent
|
||||
// requests. If this fails with ConnectException, we will retry the
|
||||
// 20 requests, but limiting the concurrency to 10 (LIMIT <- 10).
|
||||
// If this fails again, the test will fail.
|
||||
boolean done = false;
|
||||
LOOP: do {
|
||||
RequestLimiter limiter = new RequestLimiter(LIMIT.get());
|
||||
Random rand = RANDOM;
|
||||
CompletableFuture<?>[] results = new CompletableFuture<?>[REQUESTS];
|
||||
Map<HttpRequest,byte[]> bodies = new ConcurrentHashMap<>();
|
||||
|
||||
for (int i=0; i<REQUESTS; i++) {
|
||||
byte[] buf = new byte[(i+1)*CHUNK_SIZE+i+1]; // different size bodies
|
||||
rand.nextBytes(buf);
|
||||
URI uri = new URI(baseURI.toString() + String.valueOf(i+1));
|
||||
HttpRequest r = HttpRequest.newBuilder(uri)
|
||||
.header("XFixed", "true")
|
||||
.POST(BodyPublishers.ofByteArray(buf))
|
||||
.build();
|
||||
bodies.put(r, buf);
|
||||
long start = System.nanoTime();
|
||||
|
||||
results[i] =
|
||||
limiter.whenOkToSend()
|
||||
.thenCompose((v) -> {
|
||||
System.out.println("Client: sendAsync: " + r.uri());
|
||||
return client.sendAsync(r, BodyHandlers.ofByteArray());
|
||||
})
|
||||
.thenCompose((resp) -> {
|
||||
limiter.requestComplete();
|
||||
if (resp.statusCode() != 200) {
|
||||
String s = "Expected 200, got: " + resp.statusCode();
|
||||
System.out.println(s + " from "
|
||||
+ resp.request().uri().getPath());
|
||||
return completedWithIOException(s);
|
||||
} else {
|
||||
counter++;
|
||||
System.out.println("Result (" + counter + ") from "
|
||||
+ resp.request().uri().getPath());
|
||||
}
|
||||
return CompletableFuture.completedStage(resp.body())
|
||||
.thenApply((b) -> new Pair<>(resp, b));
|
||||
})
|
||||
.thenAccept((pair) -> {
|
||||
HttpRequest request = pair.t.request();
|
||||
byte[] requestBody = bodies.get(request);
|
||||
check(Arrays.equals(requestBody, pair.u),
|
||||
"bodies not equal:[" + bytesToHexString(requestBody)
|
||||
+ "] [" + bytesToHexString(pair.u) + "]");
|
||||
for (int i = 0; i < REQUESTS; i++) {
|
||||
byte[] buf = new byte[(i + 1) * CHUNK_SIZE + i + 1]; // different size bodies
|
||||
rand.nextBytes(buf);
|
||||
URI uri = new URI(baseURI.toString() + String.valueOf(i + 1));
|
||||
HttpRequest r = HttpRequest.newBuilder(uri)
|
||||
.header("XFixed", "true")
|
||||
.POST(BodyPublishers.ofByteArray(buf))
|
||||
.build();
|
||||
bodies.put(r, buf);
|
||||
|
||||
});
|
||||
}
|
||||
results[i] =
|
||||
limiter.whenOkToSend()
|
||||
.thenCompose((v) -> {
|
||||
System.out.println("Client: sendAsync: " + r.uri());
|
||||
return client.sendAsync(r, BodyHandlers.ofByteArray());
|
||||
})
|
||||
.handle((resp, t) -> {
|
||||
limiter.requestComplete();
|
||||
CompletionStage<Pair<HttpResponse<byte[]>, byte[]>> res;
|
||||
String now = now(start);
|
||||
if (t == null) {
|
||||
if (resp.statusCode() != 200) {
|
||||
String s = "Expected 200, got: " + resp.statusCode();
|
||||
System.out.println(now + s + " from "
|
||||
+ resp.request().uri().getPath());
|
||||
res = completedWithIOException(s);
|
||||
return res;
|
||||
} else {
|
||||
int counter = COUNT.incrementAndGet();
|
||||
System.out.println(now + "Result (" + counter + ") from "
|
||||
+ resp.request().uri().getPath());
|
||||
}
|
||||
res = CompletableFuture.completedStage(resp.body())
|
||||
.thenApply((b) -> new Pair<>(resp, b));
|
||||
return res;
|
||||
} else {
|
||||
int counter = COUNT.incrementAndGet();
|
||||
System.out.println(now + "Result (" + counter + ") from "
|
||||
+ r.uri().getPath()
|
||||
+ failure(t));
|
||||
res = CompletableFuture.failedFuture(t);
|
||||
return res;
|
||||
}
|
||||
})
|
||||
.thenCompose(c -> c)
|
||||
.thenAccept((pair) -> {
|
||||
HttpRequest request = pair.t.request();
|
||||
byte[] requestBody = bodies.get(request);
|
||||
check(Arrays.equals(requestBody, pair.u),
|
||||
"bodies not equal:[" + bytesToHexString(requestBody)
|
||||
+ "] [" + bytesToHexString(pair.u) + "]");
|
||||
|
||||
});
|
||||
}
|
||||
|
||||
// wait for them all to complete and throw exception in case of err
|
||||
try {
|
||||
CompletableFuture.allOf(results).join();
|
||||
done = true;
|
||||
} catch (CompletionException e) {
|
||||
if (!Platform.isWindows()) throw e;
|
||||
if (LIMIT.get() < REQUESTS) throw e;
|
||||
Throwable cause = e;
|
||||
while ((cause = cause.getCause()) != null) {
|
||||
if (cause instanceof ConnectException) {
|
||||
// try again, limit concurrency by half
|
||||
COUNT.set(0);
|
||||
LIMIT.set(REQUESTS/2);
|
||||
System.out.println("*** Retrying due to " + cause);
|
||||
continue LOOP;
|
||||
}
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
} while (!done);
|
||||
|
||||
// wait for them all to complete and throw exception in case of error
|
||||
CompletableFuture.allOf(results).join();
|
||||
}
|
||||
|
||||
static <T> CompletableFuture<T> completedWithIOException(String message) {
|
||||
@ -213,13 +295,7 @@ public class ManyRequests {
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
static final class Pair<T,U> {
|
||||
Pair(T t, U u) {
|
||||
this.t = t; this.u = u;
|
||||
}
|
||||
T t;
|
||||
U u;
|
||||
}
|
||||
record Pair<T,U>(T t, U u) { }
|
||||
|
||||
/**
|
||||
* A simple limiter for controlling the number of requests to be run in
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2017, 2020, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2017, 2021, 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
|
||||
@ -33,10 +33,14 @@
|
||||
* @compile ../../../com/sun/net/httpserver/EchoHandler.java
|
||||
* @compile ../../../com/sun/net/httpserver/FileServerHandler.java
|
||||
* @build ManyRequests ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Dtest.XFixed=true ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Dtest.XFixed=true -Dtest.insertDelay=true ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Dtest.XFixed=true -Dtest.chunkSize=64 ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Dtest.XFixed=true
|
||||
* -Djdk.httpclient.HttpClient.log=channel ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Dtest.XFixed=true -Dtest.insertDelay=true
|
||||
* -Djdk.httpclient.HttpClient.log=channel ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Dtest.XFixed=true -Dtest.chunkSize=64
|
||||
* -Djdk.httpclient.HttpClient.log=channel ManyRequests2
|
||||
* @run main/othervm/timeout=40 -Djdk.internal.httpclient.debug=true
|
||||
* -Djdk.httpclient.HttpClient.log=channel
|
||||
* -Dtest.XFixed=true -Dtest.insertDelay=true
|
||||
* -Dtest.chunkSize=64 ManyRequests2
|
||||
* @summary Send a large number of requests asynchronously.
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
|
||||
* Copyright (c) 2015, 2021, 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
|
||||
@ -43,6 +43,7 @@
|
||||
|
||||
import javax.net.ssl.HttpsURLConnection;
|
||||
import javax.net.ssl.HostnameVerifier;
|
||||
|
||||
import com.sun.net.httpserver.HttpsConfigurator;
|
||||
import com.sun.net.httpserver.HttpsParameters;
|
||||
import com.sun.net.httpserver.HttpsServer;
|
||||
@ -50,12 +51,18 @@ import com.sun.net.httpserver.HttpExchange;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.net.ConnectException;
|
||||
import java.net.HttpURLConnection;
|
||||
import java.net.InetAddress;
|
||||
import java.net.URI;
|
||||
import java.net.URLConnection;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.stream.Collectors;
|
||||
import javax.net.ssl.SSLContext;
|
||||
import javax.net.ssl.SSLSession;
|
||||
@ -73,12 +80,20 @@ import java.util.LinkedList;
|
||||
import java.util.Random;
|
||||
import java.util.logging.Logger;
|
||||
import java.util.logging.Level;
|
||||
|
||||
import jdk.test.lib.Platform;
|
||||
import jdk.test.lib.RandomFactory;
|
||||
import jdk.test.lib.net.SimpleSSLContext;
|
||||
import static java.net.Proxy.NO_PROXY;
|
||||
|
||||
public class ManyRequestsLegacy {
|
||||
|
||||
volatile static int counter = 0;
|
||||
static final int MAX_COUNT = 20;
|
||||
static final int MAX_LIMIT = 40;
|
||||
static final AtomicInteger COUNT = new AtomicInteger();
|
||||
static final AtomicInteger LIMIT = new AtomicInteger(MAX_LIMIT);
|
||||
static final Random RANDOM = RandomFactory.getRandom();
|
||||
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
Logger logger = Logger.getLogger("com.sun.net.httpserver");
|
||||
@ -110,7 +125,7 @@ public class ManyRequestsLegacy {
|
||||
}
|
||||
|
||||
//static final int REQUESTS = 1000;
|
||||
static final int REQUESTS = 20;
|
||||
static final int REQUESTS = MAX_COUNT;
|
||||
static final boolean INSERT_DELAY = Boolean.getBoolean("test.insertDelay");
|
||||
static final int CHUNK_SIZE = Math.max(0,
|
||||
Integer.parseInt(System.getProperty("test.chunkSize", "0")));
|
||||
@ -194,7 +209,7 @@ public class ManyRequestsLegacy {
|
||||
}
|
||||
|
||||
static class TestEchoHandler extends EchoHandler {
|
||||
final Random rand = new Random();
|
||||
final Random rand = RANDOM;
|
||||
@Override
|
||||
public void handle(HttpExchange e) throws IOException {
|
||||
System.out.println("Server: received " + e.getRequestURI());
|
||||
@ -220,60 +235,119 @@ public class ManyRequestsLegacy {
|
||||
}
|
||||
}
|
||||
|
||||
static String now(long start) {
|
||||
long elapsed = System.nanoTime() - start;
|
||||
long ms = elapsed / 1000_000L;
|
||||
long s = ms / 1000L;
|
||||
if (s == 0) return ms + "ms: ";
|
||||
return s + "s, " + (ms - s * 1000L) + "ms: ";
|
||||
}
|
||||
|
||||
static String failure(Throwable t) {
|
||||
String s = "\n\t failed: " + t;
|
||||
for (t = t.getCause(); t != null ; t = t.getCause()) {
|
||||
s = s + "\n\t\t Caused by: " + t;
|
||||
}
|
||||
return s;
|
||||
}
|
||||
|
||||
static void test(HttpsServer server, LegacyHttpClient client) throws Exception {
|
||||
int port = server.getAddress().getPort();
|
||||
URI baseURI = new URI("https://localhost:" + port + "/foo/x");
|
||||
server.createContext("/foo", new TestEchoHandler());
|
||||
server.start();
|
||||
|
||||
RequestLimiter limiter = new RequestLimiter(40);
|
||||
Random rand = new Random();
|
||||
CompletableFuture<?>[] results = new CompletableFuture<?>[REQUESTS];
|
||||
HashMap<HttpRequest,byte[]> bodies = new HashMap<>();
|
||||
// This loop implements a retry mechanism to work around an issue
|
||||
// on some systems (observed on Windows 10) that seem to be trying to
|
||||
// throttle the number of connections that can be made concurrently by
|
||||
// rejecting connection attempts.
|
||||
// On the first iteration of this loop, we will attempt 20 concurrent
|
||||
// requests. If this fails with ConnectException, we will retry the
|
||||
// 20 requests, but limiting the concurrency to 10 (LIMIT <- 10).
|
||||
// If this fails again, the test will fail.
|
||||
boolean done = false;
|
||||
LOOP: do {
|
||||
RequestLimiter limiter = new RequestLimiter(LIMIT.get());
|
||||
Random rand = RANDOM;
|
||||
CompletableFuture<?>[] results = new CompletableFuture<?>[REQUESTS];
|
||||
Map<HttpRequest,byte[]> bodies = new ConcurrentHashMap<>();
|
||||
long start = System.nanoTime();
|
||||
|
||||
for (int i=0; i<REQUESTS; i++) {
|
||||
byte[] buf = new byte[(i+1)*CHUNK_SIZE+i+1]; // different size bodies
|
||||
rand.nextBytes(buf);
|
||||
URI uri = new URI(baseURI.toString() + String.valueOf(i+1));
|
||||
HttpRequest r = HttpRequest.newBuilder(uri)
|
||||
.header("XFixed", "true")
|
||||
.POST(BodyPublishers.ofByteArray(buf))
|
||||
.build();
|
||||
bodies.put(r, buf);
|
||||
for (int i = 0; i < REQUESTS; i++) {
|
||||
byte[] buf = new byte[(i + 1) * CHUNK_SIZE + i + 1]; // different size bodies
|
||||
rand.nextBytes(buf);
|
||||
URI uri = new URI(baseURI.toString() + String.valueOf(i + 1));
|
||||
HttpRequest r = HttpRequest.newBuilder(uri)
|
||||
.header("XFixed", "true")
|
||||
.POST(BodyPublishers.ofByteArray(buf))
|
||||
.build();
|
||||
bodies.put(r, buf);
|
||||
|
||||
results[i] =
|
||||
limiter.whenOkToSend()
|
||||
.thenCompose((v) -> {
|
||||
System.out.println("Client: sendAsync: " + r.uri());
|
||||
return client.sendAsync(r, buf);
|
||||
})
|
||||
.thenCompose((resp) -> {
|
||||
limiter.requestComplete();
|
||||
if (resp.statusCode() != 200) {
|
||||
String s = "Expected 200, got: " + resp.statusCode();
|
||||
System.out.println(s + " from "
|
||||
+ resp.request().uri().getPath());
|
||||
return completedWithIOException(s);
|
||||
} else {
|
||||
counter++;
|
||||
System.out.println("Result (" + counter + ") from "
|
||||
+ resp.request().uri().getPath());
|
||||
}
|
||||
return CompletableFuture.completedStage(resp.body())
|
||||
.thenApply((b) -> new Pair<>(resp, b));
|
||||
})
|
||||
.thenAccept((pair) -> {
|
||||
HttpRequest request = pair.t.request();
|
||||
byte[] requestBody = bodies.get(request);
|
||||
check(Arrays.equals(requestBody, pair.u),
|
||||
"bodies not equal:[" + bytesToHexString(requestBody)
|
||||
+ "] [" + bytesToHexString(pair.u) + "]");
|
||||
results[i] =
|
||||
limiter.whenOkToSend()
|
||||
.thenCompose((v) -> {
|
||||
System.out.println("Client: sendAsync: " + r.uri());
|
||||
return client.sendAsync(r, buf);
|
||||
})
|
||||
.handle((resp, t) -> {
|
||||
limiter.requestComplete();
|
||||
CompletionStage<Pair<HttpResponse<byte[]>, byte[]>> res;
|
||||
String now = now(start);
|
||||
if (t == null) {
|
||||
if (resp.statusCode() != 200) {
|
||||
String s = "Expected 200, got: " + resp.statusCode();
|
||||
System.out.println(now + s + " from "
|
||||
+ resp.request().uri().getPath());
|
||||
res = completedWithIOException(s);
|
||||
return res;
|
||||
} else {
|
||||
int counter = COUNT.incrementAndGet();
|
||||
System.out.println(now + "Result (" + counter + ") from "
|
||||
+ resp.request().uri().getPath());
|
||||
}
|
||||
res = CompletableFuture.completedStage(resp.body())
|
||||
.thenApply((b) -> new Pair<>(resp, b));
|
||||
return res;
|
||||
} else {
|
||||
int counter = COUNT.incrementAndGet();
|
||||
System.out.println(now + "Result (" + counter + ") from "
|
||||
+ r.uri().getPath()
|
||||
+ failure(t));
|
||||
res = CompletableFuture.failedFuture(t);
|
||||
return res;
|
||||
}
|
||||
})
|
||||
.thenCompose(c -> c)
|
||||
.thenAccept((pair) -> {
|
||||
HttpRequest request = pair.t.request();
|
||||
byte[] requestBody = bodies.get(request);
|
||||
check(Arrays.equals(requestBody, pair.u),
|
||||
"bodies not equal:[" + bytesToHexString(requestBody)
|
||||
+ "] [" + bytesToHexString(pair.u) + "]");
|
||||
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// wait for them all to complete and throw exception in case of error
|
||||
CompletableFuture.allOf(results).join();
|
||||
try {
|
||||
// wait for them all to complete and throw exception in case of error
|
||||
CompletableFuture.allOf(results).join();
|
||||
done = true;
|
||||
} catch (CompletionException e) {
|
||||
if (!Platform.isWindows()) throw e;
|
||||
if (LIMIT.get() < REQUESTS) throw e;
|
||||
Throwable cause = e;
|
||||
while ((cause = cause.getCause()) != null) {
|
||||
if (cause instanceof ConnectException) {
|
||||
// try again, limit concurrency by half
|
||||
COUNT.set(0);
|
||||
LIMIT.set(REQUESTS/2);
|
||||
System.out.println("*** Retrying due to " + cause);
|
||||
continue LOOP;
|
||||
}
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
} while (!done);
|
||||
}
|
||||
|
||||
static <T> CompletableFuture<T> completedWithIOException(String message) {
|
||||
@ -294,13 +368,7 @@ public class ManyRequestsLegacy {
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
static final class Pair<T,U> {
|
||||
Pair(T t, U u) {
|
||||
this.t = t; this.u = u;
|
||||
}
|
||||
T t;
|
||||
U u;
|
||||
}
|
||||
record Pair<T,U>(T t, U u) { }
|
||||
|
||||
/**
|
||||
* A simple limiter for controlling the number of requests to be run in
|
||||
|
||||
Loading…
Reference in New Issue
Block a user