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

equal deleted inserted replaced

-:f1f4b8cd0192
+:cd41f34e548c
 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.ref.Reference;
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
 private static final ByteBuffer SENTINEL = Utils.EMPTY_BYTEBUFFER;
 private static final ByteBuffer HS_TRIGGER = ByteBuffer.allocate(0);
 // When handshake is in progress trying to wrap may produce no bytes.
 private static final ByteBuffer NOTHING = ByteBuffer.allocate(0);
 private static final String monProp = Utils.getProperty("jdk.internal.httpclient.monitorFlowDelegate");
+private static final boolean isMonitored =
+monProp != null && (monProp.equals("") || monProp.equalsIgnoreCase("true"));
 final Executor exec;
 final Reader reader;
 final Writer writer;
 final SSLEngine engine;
 final String tubeName; // hack
 final CompletableFuture<String> alpnCF; // completes on initial handshake
+final Monitorable monitor = isMonitored ? this::monitor : null; // prevent GC until SSLFD is stopped
 volatile boolean close_notify_received;
 final CompletableFuture<Void> readerCF;
 final CompletableFuture<Void> writerCF;
 final Consumer<ByteBuffer> recycler;
 static AtomicInteger scount = new AtomicInteger(1);
 // connect the Reader to the downReader and the
 // Writer to the downWriter.
 connect(downReader, downWriter);
