russell@unturf.com
0a580b313d
Authors: russell@unturf.com · brackishbert@gmail.com · foxhop.net · TimeHexOn.com Patches, unit tests, benchmarks, whitepaper, and outreach briefs. Public domain — no copyright claimed. Use freely.
436 lines
16 KiB
Java
436 lines
16 KiB
Java
/*
|
|
* Copyright (c) 2015, 2026, 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
|
|
* @run junit/othervm
|
|
* -Djdk.internal.httpclient.debug=out
|
|
* PeerUniStreamDispatcherTest
|
|
* @summary Unit test for the PeerUniStreamDispatcher
|
|
*/
|
|
|
|
import java.io.IOException;
|
|
import java.nio.ByteBuffer;
|
|
import java.util.List;
|
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
|
import java.util.concurrent.CopyOnWriteArrayList;
|
|
|
|
import jdk.internal.net.http.common.Logger;
|
|
import jdk.internal.net.http.common.SequentialScheduler;
|
|
import jdk.internal.net.http.common.Utils;
|
|
import jdk.internal.net.http.http3.streams.Http3Streams;
|
|
import jdk.internal.net.http.http3.streams.PeerUniStreamDispatcher;
|
|
import jdk.internal.net.http.quic.VariableLengthEncoder;
|
|
import jdk.internal.net.http.quic.streams.QuicReceiverStream;
|
|
import jdk.internal.net.http.quic.streams.QuicStreamReader;
|
|
import jdk.internal.net.http.quic.streams.QuicStreams;
|
|
|
|
import static org.junit.jupiter.api.Assertions.*;
|
|
import org.junit.jupiter.api.Test;
|
|
|
|
public class PeerUniStreamDispatcherTest {
|
|
|
|
final Logger debug = Utils.getDebugLogger(() -> "PeerUniStreamDispatcherStub");
|
|
|
|
enum DISPATCHED_STREAM {
|
|
CONTROL, ENCODER, DECODER, PUSH, RESERVED, UNKNOWN
|
|
}
|
|
|
|
sealed interface DispatchedStream {
|
|
record StandardStream(DISPATCHED_STREAM type, String description, QuicReceiverStream stream)
|
|
implements DispatchedStream { }
|
|
record PushStream(DISPATCHED_STREAM type, String description, QuicReceiverStream stream, long pushId)
|
|
implements DispatchedStream { }
|
|
record UnknownStream(DISPATCHED_STREAM type, long code, QuicReceiverStream stream)
|
|
implements DispatchedStream { }
|
|
record ReservedStream(DISPATCHED_STREAM type, long code, QuicReceiverStream stream)
|
|
implements DispatchedStream { }
|
|
|
|
static DispatchedStream of(DISPATCHED_STREAM type, String description, QuicReceiverStream stream) {
|
|
return new StandardStream(type, description, stream);
|
|
}
|
|
static DispatchedStream of(DISPATCHED_STREAM type, String description, QuicReceiverStream stream, long pushId) {
|
|
return new PushStream(type, description, stream, pushId);
|
|
}
|
|
static DispatchedStream reserved(DISPATCHED_STREAM type, long code, QuicReceiverStream stream) {
|
|
return new ReservedStream(type, code, stream);
|
|
}
|
|
static DispatchedStream unknown(DISPATCHED_STREAM type, long code, QuicReceiverStream stream) {
|
|
return new UnknownStream(type, code, stream);
|
|
}
|
|
}
|
|
|
|
class PeerUniStreamDispatcherStub extends PeerUniStreamDispatcher {
|
|
|
|
final List<DispatchedStream> dispatched = new CopyOnWriteArrayList<>();
|
|
|
|
PeerUniStreamDispatcherStub(QuicReceiverStream stream) {
|
|
super(stream);
|
|
}
|
|
|
|
private void dispatched(DISPATCHED_STREAM type, String description, QuicReceiverStream stream) {
|
|
dispatched.add(DispatchedStream.of(type, description, stream));
|
|
}
|
|
|
|
private void dispatched(DISPATCHED_STREAM type, String description, QuicReceiverStream stream, long pushId) {
|
|
dispatched.add(DispatchedStream.of(type, description, stream, pushId));
|
|
}
|
|
|
|
private void dispatched(DISPATCHED_STREAM type, long code, QuicReceiverStream stream) {
|
|
dispatched.add(switch (type) {
|
|
case UNKNOWN -> DispatchedStream.unknown(type, code, stream);
|
|
case RESERVED -> DispatchedStream.reserved(type, code, stream);
|
|
default -> throw new IllegalArgumentException(String.valueOf(type));
|
|
});
|
|
}
|
|
|
|
@Override
|
|
protected Logger debug() {
|
|
return debug;
|
|
}
|
|
|
|
@Override
|
|
protected void onControlStreamCreated(String description, QuicReceiverStream stream) {
|
|
dispatched(DISPATCHED_STREAM.CONTROL, description, stream);
|
|
}
|
|
|
|
@Override
|
|
protected void onEncoderStreamCreated(String description, QuicReceiverStream stream) {
|
|
dispatched(DISPATCHED_STREAM.ENCODER, description, stream);
|
|
}
|
|
|
|
@Override
|
|
protected void onDecoderStreamCreated(String description, QuicReceiverStream stream) {
|
|
dispatched(DISPATCHED_STREAM.DECODER, description, stream);
|
|
}
|
|
|
|
@Override
|
|
protected void onPushStreamCreated(String description, QuicReceiverStream stream, long pushId) {
|
|
dispatched(DISPATCHED_STREAM.PUSH, description, stream, pushId);
|
|
}
|
|
|
|
@Override
|
|
protected void onReservedStreamType(long code, QuicReceiverStream stream) {
|
|
dispatched(DISPATCHED_STREAM.RESERVED, code, stream);
|
|
super.onReservedStreamType(code, stream);
|
|
}
|
|
|
|
@Override
|
|
protected void onUnknownStreamType(long code, QuicReceiverStream stream) {
|
|
dispatched(DISPATCHED_STREAM.UNKNOWN, code, stream);
|
|
super.onUnknownStreamType(code, stream);
|
|
}
|
|
|
|
@Override
|
|
public void start() {
|
|
super.start();
|
|
}
|
|
}
|
|
|
|
static class QuicReceiverStreamStub implements QuicReceiverStream {
|
|
|
|
class QuicStreamReaderStub extends QuicStreamReader {
|
|
|
|
volatile boolean connected, started;
|
|
QuicStreamReaderStub(SequentialScheduler scheduler) {
|
|
super(scheduler);
|
|
}
|
|
|
|
@Override
|
|
public ReceivingStreamState receivingState() {
|
|
return QuicReceiverStreamStub.this.receivingState();
|
|
}
|
|
|
|
@Override
|
|
public ByteBuffer poll() throws IOException {
|
|
return buffers.poll();
|
|
}
|
|
|
|
@Override
|
|
public ByteBuffer peek() throws IOException {
|
|
return buffers.peek();
|
|
}
|
|
|
|
@Override
|
|
public QuicReceiverStream stream() {
|
|
return QuicReceiverStreamStub.this;
|
|
}
|
|
|
|
@Override
|
|
public boolean connected() {
|
|
return connected;
|
|
}
|
|
|
|
@Override
|
|
public boolean started() {
|
|
return started;
|
|
}
|
|
|
|
@Override
|
|
public void start() {
|
|
started = true;
|
|
if (!buffers.isEmpty()) scheduler.runOrSchedule();
|
|
}
|
|
}
|
|
|
|
volatile QuicStreamReaderStub reader;
|
|
volatile SequentialScheduler scheduler;
|
|
volatile long errorCode;
|
|
final long streamId;
|
|
ConcurrentLinkedQueue<ByteBuffer> buffers = new ConcurrentLinkedQueue<>();
|
|
|
|
QuicReceiverStreamStub(long streamId) {
|
|
this.streamId = streamId;
|
|
}
|
|
|
|
|
|
@Override
|
|
public ReceivingStreamState receivingState() {
|
|
return ReceivingStreamState.RECV;
|
|
}
|
|
|
|
@Override
|
|
public QuicStreamReader connectReader(SequentialScheduler scheduler) {
|
|
this.scheduler = scheduler;
|
|
var reader = this.reader
|
|
= new QuicStreamReaderStub(scheduler);
|
|
reader.connected = true;
|
|
return reader;
|
|
}
|
|
|
|
@Override
|
|
public void disconnectReader(QuicStreamReader reader) {
|
|
this.scheduler = null;
|
|
this.reader = null;
|
|
((QuicStreamReaderStub) reader).connected = false;
|
|
}
|
|
|
|
@Override
|
|
public void requestStopSending(long errorCode) {
|
|
this.errorCode = errorCode;
|
|
}
|
|
|
|
@Override
|
|
public long dataReceived() {
|
|
return 0;
|
|
}
|
|
|
|
@Override
|
|
public long maxStreamData() {
|
|
return 0;
|
|
}
|
|
|
|
@Override
|
|
public long rcvErrorCode() {
|
|
return errorCode;
|
|
}
|
|
|
|
@Override
|
|
public long streamId() {
|
|
return streamId;
|
|
}
|
|
|
|
@Override
|
|
public StreamMode mode() {
|
|
return StreamMode.READ_ONLY;
|
|
}
|
|
|
|
@Override
|
|
public boolean isClientInitiated() {
|
|
return QuicStreams.isClientInitiated(streamId);
|
|
}
|
|
|
|
@Override
|
|
public boolean isServerInitiated() {
|
|
return QuicStreams.isServerInitiated(streamId);
|
|
}
|
|
|
|
@Override
|
|
public boolean isBidirectional() {
|
|
return QuicStreams.isBidirectional(streamId);
|
|
}
|
|
|
|
@Override
|
|
public boolean isLocalInitiated() {
|
|
return isClientInitiated();
|
|
}
|
|
|
|
@Override
|
|
public boolean isRemoteInitiated() {
|
|
return !isClientInitiated();
|
|
}
|
|
|
|
@Override
|
|
public int type() {
|
|
return QuicStreams.streamType(streamId);
|
|
}
|
|
|
|
@Override
|
|
public StreamState state() {
|
|
return ReceivingStreamState.RECV;
|
|
}
|
|
|
|
}
|
|
|
|
private void simpleStreamType(DISPATCHED_STREAM type, long code) {
|
|
System.out.println("Testing " + type + " with " + code);
|
|
QuicReceiverStreamStub stream = new QuicReceiverStreamStub(QuicStreams.UNI_MASK + QuicStreams.SRV_MASK);
|
|
PeerUniStreamDispatcherStub dispatcher = new PeerUniStreamDispatcherStub(stream);
|
|
QuicStreamReader reader = stream.reader;
|
|
SequentialScheduler scheduler = stream.scheduler;
|
|
assertTrue(reader.connected());
|
|
int size = VariableLengthEncoder.getEncodedSize(code);
|
|
ByteBuffer buffer = ByteBuffer.allocate(size);
|
|
assertEquals(size, buffer.remaining());
|
|
VariableLengthEncoder.encode(buffer, code);
|
|
buffer.flip();
|
|
stream.buffers.add(buffer);
|
|
scheduler.runOrSchedule();
|
|
dispatcher.start();
|
|
if (type == DISPATCHED_STREAM.PUSH) {
|
|
// we want to encode the pushId in multiple buffers, but call
|
|
// the scheduler only once to check that the dispatcher
|
|
// will loop correctly.
|
|
size = VariableLengthEncoder.getEncodedSize(1L << 62 - 5);
|
|
ByteBuffer buffer2 = ByteBuffer.allocate(size);
|
|
assertEquals(size, buffer2.remaining());
|
|
VariableLengthEncoder.encode(buffer2, 1L << 62 - 5);
|
|
buffer2.flip();
|
|
stream.buffers.add(ByteBuffer.wrap(new byte[] {buffer2.get()}));
|
|
scheduler.runOrSchedule(); // call runOrSchedule after supplying the first byte.
|
|
assertTrue(reader.connected());
|
|
assert buffer2.remaining() > 1; // should always be true
|
|
while (buffer2.hasRemaining()) {
|
|
stream.buffers.add(ByteBuffer.wrap(new byte[] {buffer2.get()}));
|
|
}
|
|
}
|
|
scheduler.runOrSchedule();
|
|
assertFalse(reader.connected());
|
|
assertFalse(dispatcher.dispatched.isEmpty());
|
|
assertTrue(stream.buffers.isEmpty());
|
|
assertEquals(1, dispatcher.dispatched.size());
|
|
var dispatched = dispatcher.dispatched.get(0);
|
|
checkDispatched(type, code, stream, dispatched);
|
|
}
|
|
|
|
private void checkDispatched(DISPATCHED_STREAM type,
|
|
long code,
|
|
QuicReceiverStream stream,
|
|
DispatchedStream dispatched) {
|
|
var streamClass = switch (type) {
|
|
case CONTROL, ENCODER, DECODER -> DispatchedStream.StandardStream.class;
|
|
case PUSH -> DispatchedStream.PushStream.class;
|
|
case RESERVED -> DispatchedStream.ReservedStream.class;
|
|
case UNKNOWN -> DispatchedStream.UnknownStream.class;
|
|
};
|
|
assertEquals(streamClass, dispatched.getClass(),
|
|
"unexpected dispatched class " + dispatched + " for " + type);
|
|
if (dispatched instanceof DispatchedStream.StandardStream st) {
|
|
System.out.println("Got expected stream: " + st);
|
|
assertEquals(type, st.type());
|
|
assertEquals(stream, st.stream);
|
|
} else if (dispatched instanceof DispatchedStream.ReservedStream res) {
|
|
System.out.println("Got expected stream: " + res);
|
|
assertEquals(type, res.type());
|
|
assertEquals(stream, res.stream);
|
|
assertEquals(code, res.code());
|
|
assertTrue(Http3Streams.isReserved(res.code()));
|
|
} else if (dispatched instanceof DispatchedStream.UnknownStream unk) {
|
|
System.out.println("Got expected stream: " + unk);
|
|
assertEquals(type, unk.type());
|
|
assertEquals(stream, unk.stream);
|
|
assertEquals(code, unk.code());
|
|
assertFalse(Http3Streams.isReserved(unk.code()));
|
|
} else if (dispatched instanceof DispatchedStream.PushStream push) {
|
|
System.out.println("Got expected stream: " + push);
|
|
assertEquals(type, push.type());
|
|
assertEquals(stream, push.stream);
|
|
assertEquals(1L << 62 - 5, push.pushId);
|
|
assertEquals(DISPATCHED_STREAM.PUSH, push.type());
|
|
}
|
|
|
|
}
|
|
@Test
|
|
public void simpleControl() {
|
|
simpleStreamType(DISPATCHED_STREAM.CONTROL, Http3Streams.CONTROL_STREAM_CODE);
|
|
}
|
|
@Test
|
|
public void simpleDecoder() {
|
|
simpleStreamType(DISPATCHED_STREAM.DECODER, Http3Streams.QPACK_DECODER_STREAM_CODE);
|
|
}
|
|
@Test
|
|
public void simpleEncoder() {
|
|
simpleStreamType(DISPATCHED_STREAM.ENCODER, Http3Streams.QPACK_ENCODER_STREAM_CODE);
|
|
}
|
|
@Test
|
|
public void simplePush() {
|
|
simpleStreamType(DISPATCHED_STREAM.PUSH, Http3Streams.PUSH_STREAM_CODE);
|
|
}
|
|
@Test
|
|
public void simpleUknown() {
|
|
simpleStreamType(DISPATCHED_STREAM.UNKNOWN, VariableLengthEncoder.MAX_ENCODED_INTEGER);
|
|
}
|
|
@Test
|
|
public void simpleReserved() {
|
|
simpleStreamType(DISPATCHED_STREAM.RESERVED, 31 * 256 + 2);
|
|
}
|
|
|
|
@Test
|
|
public void multyBytes() {
|
|
DISPATCHED_STREAM type = DISPATCHED_STREAM.UNKNOWN;
|
|
long code = VariableLengthEncoder.MAX_ENCODED_INTEGER;
|
|
System.out.println("Testing multi byte " + type + " with " + code);
|
|
QuicReceiverStreamStub stream = new QuicReceiverStreamStub(QuicStreams.UNI_MASK + QuicStreams.SRV_MASK);
|
|
PeerUniStreamDispatcherStub dispatcher = new PeerUniStreamDispatcherStub(stream);
|
|
QuicStreamReader reader = stream.reader;
|
|
SequentialScheduler scheduler = stream.scheduler;
|
|
assertTrue(reader.connected());
|
|
int size = VariableLengthEncoder.getEncodedSize(code);
|
|
assertEquals(8, size);
|
|
ByteBuffer buffer = ByteBuffer.allocate(size);
|
|
assertEquals(size, buffer.remaining());
|
|
VariableLengthEncoder.encode(buffer, code);
|
|
buffer.flip();
|
|
dispatcher.start();
|
|
for (int i=0; i<size; i++) {
|
|
System.out.printf("Submitting buffer[%s]=%s%n", i, buffer.get(buffer.position()) & 0xFF);
|
|
ByteBuffer buf = ByteBuffer.allocate(1);
|
|
buf.put(buffer.get());
|
|
buf.flip();
|
|
stream.buffers.add(buf);
|
|
scheduler.runOrSchedule();
|
|
if (i + 1 < size) {
|
|
assertTrue(reader.connected());
|
|
assertTrue(dispatcher.dispatched.isEmpty());
|
|
}
|
|
|
|
}
|
|
assertFalse(reader.connected());
|
|
assertFalse(dispatcher.dispatched.isEmpty());
|
|
assertTrue(stream.buffers.isEmpty());
|
|
assertEquals(1, dispatcher.dispatched.size());
|
|
var dispatched = dispatcher.dispatched.get(0);
|
|
checkDispatched(type, code, stream, dispatched);
|
|
}
|
|
|
|
}
|