a3b61fd4c7
Co-authored-by: Daniel Fuchs <daniel.fuchs@oracle.com> Co-authored-by: Michael McMahon <michael.x.mcmahon@oracle.com> Co-authored-by: Pavel Rappo <pavel.rappo@oracle.com> Reviewed-by: chegar, dfuchs, michaelm, prappo
475 lines
20 KiB
Java
475 lines
20 KiB
Java
/*
|
|
* Copyright (c) 2017, 2018, 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.
|
|
*/
|
|
|
|
import java.nio.ByteBuffer;
|
|
import java.util.List;
|
|
import java.util.Random;
|
|
import java.util.concurrent.CompletableFuture;
|
|
import java.util.concurrent.CompletionStage;
|
|
import java.util.concurrent.Executor;
|
|
import java.util.concurrent.ExecutorService;
|
|
import java.util.concurrent.Executors;
|
|
import java.util.concurrent.Flow;
|
|
import java.util.concurrent.Flow.Subscription;
|
|
import java.util.concurrent.SubmissionPublisher;
|
|
import java.util.function.BiConsumer;
|
|
import java.net.http.HttpResponse.BodyHandler;
|
|
import java.net.http.HttpResponse.BodyHandlers;
|
|
import java.net.http.HttpResponse.BodySubscriber;
|
|
import java.net.http.HttpResponse.BodySubscribers;
|
|
import jdk.test.lib.RandomFactory;
|
|
import org.testng.annotations.DataProvider;
|
|
import org.testng.annotations.Test;
|
|
import static java.lang.Long.MAX_VALUE;
|
|
import static java.lang.Long.min;
|
|
import static java.lang.System.out;
|
|
import static java.util.concurrent.CompletableFuture.delayedExecutor;
|
|
import static java.util.concurrent.TimeUnit.MILLISECONDS;
|
|
import static org.testng.Assert.*;
|
|
|
|
/*
|
|
* @test
|
|
* @bug 8184285
|
|
* @summary Direct test for HttpResponse.BodySubscriber.buffering() API
|
|
* @key randomness
|
|
* @library /test/lib
|
|
* @build jdk.test.lib.RandomFactory
|
|
* @run testng/othervm -Djdk.internal.httpclient.debug=true BufferingSubscriberTest
|
|
*/
|
|
|
|
public class BufferingSubscriberTest {
|
|
|
|
// If we compute that a test will take less that 10s
|
|
// we judge it acceptable
|
|
static final long LOWER_THRESHOLD = 10_000; // 10 sec.
|
|
// If we compute that a test will take more than 20 sec
|
|
// we judge it problematic: we will try to adjust the
|
|
// buffer sizes, and if we can't we will print a warning
|
|
static final long UPPER_THRESHOLD = 20_000; // 20 sec.
|
|
|
|
static final Random random = RandomFactory.getRandom();
|
|
static final long start = System.nanoTime();
|
|
static final String START = "start";
|
|
static final String END = "end ";
|
|
static long elapsed() { return (System.nanoTime() - start)/1000_000;}
|
|
static void printStamp(String what, String fmt, Object... args) {
|
|
long elapsed = elapsed();
|
|
long sec = elapsed/1000;
|
|
long ms = elapsed % 1000;
|
|
String time = sec > 0 ? sec + "sec " : "";
|
|
time = time + ms + "ms";
|
|
out.println(what + "\t ["+time+"]\t "+ String.format(fmt,args));
|
|
}
|
|
@DataProvider(name = "negatives")
|
|
public Object[][] negatives() {
|
|
return new Object[][] { { 0 }, { -1 }, { -1000 } };
|
|
}
|
|
|
|
@Test(dataProvider = "negatives", expectedExceptions = IllegalArgumentException.class)
|
|
public void subscriberThrowsIAE(int bufferSize) {
|
|
printStamp(START, "subscriberThrowsIAE(%d)", bufferSize);
|
|
try {
|
|
BodySubscriber<?> bp = BodySubscribers.ofByteArray();
|
|
BodySubscribers.buffering(bp, bufferSize);
|
|
} finally {
|
|
printStamp(END, "subscriberThrowsIAE(%d)", bufferSize);
|
|
}
|
|
}
|
|
|
|
@Test(dataProvider = "negatives", expectedExceptions = IllegalArgumentException.class)
|
|
public void handlerThrowsIAE(int bufferSize) {
|
|
printStamp(START, "handlerThrowsIAE(%d)", bufferSize);
|
|
try {
|
|
BodyHandler<?> bp = BodyHandlers.ofByteArray();
|
|
BodyHandlers.buffering(bp, bufferSize);
|
|
} finally {
|
|
printStamp(END, "handlerThrowsIAE(%d)", bufferSize);
|
|
}
|
|
}
|
|
|
|
// ---
|
|
|
|
@DataProvider(name = "config")
|
|
public Object[][] config() {
|
|
return new Object[][] {
|
|
// iterations delayMillis numBuffers bufferSize maxBufferSize minBufferSize
|
|
{ 1, 0, 1, 1, 2, 1 },
|
|
{ 1, 5, 1, 100, 2, 1 },
|
|
{ 1, 0, 1, 10, 1000, 1 },
|
|
{ 1, 10, 1, 10, 1000, 1 },
|
|
{ 1, 0, 1, 1000, 1000, 10 },
|
|
{ 1, 0, 10, 1000, 1000, 50 },
|
|
{ 1, 0, 1000, 10 , 1000, 50 },
|
|
{ 1, 100, 1, 1000 * 4, 1000, 50 },
|
|
{ 100, 0, 1000, 1, 2, 1 },
|
|
{ 3, 0, 4, 5006, 1000, 50 },
|
|
{ 20, 0, 100, 4888, 1000, 100 },
|
|
{ 16, 10, 1000, 50 , 1000, 100 },
|
|
{ 16, 10, 1000, 50 , 657, 657 },
|
|
};
|
|
}
|
|
|
|
@Test(dataProvider = "config")
|
|
public void test(int iterations,
|
|
int delayMillis,
|
|
int numBuffers,
|
|
int bufferSize,
|
|
int maxBufferSize,
|
|
int minbufferSize) {
|
|
for (long perRequestAmount : new long[] { 1L, MAX_VALUE })
|
|
test(iterations,
|
|
delayMillis,
|
|
numBuffers,
|
|
bufferSize,
|
|
maxBufferSize,
|
|
minbufferSize,
|
|
perRequestAmount);
|
|
}
|
|
|
|
volatile boolean onNextThrew;
|
|
|
|
BiConsumer<Flow.Subscriber<?>, ? super Throwable> onNextThrowHandler =
|
|
(sub, ex) -> {
|
|
onNextThrew = true;
|
|
System.out.println("onNext threw " + ex);
|
|
ex.printStackTrace();
|
|
};
|
|
|
|
public void test(int iterations,
|
|
int delayMillis,
|
|
int numBuffers,
|
|
int bufferSize,
|
|
int maxBufferSize,
|
|
int minBufferSize,
|
|
long requestAmount) {
|
|
ExecutorService executor = Executors.newFixedThreadPool(1);
|
|
try {
|
|
out.printf("Iterations %d\n", iterations);
|
|
for (int i=0; i<iterations; i++ ) {
|
|
printStamp(START, "Iteration %d", i);
|
|
try {
|
|
SubmissionPublisher<List<ByteBuffer>> publisher =
|
|
new SubmissionPublisher<>(executor,
|
|
1, // lock-step with the publisher, for now
|
|
onNextThrowHandler);
|
|
CompletableFuture<?> cf = sink(publisher,
|
|
delayMillis,
|
|
numBuffers * bufferSize,
|
|
requestAmount,
|
|
maxBufferSize,
|
|
minBufferSize);
|
|
source(publisher, numBuffers, bufferSize);
|
|
publisher.close();
|
|
cf.join();
|
|
} finally {
|
|
printStamp(END, "Iteration %d\n", i);
|
|
}
|
|
}
|
|
|
|
assertFalse(onNextThrew, "Unexpected onNextThrew, check output");
|
|
|
|
out.println("OK");
|
|
} finally {
|
|
executor.shutdown();
|
|
}
|
|
}
|
|
|
|
static long accumulatedDataSize(List<ByteBuffer> bufs) {
|
|
return bufs.stream().mapToLong(ByteBuffer::remaining).sum();
|
|
}
|
|
|
|
/** Returns a new BB with its contents set to monotonically increasing
|
|
* values, staring at the given start index and wrapping every 100. */
|
|
static ByteBuffer allocateBuffer(int size, int startIdx) {
|
|
ByteBuffer b = ByteBuffer.allocate(size);
|
|
for (int i=0; i<size; i++)
|
|
b.put((byte)((startIdx + i) % 100));
|
|
b.position(0);
|
|
return b;
|
|
}
|
|
|
|
static class TestSubscriber implements BodySubscriber<Integer> {
|
|
final int delayMillis;
|
|
final int bufferSize;
|
|
final int expectedTotalSize;
|
|
final long requestAmount;
|
|
final CompletableFuture<Integer> completion;
|
|
final Executor delayedExecutor;
|
|
volatile Flow.Subscription subscription;
|
|
|
|
TestSubscriber(int bufferSize,
|
|
int delayMillis,
|
|
int expectedTotalSize,
|
|
long requestAmount) {
|
|
this.bufferSize = bufferSize;
|
|
this.completion = new CompletableFuture<>();
|
|
this.delayMillis = delayMillis;
|
|
this.delayedExecutor = delayedExecutor(delayMillis, MILLISECONDS);
|
|
this.expectedTotalSize = expectedTotalSize;
|
|
this.requestAmount = requestAmount;
|
|
}
|
|
|
|
/**
|
|
* Example of a factory method which would decorate a buffering
|
|
* subscriber to create a new subscriber dependent on buffering capability.
|
|
* <p>
|
|
* The integer type parameter simulates the body just by counting the
|
|
* number of bytes in the body.
|
|
*/
|
|
static BodySubscriber<Integer> createSubscriber(int bufferSize,
|
|
int delay,
|
|
int expectedTotalSize,
|
|
long requestAmount) {
|
|
TestSubscriber s = new TestSubscriber(bufferSize,
|
|
delay,
|
|
expectedTotalSize,
|
|
requestAmount);
|
|
return BodySubscribers.buffering(s, bufferSize);
|
|
}
|
|
|
|
private void requestMore() {
|
|
subscription.request(requestAmount);
|
|
}
|
|
|
|
@Override
|
|
public void onSubscribe(Subscription subscription) {
|
|
assertNull(this.subscription);
|
|
this.subscription = subscription;
|
|
if (delayMillis > 0)
|
|
delayedExecutor.execute(this::requestMore);
|
|
else
|
|
requestMore();
|
|
}
|
|
|
|
volatile int wrongSizes;
|
|
volatile int totalBytesReceived;
|
|
volatile int onNextInvocations;
|
|
volatile long lastSeenSize = -1;
|
|
volatile boolean noMoreOnNext; // false
|
|
volatile int index; // 0
|
|
volatile long count;
|
|
|
|
@Override
|
|
public void onNext(List<ByteBuffer> items) {
|
|
try {
|
|
long sz = accumulatedDataSize(items);
|
|
boolean printStamp = delayMillis > 0
|
|
&& requestAmount < Long.MAX_VALUE
|
|
&& count % 20 == 0;
|
|
if (printStamp) {
|
|
printStamp("stamp", "count=%d sz=%d accumulated=%d",
|
|
count, sz, (totalBytesReceived + sz));
|
|
}
|
|
count++;
|
|
onNextInvocations++;
|
|
assertNotEquals(sz, 0L, "Unexpected empty buffers");
|
|
items.stream().forEach(b -> assertEquals(b.position(), 0));
|
|
assertFalse(noMoreOnNext);
|
|
|
|
if (sz != bufferSize) {
|
|
String msg = sz + ", should be less than bufferSize, " + bufferSize;
|
|
assertTrue(sz < bufferSize, msg);
|
|
assertTrue(lastSeenSize == -1 || lastSeenSize == bufferSize);
|
|
noMoreOnNext = true;
|
|
wrongSizes++;
|
|
printStamp("onNext",
|
|
"Possibly received last buffer: sz=%d, accumulated=%d, total=%d",
|
|
sz, totalBytesReceived, totalBytesReceived + sz);
|
|
} else {
|
|
assertEquals(sz, bufferSize, "Expected to receive exactly bufferSize");
|
|
}
|
|
lastSeenSize = sz;
|
|
|
|
// Ensure expected contents
|
|
for (ByteBuffer b : items) {
|
|
while (b.hasRemaining()) {
|
|
assertEquals(b.get(), (byte) (index % 100));
|
|
index++;
|
|
}
|
|
}
|
|
|
|
totalBytesReceived += sz;
|
|
assertEquals(totalBytesReceived, index);
|
|
if (delayMillis > 0 && ((expectedTotalSize - totalBytesReceived) > bufferSize))
|
|
delayedExecutor.execute(this::requestMore);
|
|
else
|
|
requestMore();
|
|
} catch (Throwable t) {
|
|
completion.completeExceptionally(t);
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public void onError(Throwable throwable) {
|
|
completion.completeExceptionally(throwable);
|
|
}
|
|
|
|
@Override
|
|
public void onComplete() {
|
|
if (wrongSizes > 1) { // allow just the final item to be smaller
|
|
String msg = "Wrong sizes. Expected no more than 1. [" + this + "]";
|
|
completion.completeExceptionally(new Throwable(msg));
|
|
}
|
|
if (totalBytesReceived != expectedTotalSize) {
|
|
String msg = "Wrong number of bytes. [" + this + "]";
|
|
completion.completeExceptionally(new Throwable(msg));
|
|
} else {
|
|
completion.complete(totalBytesReceived);
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public CompletionStage<Integer> getBody() {
|
|
return completion;
|
|
}
|
|
|
|
@Override
|
|
public String toString() {
|
|
StringBuilder sb = new StringBuilder();
|
|
sb.append(super.toString());
|
|
sb.append(", bufferSize=").append(bufferSize);
|
|
sb.append(", onNextInvocations=").append(onNextInvocations);
|
|
sb.append(", totalBytesReceived=").append(totalBytesReceived);
|
|
sb.append(", expectedTotalSize=").append(expectedTotalSize);
|
|
sb.append(", requestAmount=").append(requestAmount);
|
|
sb.append(", lastSeenSize=").append(lastSeenSize);
|
|
sb.append(", wrongSizes=").append(wrongSizes);
|
|
sb.append(", index=").append(index);
|
|
return sb.toString();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Publishes data, through the given publisher, using the main thread.
|
|
*
|
|
* Note: The executor supplied when creating the SubmissionPublisher provides
|
|
* the threads for executing the Subscribers.
|
|
*
|
|
* @param publisher the publisher
|
|
* @param numBuffers the number of buffers to send ( before splitting in two )
|
|
* @param bufferSize the total size of the data to send ( before splitting in two )
|
|
*/
|
|
static void source(SubmissionPublisher<List<ByteBuffer>> publisher,
|
|
int numBuffers,
|
|
int bufferSize) {
|
|
printStamp("source","Publishing %d buffers of size %d each", numBuffers, bufferSize);
|
|
int index = 0;
|
|
for (int i=0; i<numBuffers; i++) {
|
|
int chunkSize = random.nextInt(bufferSize);
|
|
ByteBuffer buf1 = allocateBuffer(chunkSize, index);
|
|
index += chunkSize;
|
|
ByteBuffer buf2 = allocateBuffer(bufferSize - chunkSize, index);
|
|
index += bufferSize - chunkSize;
|
|
publisher.submit(List.of(buf1, buf2));
|
|
}
|
|
printStamp("source", "complete");
|
|
}
|
|
|
|
/**
|
|
* Creates and subscribes Subscribers that receive data from the given
|
|
* publisher.
|
|
*
|
|
* @param publisher the publisher
|
|
* @param delayMillis time, in milliseconds, to delay the Subscription
|
|
* requesting more bytes ( for simulating slow consumption )
|
|
* @param expectedTotalSize the total number of bytes expected to be received
|
|
* by the subscribers
|
|
* @return a CompletableFuture which completes when the subscription is complete
|
|
*/
|
|
static CompletableFuture<?> sink(SubmissionPublisher<List<ByteBuffer>> publisher,
|
|
int delayMillis,
|
|
int expectedTotalSize,
|
|
long requestAmount,
|
|
int maxBufferSize,
|
|
int minBufferSize) {
|
|
int bufferSize = chooseBufferSize(maxBufferSize,
|
|
minBufferSize,
|
|
delayMillis,
|
|
expectedTotalSize,
|
|
requestAmount);
|
|
assert bufferSize > 0;
|
|
assert bufferSize >= minBufferSize;
|
|
assert bufferSize <= maxBufferSize;
|
|
BodySubscriber<Integer> sub = TestSubscriber.createSubscriber(bufferSize,
|
|
delayMillis,
|
|
expectedTotalSize,
|
|
requestAmount);
|
|
publisher.subscribe(sub);
|
|
printStamp("sink","Subscriber reads data with buffer size: %d", bufferSize);
|
|
out.printf("Subscription delay is %d msec\n", delayMillis);
|
|
long delay = (((long)delayMillis * expectedTotalSize) / bufferSize) / requestAmount;
|
|
out.printf("Minimum total delay is %d sec %d ms\n", delay / 1000, delay % 1000);
|
|
out.printf("Request amount is %d items\n", requestAmount);
|
|
return sub.getBody().toCompletableFuture();
|
|
}
|
|
|
|
static int chooseBufferSize(int maxBufferSize,
|
|
int minBufferSize,
|
|
int delaysMillis,
|
|
int expectedTotalSize,
|
|
long requestAmount) {
|
|
assert minBufferSize > 0 && maxBufferSize > 0 && requestAmount > 0;
|
|
int bufferSize = maxBufferSize == minBufferSize ? maxBufferSize :
|
|
(random.nextInt(maxBufferSize - minBufferSize)
|
|
+ minBufferSize);
|
|
if (requestAmount == Long.MAX_VALUE) return bufferSize;
|
|
long minDelay = (((long)delaysMillis * expectedTotalSize) / maxBufferSize)
|
|
/ requestAmount;
|
|
long maxDelay = (((long)delaysMillis * expectedTotalSize) / minBufferSize)
|
|
/ requestAmount;
|
|
// if the maximum delay is < 10s just take a random number between min and max.
|
|
if (maxDelay <= LOWER_THRESHOLD) {
|
|
return bufferSize;
|
|
}
|
|
// if minimum delay is greater than 20s then print a warning and use max buffer.
|
|
if (minDelay >= UPPER_THRESHOLD) {
|
|
System.out.println("Warning: minimum delay is "
|
|
+ minDelay/1000 + "sec " + minDelay%1000 + "ms");
|
|
System.err.println("Warning: minimum delay is "
|
|
+ minDelay/1000 + "sec " + minDelay%1000 + "ms");
|
|
return maxBufferSize;
|
|
}
|
|
// maxDelay could be anything, but minDelay is below the UPPER_THRESHOLD
|
|
// try to pick up a buffer size that keeps the delay below the
|
|
// UPPER_THRESHOLD
|
|
while (minBufferSize < maxBufferSize) {
|
|
bufferSize = random.nextInt(maxBufferSize - minBufferSize)
|
|
+ minBufferSize;
|
|
long delay = (((long)delaysMillis * expectedTotalSize) / bufferSize)
|
|
/ requestAmount;
|
|
if (delay < UPPER_THRESHOLD) return bufferSize;
|
|
minBufferSize++;
|
|
}
|
|
return minBufferSize;
|
|
}
|
|
|
|
// ---
|
|
|
|
/* Main entry point for standalone testing of the main functional test. */
|
|
public static void main(String... args) {
|
|
BufferingSubscriberTest t = new BufferingSubscriberTest();
|
|
for (Object[] objs : t.config())
|
|
t.test((int)objs[0], (int)objs[1], (int)objs[2], (int)objs[3], (int)objs[4], (int)objs[5]);
|
|
}
|
|
}
|