-if (monProp != null && (monProp.equals("") || monProp.equalsIgnoreCase("true")))
+if (isMonitored) Monitor.add(monitor);
-Monitor.add(this::monitor);
 }
 /**
 * Returns true if the SSLFlowDelegate has detected a TLS
 * close_notify from the server.
 }
 public String monitor() {
 StringBuilder sb = new StringBuilder();
 sb.append("SSL: id ").append(id);
+sb.append(" ").append(dbgString());
 sb.append(" HS state: " + states(handshakeState));
 sb.append(" Engine state: " + engine.getHandshakeStatus().toString());
 if (stateList != null) {
 sb.append(" LL : ");
 for (String s : stateList) {
 scheduler.runOrSchedule(exec);
 }
 @Override
 public String toString() {
-return "READER: " + super.toString() + " readBuf: " + readBuf.toString()
+return "READER: " + super.toString() + ", readBuf: " + readBuf.toString()
-+ " count: " + count.toString();
++ ", count: " + count.toString() + ", scheduler: "
++ (scheduler.isStopped() ? "stopped" : "running")
++ ", status: " + lastUnwrapStatus;
 }
 private void reallocReadBuf() {
 int sz = readBuf.capacity();
 ByteBuffer newb = ByteBuffer.allocate(sz * 2);
 // WARNING: do not touch buf after this point!
 if (recycler != null) recycler.accept(buf);
 }
 if (complete) {
 this.completing = complete;
+minBytesRequired = 0;
 }
 }
 }
 void schedule() {
 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) {
+minBytesRequired = len;
 // 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;
+else if (this.completing) {
+if (debug.on()) {
+debugr.log("BUFFER_UNDERFLOW with EOF," +
+" %d bytes non decrypted.", len);
+}
+// The channel won't send us any more data, and
+// we are in underflow: we need to fail.
+throw new IOException("BUFFER_UNDERFLOW with EOF, "
++ len + " bytes non decrypted.");
+}
 }
 // request more data and return.
 requestMore();
 return;
 }
 }
 }
 } catch (IOException ex) {
 errorCommon(ex);
 handleError(ex);
+return;
 }
 if (handshaking && !complete)
 return;
 }
 if (!complete) {
 errorCommon(ex);
 handleError(ex);
 }
 }
+private volatile Status lastUnwrapStatus;
 EngineResult unwrapBuffer(ByteBuffer src) throws IOException {
 ByteBuffer dst = getAppBuffer();
 int len = src.remaining();
 while (true) {
 SSLEngineResult sslResult = engine.unwrap(src, dst);
-switch (sslResult.getStatus()) {
+switch (lastUnwrapStatus = sslResult.getStatus()) {
 case BUFFER_OVERFLOW:
 // may happen if app size buffer was changed, or if
 // our 'adaptiveBufferSize' guess was too small for
 // the current payload. In that case, update the
 // value of applicationBufferSize, and allocate a
 public interface Monitorable {
 public String getInfo();
 }
 public static class Monitor extends Thread {
-final List<Monitorable> list;
+final List<WeakReference<Monitorable>> list;
+final List<FinalMonitorable> finalList;
+final ReferenceQueue<Monitorable> queue = new ReferenceQueue<>();
 static Monitor themon;
 static {
 themon = new Monitor();
 themon.start(); // uncomment to enable Monitor
+}
+// An instance used to temporarily store the
+// last observable state of a monitorable object.
+// When Monitor.remove(o) is called, we replace
+// 'o' with a FinalMonitorable whose reference
+// will be enqueued after the last observable state
+// has been printed.
+final class FinalMonitorable implements Monitorable {
+final String finalState;
+FinalMonitorable(Monitorable o) {
+finalState = o.getInfo();
+finalList.add(this);
+}
+@Override
+public String getInfo() {
+finalList.remove(this);
+return finalState;
+}
 }
 Monitor() {
 super("Monitor");
 setDaemon(true);
 list = Collections.synchronizedList(new LinkedList<>());
+finalList = new ArrayList<>(); // access is synchronized on list above
 }
 void addTarget(Monitorable o) {
-list.add(o);
+list.add(new WeakReference<>(o, queue));
+}
+void removeTarget(Monitorable o) {
+// It can take a long time for GC to clean up references.
+// Calling Monitor.remove() early helps removing noise from the
+// logs/
+synchronized (list) {
+Iterator<WeakReference<Monitorable>> it = list.iterator();
+while (it.hasNext()) {
+Monitorable m = it.next().get();
+if (m == null) it.remove();
+if (o == m) {
+it.remove();
+break;
+}
+}
+FinalMonitorable m = new FinalMonitorable(o);
+addTarget(m);
+Reference.reachabilityFence(m);
+}
 }
 public static void add(Monitorable o) {
 themon.addTarget(o);
+}
+public static void remove(Monitorable o) {
+themon.removeTarget(o);
 }
 @Override
 public void run() {
 System.out.println("Monitor starting");
 try {
 while (true) {
 Thread.sleep(20 * 1000);
 synchronized (list) {
-for (Monitorable o : list) {
+Reference<? extends Monitorable> expired;
+while ((expired = queue.poll()) != null) list.remove(expired);
+for (WeakReference<Monitorable> ref : list) {
+Monitorable o = ref.get();
+if (o == null) continue;
+if (o instanceof FinalMonitorable) {
+ref.enqueue();
+}
 System.out.println(o.getInfo());
 System.out.println("-------------------------");
 }
 }
 System.out.println("--o-o-o-o-o-o-o-o-o-o-o-o-o-o-");
 // If the SSLEngine produces less than writeBuffer.capacity() / 2,
 // then we copy off the bytes to a smaller buffer that we send
 // downstream. Otherwise, we send the writeBuffer downstream
 // and will allocate a new one next time.
 volatile ByteBuffer writeBuffer;
+private volatile Status lastWrappedStatus;
 @SuppressWarnings("fallthrough")
 EngineResult wrapBuffers(ByteBuffer[] src) throws SSLException {
 long len = Utils.remaining(src);
 if (debugw.on())
 debugw.log("wrapping " + len + " bytes");
 assert dst.hasRemaining() : "buffer has no remaining space: capacity=" + dst.capacity();
 while (true) {
 SSLEngineResult sslResult = engine.wrap(src, dst);
 if (debugw.on()) debugw.log("SSLResult: " + sslResult);
-switch (sslResult.getStatus()) {
+switch (lastWrappedStatus = 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 netSize = packetBufferSize
 }
 }
 @Override
 public String toString() {
-return "WRITER: " + super.toString() +
+return "WRITER: " + super.toString()
-" writeList size " + Integer.toString(writeList.size());
++ ", writeList size: " + Integer.toString(writeList.size())
++ ", scheduler: " + (scheduler.isStopped() ? "stopped" : "running")
++ ", status: " + lastWrappedStatus;
 //" writeList: " + writeList.toString();
 }
 }
 private void handleError(Throwable t) {
 if (stopped)
 return;
 stopped = true;
 reader.stop();
 writer.stop();
+if (isMonitored) Monitor.remove(monitor);
 }
 private Void stopOnError(Throwable currentlyUnused) {
 // maybe log, etc
 normalStop();
 break;
 case NEED_UNWRAP:
 case NEED_UNWRAP_AGAIN:
 // do nothing else
 // receiving-side data will trigger unwrap
+if (caller == WRITER) {
+reader.schedule();
+return false;
+}
 break;
 default:
 throw new InternalError("Unexpected handshake status:"
 + r.handshakeStatus());
 }

changeset 50985	cd41f34e548c
parent 50681	4254bed3c09d
child 52902	e3398b2e1ab0
child 56833	be0819373531