jdk-sandbox: comparison src/java.net.http/share/classes/jdk/internal/net/http/common/SSLFlowDelegate.java

equal deleted inserted replaced

-:8e1ed2a15845
+:4690a2871b44
 import javax.net.ssl.SSLEngineResult;
 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
 import javax.net.ssl.SSLEngineResult.Status;
 import javax.net.ssl.SSLException;
 import java.io.IOException;
-import java.lang.System.Logger.Level;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 /**
 * Implements SSL using two SubscriberWrappers.
 *
 * <p> Constructor takes two Flow.Subscribers: one that receives the network
 final CompletableFuture<String> alpnCF; // completes on initial handshake
 final static ByteBuffer SENTINEL = Utils.EMPTY_BYTEBUFFER;
 volatile boolean close_notify_received;
 final CompletableFuture<Void> readerCF;
 final CompletableFuture<Void> writerCF;
+final Consumer<ByteBuffer> recycler;
 static AtomicInteger scount = new AtomicInteger(1);
 final int id;
 /**
 * Creates an SSLFlowDelegate fed from two Flow.Subscribers. Each
 public SSLFlowDelegate(SSLEngine engine,
 Executor exec,
 Subscriber<? super List<ByteBuffer>> downReader,
 Subscriber<? super List<ByteBuffer>> downWriter)
 {
+this(engine, exec, null, downReader, downWriter);
+}
+/**
+* Creates an SSLFlowDelegate fed from two Flow.Subscribers. Each
+* Flow.Subscriber requires an associated {@link CompletableFuture}
+* for errors that need to be signaled from downstream to upstream.
+*/
+public SSLFlowDelegate(SSLEngine engine,
+Executor exec,
+Consumer<ByteBuffer> recycler,
+Subscriber<? super List<ByteBuffer>> downReader,
+Subscriber<? super List<ByteBuffer>> downWriter)
+{
 this.id = scount.getAndIncrement();
 this.tubeName = String.valueOf(downWriter);
+this.recycler = recycler;
 this.reader = new Reader();
 this.writer = new Writer();
 this.engine = engine;
 this.exec = exec;
 this.handshakeState = new AtomicInteger(NOT_HANDSHAKING);
 public String monitor() {
 StringBuilder sb = new StringBuilder();
 sb.append("SSL: id ").append(id);
 sb.append(" HS state: " + states(handshakeState));
 sb.append(" Engine state: " + engine.getHandshakeStatus().toString());
-sb.append(" LL : ");
+if (stateList != null) {
-for (String s: stateList) {
+sb.append(" LL : ");
-sb.append(s).append(" ");
+for (String s : stateList) {
+sb.append(s).append(" ");
+}
 }
 sb.append("\r\n");
 sb.append("Reader:: ").append(reader.toString());
 sb.append("\r\n");
 sb.append("Writer:: ").append(writer.toString());
 *     OK: return generated buffers.
 *
 * Upstream subscription strategy is to try and keep no more than
 * TARGET_BUFSIZE bytes in readBuf
 */
-class Reader extends SubscriberWrapper {
+final class Reader extends SubscriberWrapper implements FlowTube.TubeSubscriber {
+// Maximum record size is 16k.
+// Because SocketTube can feeds us up to 3 16K buffers,
+// then setting this size to 16K means that the readBuf
+// can store up to 64K-1 (16K-1 + 3*16K)
+static final int TARGET_BUFSIZE = 16 * 1024;
 final SequentialScheduler scheduler;
-static final int TARGET_BUFSIZE = 16 * 1024;
 volatile ByteBuffer readBuf;
 volatile boolean completing;
 final Object readBufferLock = new Object();
 final Logger debugr = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);
-class ReaderDownstreamPusher implements Runnable {
+private final class ReaderDownstreamPusher implements Runnable {
 @Override public void run() { processData(); }
 }
 Reader() {
 super();
 scheduler = SequentialScheduler.synchronizedScheduler(
 new ReaderDownstreamPusher());
 this.readBuf = ByteBuffer.allocate(1024);
 readBuf.limit(0); // keep in read mode
+}
+@Override
+public boolean supportsRecycling() {
+return recycler != null;
 }
 protected SchedulingAction enterScheduling() {
 return enterReadScheduling();
 }
 public void incoming(List<ByteBuffer> buffers, boolean complete) {
 if (debugr.on())
 debugr.log("Adding %d bytes to read buffer",
 Utils.remaining(buffers));
 addToReadBuf(buffers, complete);
-scheduler.runOrSchedule();
+scheduler.runOrSchedule(exec);
 }
 @Override
 public String toString() {
 return "READER: " + super.toString() + " readBuf: " + readBuf.toString()
 }
 @Override
 protected long upstreamWindowUpdate(long currentWindow, long downstreamQsize) {
 if (readBuf.remaining() > TARGET_BUFSIZE) {
+if (debugr.on())
+debugr.log("readBuf has more than TARGET_BUFSIZE: %d",
+readBuf.remaining());
 return 0;
 } else {
 return super.upstreamWindowUpdate(currentWindow, downstreamQsize);
 }
 }
 readBuf.compact();
 while (readBuf.remaining() < buf.remaining())
 reallocReadBuf();
 readBuf.put(buf);
 readBuf.flip();
+// should be safe to call inside lock
+// since the only implementation
+// offers the buffer to an unbounded queue.
+// WARNING: do not touch buf after this point!
+if (recycler != null) recycler.accept(buf);
 }
 if (complete) {
 this.completing = complete;
 }
 }
 }
 void schedule() {
-scheduler.runOrSchedule();
+scheduler.runOrSchedule(exec);
 }
 void stop() {
 if (debugr.on()) debugr.log("stop");
 scheduler.stop();
 }
 AtomicInteger count = new AtomicInteger(0);
+// minimum number of bytes required to call unwrap.
+// Usually this is 0, unless there was a buffer underflow.
+// In this case we need to wait for more bytes than what
+// we had before calling unwrap() again.
+volatile int minBytesRequired;
 // work function where it all happens
-void processData() {
+final void processData() {
 try {
 if (debugr.on())
 debugr.log("processData:"
 + " readBuf remaining:" + readBuf.remaining()
 + ", state:" + states(handshakeState)
 + ", engine handshake status:" + engine.getHandshakeStatus());
 int len;
 boolean complete = false;
-while ((len = readBuf.remaining()) > 0) {
+while (readBuf.remaining() > (len = minBytesRequired)) {
 boolean handshaking = false;
 try {
 EngineResult result;
 synchronized (readBufferLock) {
 complete = this.completing;
+if (debugr.on()) debugr.log("Unwrapping: %s", readBuf.remaining());
+// Unless there is a BUFFER_UNDERFLOW, we should try to
+// unwrap any number of bytes. Set minBytesRequired to 0:
+// we only need to do that if minBytesRequired is not already 0.
+len = len > 0 ? minBytesRequired = 0 : len;
 result = unwrapBuffer(readBuf);
-if (debugr.on())
+len = readBuf.remaining();
-debugr.log("Unwrapped: %s", result.result);
+if (debugr.on()) {
+debugr.log("Unwrapped: result: %s", result.result);
+debugr.log("Unwrapped: consumed: %s", result.bytesConsumed());
+}
 }
 if (result.bytesProduced() > 0) {
 if (debugr.on())
 debugr.log("sending %d", result.bytesProduced());
 count.addAndGet(result.bytesProduced());
 outgoing(result.destBuffer, false);
 }
 if (result.status() == Status.BUFFER_UNDERFLOW) {
 if (debugr.on()) debugr.log("BUFFER_UNDERFLOW");
 // not enough data in the read buffer...
+// no need to try to unwrap again unless we get more bytes
+// than minBytesRequired = len in the read buffer.
+minBytesRequired = len;
+synchronized (readBufferLock) {
+// more bytes could already have been added...
+assert readBuf.remaining() >= len;
+// check if we have received some data, and if so
+// we can just re-spin the loop
+if (readBuf.remaining() > len) continue;
+}
+// request more data and return.
 requestMore();
-synchronized (readBufferLock) {
+return;
-// check if we have received some data
-if (readBuf.remaining() > len) continue;
-return;
-}
 }
 if (complete && result.status() == Status.CLOSED) {
 if (debugr.on()) debugr.log("Closed: completing");
 outgoing(Utils.EMPTY_BB_LIST, true);
 return;
 resumeActivity();
 }
 handshaking = true;
 } else {
 if ((handshakeState.getAndSet(NOT_HANDSHAKING)& ~DOING_TASKS) == HANDSHAKING) {
+handshaking = false;
+applicationBufferSize = engine.getSession().getApplicationBufferSize();
+packetBufferSize = engine.getSession().getPacketBufferSize();
 setALPN();
-handshaking = false;
 resumeActivity();
 }
 }
 } catch (IOException ex) {
 errorCommon(ex);
 SSLEngineResult sslResult = engine.unwrap(src, dst);
 switch (sslResult.getStatus()) {
 case BUFFER_OVERFLOW:
 // may happen only if app size buffer was changed.
 // get it again if app buffer size changed
-int appSize = engine.getSession().getApplicationBufferSize();
+int appSize = applicationBufferSize =
+engine.getSession().getApplicationBufferSize();
 ByteBuffer b = ByteBuffer.allocate(appSize + dst.position());
 dst.flip();
 b.put(dst);
 dst = b;
 break;
 scheduler = new SequentialScheduler(new WriterDownstreamPusher());
 }
 @Override
 protected void incoming(List<ByteBuffer> buffers, boolean complete) {
-assert complete ? buffers ==  Utils.EMPTY_BB_LIST : true;
+assert complete ? buffers == Utils.EMPTY_BB_LIST : true;
 assert buffers != Utils.EMPTY_BB_LIST ? complete == false : true;
 if (complete) {
 if (debugw.on()) debugw.log("adding SENTINEL");
 completing = true;
 writeList.add(SENTINEL);
 return true;
 return false;
 }
 }
+void triggerWrite() {
+synchronized (writeList) {
+if (writeList.isEmpty()) {
+writeList.add(HS_TRIGGER);
+}
+}
+scheduler.runOrSchedule();
+}
 private void processData() {
 boolean completing = isCompleting();
 try {
 if (debugw.on())
 if (debugw.on()) debugw.log("handshaking");
 doHandshake(result, WRITER);  // ok to ignore return
 handshaking = true;
 } else {
 if ((handshakeState.getAndSet(NOT_HANDSHAKING) & ~DOING_TASKS) == HANDSHAKING) {
+applicationBufferSize = engine.getSession().getApplicationBufferSize();
+packetBufferSize = engine.getSession().getPacketBufferSize();
 setALPN();
 resumeActivity();
 }
 }
 cleanList(writeList); // tidy up the source list
 switch (sslResult.getStatus()) {
 case BUFFER_OVERFLOW:
 // Shouldn't happen. We allocated buffer with packet size
 // get it again if net buffer size was changed
 if (debugw.on()) debugw.log("BUFFER_OVERFLOW");
-int appSize = engine.getSession().getApplicationBufferSize();
+int netSize = packetBufferSize
-ByteBuffer b = ByteBuffer.allocate(appSize + dst.position());
+= engine.getSession().getPacketBufferSize();
+ByteBuffer b = ByteBuffer.allocate(netSize + dst.position());
 dst.flip();
 b.put(dst);
 dst = b;
 break; // try again
 case CLOSED:
 reader.schedule();
 writer.schedule();
 }
 final AtomicInteger handshakeState;
-final ConcurrentLinkedQueue<String> stateList = new ConcurrentLinkedQueue<>();
+final ConcurrentLinkedQueue<String> stateList =
+debug.on() ? new ConcurrentLinkedQueue<>() : null;
 private boolean doHandshake(EngineResult r, int caller) {
 // unconditionally sets the HANDSHAKING bit, while preserving DOING_TASKS
 handshakeState.getAndAccumulate(HANDSHAKING, (current, update) -> update | (current & DOING_TASKS));
-stateList.add(r.handshakeStatus().toString());
+if (stateList != null && debug.on()) {
-stateList.add(Integer.toString(caller));
+stateList.add(r.handshakeStatus().toString());
+stateList.add(Integer.toString(caller));
+}
 switch (r.handshakeStatus()) {
 case NEED_TASK:
 int s = handshakeState.getAndUpdate((current) -> current | DOING_TASKS);
 if ((s & DOING_TASKS) > 0) // someone else was doing tasks
 return false;
 List<Runnable> tasks = obtainTasks();
 executeTasks(tasks);
 return false;  // executeTasks will resume activity
 case NEED_WRAP:
 if (caller == READER) {
-writer.addData(HS_TRIGGER);
+writer.triggerWrite();
 return false;
 }
 break;
 case NEED_UNWRAP:
 case NEED_UNWRAP_AGAIN:
 } else {
 break;
 }
 } while (true);
 handshakeState.getAndUpdate((current) -> current & ~DOING_TASKS);
-//writer.addData(HS_TRIGGER);
 resumeActivity();
 } catch (Throwable t) {
 handleError(t);
 }
 });
 // an acknowledgement back. We're calling doHandshake
 // to finish the close handshake.
 if (engine.isInboundDone() && !engine.isOutboundDone()) {
 if (debug.on()) debug.log("doClosure: close_notify received");
 close_notify_received = true;
-doHandshake(r, READER);
+if (!writer.scheduler.isStopped()) {
+doHandshake(r, READER);
+} else {
+// We have received closed notify, but we
+// won't be able to send the acknowledgement.
+// Nothing more will come from the socket either,
+// so mark the reader as completed.
+synchronized (reader.readBufferLock) {
+reader.completing = true;
+}
+}
 }
 }
 return r;
 }
 SSLEngineResult.Status status() {
 return result.getStatus();
 }
 }
-public ByteBuffer getNetBuffer() {
+volatile int packetBufferSize;
-return ByteBuffer.allocate(engine.getSession().getPacketBufferSize());
+final ByteBuffer getNetBuffer() {
-}
+int netSize = packetBufferSize;
+if (netSize <= 0) {
-private ByteBuffer getAppBuffer() {
+packetBufferSize = netSize = engine.getSession().getPacketBufferSize();
-return ByteBuffer.allocate(engine.getSession().getApplicationBufferSize());
+}
+return ByteBuffer.allocate(netSize);
+}
+volatile int applicationBufferSize;
+final ByteBuffer getAppBuffer() {
+int appSize = applicationBufferSize;
+if (appSize <= 0) {
+applicationBufferSize = appSize = engine.getSession().getApplicationBufferSize();
+}
+return ByteBuffer.allocate(appSize);
 }
 final String dbgString() {
 return "SSLFlowDelegate(" + tubeName + ")";
 }

changeset 49944	4690a2871b44
parent 49765	ee6f7a61f3a5
child 50681	4254bed3c09d
child 56507	2294c51eae30